Merge branch 'feat/mcp_servers' into feat/matt

app.py

@@ -12,9 +12,17 @@ from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
 from langgraph.checkpoint.memory import MemorySaver
 
 SYSTEM_PROMPT = """
-You are an assistant that can manage Trello boards and projects.
-You will be given a set of tools to work with. Each time you decide to use a tool that modifies in any way a Trello board, you MUST ask the user if wants to proceed.
-If the user's answer is negative, then you have to abort everything and end the conversation.
+You are Trello-Assistant: you can read and write Trello boards via provided tools.
+
+SAFETY RULE  
+Before using any tool that changes data, briefly state the action and ask:
+“Proceed? (yes / no)”
+
+• Only run the tool after an explicit “yes/confirm”.  
+• If the reply is “no” or anything else, cancel the action and end the chat with “Understood—no changes made.”
+
+Think through tool choices silently; never reveal those thoughts.
+
 """
 
 

pmcp/mcp_server/github/github.py

@@ -0,0 +1,40 @@
+import os
+
+from dotenv import load_dotenv
+
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+# Load environment variables
+load_dotenv()
+
+
+# Initialize Github client and service
+try:
+    api_key = os.getenv("GITHUB_API_KEY")
+
+    if not api_key:
+        raise ValueError(
+            "GITHUB_API_KEY must be set in environment variables"
+        )
+    client = GithubClient(api_key=api_key)
+except Exception as e:
+    raise
+
+
+# Add a prompt for common Github operations
+def github_help() -> str:
+    """Provides help information about available Github operations."""
+    return """
+    Available Github Operations:
+    1. get issues
+    2. create issue
+    3. comment issue
+    4. close issue
+    5. get pull requests
+    6. create pull request
+    7. get repo stats
+    8. list branches
+    9. get recent commits
+    10. get file contents
+    """

pmcp/mcp_server/github/mcp_github_main.py

@@ -0,0 +1,35 @@
+import logging
+
+from dotenv import load_dotenv
+from mcp.server.fastmcp import FastMCP
+
+from pmcp.mcp_server.github.tools.tools import register_tools
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+# Load environment variables
+load_dotenv()
+
+
+# Initialize MCP server
+mcp = FastMCP("Github MCP Server")
+
+# Register tools
+register_tools(mcp)
+
+
+
+if __name__ == "__main__":
+    try:
+        logger.info("Starting Github MCP Server in Stdio...")
+        mcp.run()
+        logger.info("Github MCP Server started successfully")
+    except KeyboardInterrupt:
+        logger.info("Shutting down server...")
+    except Exception as e:
+        logger.error(f"Server error: {str(e)}")
+        raise

pmcp/mcp_server/github/models.py

@@ -0,0 +1,20 @@
+from pydantic import BaseModel
+from typing import List
+
+class IssuesList(BaseModel):
+    issues: List[str]
+
+class PullRequestList(BaseModel):
+    pull_requests: List[str]
+
+class BranchList(BaseModel):
+    branches: List[str]
+
+class CommitsList(BaseModel):
+    commits: List[str]
+
+class RepoStats(BaseModel):
+    stars: int
+    forks: int
+    watchers: int
+    open_issues: int

pmcp/mcp_server/github/services/branches.py

@@ -0,0 +1,21 @@
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+class BranchService:
+    """Branch enumeration service."""
+
+    def __init__(self, client: GithubClient):
+        self.client = client
+
+    async def list_branches(self, owner: str, repo: str):
+        """
+        Return branch objects for a repository.
+
+        Args:
+            owner (str): Repository owner.
+            repo (str):  Repository name.
+
+        Returns:
+            List of branch dicts (each containing ``name`` and ``commit`` keys).
+        """
+        return await self.client.GET(f"{owner}/{repo}/branches")

pmcp/mcp_server/github/services/contents.py

@@ -0,0 +1,44 @@
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+class ContentService:
+    """Commits and file-content helpers."""
+
+    def __init__(self, client: GithubClient):
+        self.client = client
+
+    async def recent_commits(
+        self, owner: str, repo: str, branch: str, per_page: int = 10
+    ):
+        """
+        Retrieve the most recent commits on a branch.
+
+        Args:
+            owner (str):    Repository owner.
+            repo (str):     Repository name.
+            branch (str):   Branch ref (e.g. ``main``).
+            per_page (int): Max commits to return (<=100).
+
+        Returns:
+            List of commit dicts.
+        """
+        params = {"sha": branch, "per_page": per_page}
+        return await self.client.GET(f"{owner}/{repo}/commits", params=params)
+
+    async def get_file(
+        self, owner: str, repo: str, path: str, ref: str | None = None
+    ):
+        """
+        Download a file’s blob (Base64) from a repo.
+
+        Args:
+            owner (str): Repository owner.
+            repo (str):  Repository name.
+            path (str):  File path within the repo.
+            ref (str):   Optional commit SHA / branch / tag.
+
+        Returns:
+            GitHub ``contents`` API response including ``content``.
+        """
+        params = {"ref": ref} if ref else None
+        return await self.client.GET(f"{owner}/{repo}/contents/{path}", params=params)

pmcp/mcp_server/github/services/issues.py

@@ -0,0 +1,86 @@
+"""
+Service for managing GitHub issues in the MCP server.
+"""
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+
+class IssueService:
+    """Business-logic layer for issue operations."""
+
+    def __init__(self, client: GithubClient):
+        self.client = client
+
+    # ────────────────────────────────────────────────────────────────── #
+    # READ
+    async def get_issues(self, owner: str, repo: str) -> any:
+        """
+        Return every open-issue JSON object for ``owner/repo``.
+
+        Args:
+            owner (str): Repository owner/organisation.
+            repo (str):  Repository name.
+
+        Returns:
+            any: List of issue dicts exactly as returned by the GitHub REST API.
+        """
+        return await self.client.GET(f"{owner}/{repo}/issues")
+
+    # ────────────────────────────────────────────────────────────────── #
+    # WRITE
+    async def create_issue(
+        self, owner: str, repo: str, title: str, body: str | None = None
+    ):
+        """
+        Create a new issue.
+
+        Args:
+            owner (str): Repository owner.
+            repo (str):  Repository name.
+            title (str): Issue title.
+            body (str):  Optional Markdown body.
+
+        Returns:
+            JSON payload describing the created issue.
+        """
+        payload = {"title": title, "body": body or ""}
+        return await self.client.POST(f"{owner}/{repo}/issues", json=payload)
+
+    async def comment_issue(
+        self, owner: str, repo: str, issue_number: int, body: str
+    ):
+        """
+        Add a comment to an existing issue.
+
+        Args:
+            owner (str):         Repository owner.
+            repo (str):          Repository name.
+            issue_number (int):  Target issue number.
+            body (str):          Comment body (Markdown).
+
+        Returns:
+            JSON with the new comment metadata.
+        """
+        payload = {"body": body}
+        return await self.client.POST(
+            f"{owner}/{repo}/issues/{issue_number}/comments", json=payload
+        )
+
+    async def close_issue(self, owner: str, repo: str, issue_number: int):
+        """
+        Close an issue by setting its state to ``closed``.
+
+        Args:
+            owner (str):        Repository owner.
+            repo (str):         Repository name.
+            issue_number (int): Issue to close.
+
+        Returns:
+            JSON for the updated issue.
+        """
+        payload = {"state": "closed"}
+        return await self.client.PATCH(
+            f"{owner}/{repo}/issues/{issue_number}", json=payload
+        )
+
+

pmcp/mcp_server/github/services/pull_requests.py

@@ -0,0 +1,58 @@
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+class PullRequestService:
+    """Read-only pull-request queries."""
+
+    def __init__(self, client: GithubClient):
+        self.client = client
+
+    async def get_pr_list(self, owner: str, repo: str, state: str = "open"):
+        """
+        List pull requests for a repository.
+
+        Args:
+            owner (str): Repository owner.
+            repo (str):  Repository name.
+            state (str): ``open``, ``closed`` or ``all``.
+
+        Returns:
+            List of PR dicts.
+        """
+        params = {"state": state}
+        return await self.client.GET(f"{owner}/{repo}/pulls", params=params)
+
+
+    async def create(
+        self,
+        owner: str,
+        repo: str,
+        title: str,
+        head: str,
+        base: str,
+        body: str | None = None,
+        draft: bool = False,
+    ):
+        """
+        Create a pull request (`POST /pulls`).
+
+        Args:
+            owner: Repository owner.
+            repo:  Repository name.
+            title: PR title.
+            head:  The branch/tag you want to merge **from** (`user:branch` accepted).
+            base:  The branch you want to merge **into** (usually `main`).
+            body:  Optional Markdown description.
+            draft: Whether to open as a draft PR.
+
+        Returns:
+            JSON describing the new pull request.
+        """
+        payload = {
+            "title": title,
+            "head": head,
+            "base": base,
+            "body": body or "",
+            "draft": draft,
+        }
+        return await self.client.POST(f"{owner}/{repo}/pulls", json=payload)

pmcp/mcp_server/github/services/repo.py

@@ -0,0 +1,21 @@
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+class RepoService:
+    """General repository information."""
+
+    def __init__(self, client: GithubClient):
+        self.client = client
+
+    async def get_stats(self, owner: str, repo: str):
+        """
+        Fetch high-level repo metadata (stars, forks, etc.).
+
+        Args:
+            owner (str): Repository owner.
+            repo (str):  Repository name.
+
+        Returns:
+            JSON with the repository resource.
+        """
+        return await self.client.GET(f"{owner}/{repo}")

pmcp/mcp_server/github/services/repo_to_text.py

@@ -0,0 +1,26 @@
+"""
+Service for managing Github issues in MCP server.
+"""
+from pmcp.mcp_server.github.utils.github_api import GithubClient
+
+
+
+class RepoToTextService:
+    """
+    Service class for managing Github repo
+    """
+
+    def __init__(self, client: GithubClient):
+        self.client = client
+
+    async def get_repo_to_text(self, repo: str) -> any:
+        """Retrieves the repo and file structure as text.
+
+        Args:
+            owner (str): The owner of the repository.
+
+        Returns:
+            any: The list of issues.
+        """
+        response = await self.client.REPO_TO_TEXT(repo)
+        return response

pmcp/mcp_server/github/tools/__init__.py

@@ -0,0 +1 @@
+

pmcp/mcp_server/github/tools/branches.py

@@ -0,0 +1,31 @@
+"""
+Branch listing tool.
+"""
+from typing import List, Dict
+from mcp.server.fastmcp import Context
+from pmcp.mcp_server.github.services.branches import BranchService
+from pmcp.mcp_server.github import client
+
+service = BranchService(client)
+
+
+async def list_branches(ctx: Context, owner: str, repo: str) -> Dict[str, List[str]]:
+    """
+    Gets the list of branches.
+
+    Args:
+        ctx:   FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str):  Repository name.
+
+    Returns:
+        {"branches": ["main", "dev", …]}
+    """
+    try:
+        branches = await service.list_branches(owner, repo)
+        names = [b["name"] for b in branches]
+        return {"branches": names}
+    except Exception as exc:
+        error_msg = f"Error while getting the list of branches for repository {repo}. Error: {str(exc)}"
+        await ctx.error(str(exc))
+        raise

pmcp/mcp_server/github/tools/contents.py

@@ -0,0 +1,62 @@
+"""
+Commit and file-content tools.
+"""
+import base64
+from typing import List, Dict
+from mcp.server.fastmcp import Context
+from pmcp.mcp_server.github.services.contents import ContentService
+from pmcp.mcp_server.github import client
+
+service = ContentService(client)
+
+
+async def get_recent_commits(
+    ctx: Context, owner: str, repo: str, branch: str, per_page: int = 10
+) -> Dict[str, List[str]]:
+    """
+    Retrieves the most recent commits.
+
+    Args:
+        ctx:   FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str):  Repository name.
+        branch (str): Name of the branch
+        per_page (int): Max commits to return
+
+    Returns:
+        {"commits": ["abc123", …]}
+    """
+    try:
+        commits = await service.recent_commits(owner, repo, branch, per_page)
+        shas = [c["sha"] for c in commits]
+        return {"commits": shas}
+    except Exception as exc:
+        error_msg = f"Error while getting the commits for branch {branch} in {repo}. Error {str(exc)}"
+        await ctx.error(str(exc))
+        raise
+
+
+async def get_file_contents(
+    ctx: Context, owner: str, repo: str, path: str, ref: str | None = None
+) -> Dict[str, str]:
+    """
+    Retrieves the most recent commits.
+
+    Args:
+        ctx:   FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str):  Repository name.
+        path (str): File path within the repo
+        ref (int): Optional commit SHA / branch / tag.
+
+    Returns:
+        {"path": "...", "content": "..."}
+    """
+    try:
+        blob = await service.get_file(owner, repo, path, ref)
+        content = base64.b64decode(blob["content"]).decode()
+        return {"path": path, "content": content}
+    except Exception as exc:
+        error_msg = f"Error while getting the content of file {path} in repository {repo}. Error: {exc}"      
+        await ctx.error(str(error_msg))
+        raise

pmcp/mcp_server/github/tools/issues.py

@@ -0,0 +1,110 @@
+"""
+MCP Tools exposing GitHub issue operations.
+"""
+from typing import List, Dict, Any
+from mcp.server.fastmcp import Context
+
+from pmcp.mcp_server.github.services.issues import IssueService
+from pmcp.mcp_server.github import client
+
+
+service = IssueService(client)
+
+
+# ───────────────────────────────────────────────────────────────────────── #
+# READ
+
+async def get_issues(ctx: Context, owner: str, repo: str) -> Dict[str, List[str]]:
+    """
+    Retrieves issues from repo
+
+    Args:
+        ctx:   FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str):  Repository name.
+
+    Returns:
+        {"issues": [title, …]}
+    """
+    try:
+        issues = await service.get_issues(owner, repo)
+        titles = [i["title"] for i in issues if "pull_request" not in i]
+        return {"issues": titles}
+    except Exception as exc:
+        error_msg = f"Failed to get issues: {str(exc)}"
+        await ctx.error(str(exc))
+        raise
+
+
+# ───────────────────────────────────────────────────────────────────────── #
+# WRITE
+
+async def create_issue(
+    ctx: Context, owner: str, repo: str, title: str, body: str | None = None
+) -> Dict[str, Any]:
+    """
+    Opens a new issue.
+
+    Args:
+        ctx: FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str): Repository name.
+        title (str): Issue title.
+        body (str): Body of the issue.
+
+    Returns:
+        {"url": "...", "number": 123}
+    """
+    try:
+        result = await service.create_issue(owner, repo, title, body)
+        return {"url": result["html_url"], "number": result["number"]}
+    except Exception as exc:
+        error_msg = f"Failed to create issue {str(exc)}"
+        await ctx.error(str(exc))
+        raise
+
+
+async def comment_issue(
+    ctx: Context, owner: str, repo: str, issue_number: int, body: str
+) -> Dict[str, str]:
+    """
+    Adds a comment on an existing issue.
+
+    Args:
+        ctx: FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str): Repository name.
+        issue_number (int): Issue number.
+        body (str): Body of the issue.
+    """
+    try:
+        result = await service.comment_issue(owner, repo, issue_number, body)
+        return {"url": result["html_url"]}
+    except Exception as exc:
+        error_msg = f"Failed to add the comment {str(exc)}"
+        await ctx.error(str(exc))
+        raise
+
+
+async def close_issue(
+    ctx: Context, owner: str, repo: str, issue_number: int
+) -> Dict[str, str]:
+    """
+    Closes an issue.
+
+    Args:
+        ctx: FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str): Repository name.
+        issue_number (int): Issue number.
+
+    Returns:
+        Dict
+    """
+    try:
+        await service.close_issue(owner, repo, issue_number)
+        return {"status": "closed"}
+    except Exception as exc:
+        error_msg = f"Failed to close the issue number {issue_number} in repo {repo}"
+        await ctx.error(str(exc))
+        raise

pmcp/mcp_server/github/tools/pull_requests.py

@@ -0,0 +1,68 @@
+"""
+Pull-request utilities (read-only).
+"""
+from typing import List, Dict
+from mcp.server.fastmcp import Context
+from pmcp.mcp_server.github.services.pull_requests import PullRequestService
+from pmcp.mcp_server.github import client
+
+service = PullRequestService(client)
+
+
+async def get_pull_requests(
+    ctx: Context, owner: str, repo: str, state: str = "open"
+) -> Dict[str, List[str]]:
+    """
+    Gets the pull requests
+
+    Args:
+        ctx:   FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str):  Repository name.
+        state (str): State of the pull request
+
+    Returns:
+        {"pull_requests": ["#1 - Add feature (open)", …]}
+    """
+    try:
+        pulls = await service.get_pr_list(owner, repo, state)
+        titles = [f"#{pr['number']} - {pr['title']} ({pr['state']})" for pr in pulls]
+        return {"pull_requests": titles}
+    except Exception as exc:
+        error_msg = f"Failed to get pull requests. Error: {str(exc)}"
+        await ctx.error(str(exc))
+        raise
+
+
+async def create_pull_request(
+    ctx: Context,
+    owner: str,
+    repo: str,
+    title: str,
+    head: str,
+    base: str,
+    body: str | None = None,
+    draft: bool = False,
+) -> Dict[str, str]:
+    """
+    Create a pull request.
+
+    Args:
+        owner: Repository owner.
+        repo:  Repository name.
+        title: PR title.
+        head:  The branch/tag you want to merge **from** (`user:branch` accepted).
+        base:  The branch you want to merge **into** (usually `main`).
+        body:  Optional Markdown description.
+        draft: Whether to open as a draft PR.
+
+    Returns:
+        {"url": pull_request_url, "number": pr_number}
+    """
+    try:
+        pr = await service.create(owner, repo, title, head, base, body, draft)
+        return {"url": pr["html_url"], "number": pr["number"]}
+    except Exception as exc:
+        error_msg = f"Error creating the pull request. Error {str(exc)}"
+        await ctx.error(str(exc))
+        raise

pmcp/mcp_server/github/tools/repo.py

@@ -0,0 +1,35 @@
+"""
+Repository-level stats tool.
+"""
+from mcp.server.fastmcp import Context
+from pmcp.mcp_server.github.services.repo import RepoService
+from pmcp.mcp_server.github import client
+
+
+service = RepoService(client)
+
+
+async def get_repo_stats(ctx: Context, owner: str, repo: str):
+    """
+    Gets the statistics of the repository
+
+    Args:
+        ctx:   FastMCP request context (handles errors).
+        owner (str): Repository owner.
+        repo (str):  Repository name.
+
+    Returns:
+        {"stars": 0, "forks": 0, "watchers": 0, "open_issues": 0}
+    """
+    try:
+        data = await service.get_stats(owner, repo)
+        return {
+            "stars": data["stargazers_count"],
+            "forks": data["forks_count"],
+            "watchers": data["watchers_count"],
+            "open_issues": data["open_issues_count"],
+        }
+    except Exception as exc:
+        error_msg = f"Failed to get statistics of repository {repo}. Error: {str(exc)}"
+        await ctx.error(str(exc))
+        raise

pmcp/mcp_server/github/tools/repo_to_text.py

@@ -0,0 +1,28 @@
+"""
+This module contains tools for managing Github.
+"""
+
+from pmcp.mcp_server.github.services.repo_to_text import RepoToTextService
+from pmcp.mcp_server.github import client
+
+from mcp.server.fastmcp import Context
+
+
+service = RepoToTextService(client)
+
+async def get_repo_to_text(ctx: Context, repo: str) -> str:
+    """Retrieves the content of a repository as text.
+
+    Args:
+        repo (str): The name of the repository.
+
+    Returns:
+        str: The content of the repository as text.
+    """
+    try:
+        content = await service.get_repo_to_text(repo)
+        return content
+    except Exception as e:
+        error_msg = f"Failed to get content from repo: {str(e)}"
+        await ctx.error(error_msg)
+        raise

pmcp/mcp_server/github/tools/tools.py

@@ -0,0 +1,31 @@
+"""
+This module contains tools for managing Github Issues
+"""
+
+from pmcp.mcp_server.github.tools import issues, pull_requests, repo, repo_to_text,branches, contents
+
+
+def register_tools(mcp):
+    """Register tools with the MCP server."""
+
+    # ISSUES
+    mcp.add_tool(issues.get_issues)
+    mcp.add_tool(issues.create_issue)
+    mcp.add_tool(issues.comment_issue)
+    mcp.add_tool(issues.close_issue)
+
+    # PULL REQUESTS
+    mcp.add_tool(pull_requests.get_pull_requests)
+    mcp.add_tool(pull_requests.create_pull_request)
+
+    # REPOSITORY
+    mcp.add_tool(repo.get_repo_stats)
+
+    # BRANCHES & COMMITS
+    mcp.add_tool(branches.list_branches)
+    mcp.add_tool(contents.get_recent_commits)
+    mcp.add_tool(contents.get_file_contents)
+
+    
+
+    mcp.add_tool(repo_to_text.get_repo_to_text)

pmcp/mcp_server/github/utils/github_api.py

@@ -0,0 +1,67 @@
+import logging
+import httpx
+from pmcp.mcp_server.github.utils.repo_to_text_utils import fetch_file_contents, fetch_repo_sha, fetch_repo_tree, format_repo_contents, parse_repo_url
+
+
+GITHUB_API_BASE = "https://api.github.com/repos"
+
+
+class GithubClient:
+    """
+    Client class for interacting with the Github API over REST.
+    """
+
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+        self.base_url = GITHUB_API_BASE
+        self.client = httpx.AsyncClient(base_url=self.base_url)
+
+    async def close(self):
+        await self.client.aclose()
+
+    async def GET(self, endpoint: str, params: dict = None):
+        return await self._request("get", endpoint, params=params)
+
+    async def POST(self, endpoint: str, json: dict):
+        return await self._request("post", endpoint, json=json)
+
+    async def PATCH(self, endpoint: str, json: dict):
+        return await self._request("patch", endpoint, json=json)
+
+    async def PUT(self, endpoint: str, json: dict = None):
+        return await self._request("put", endpoint, json=json)
+
+    async def _request(self, method: str, endpoint: str, **kwargs):
+        headers = self._get_headers()
+
+        try:
+            response = await self.client.request(method, endpoint, headers=headers, **kwargs)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            raise httpx.HTTPStatusError(
+                f"Failed to {method.upper()} {endpoint}: {str(e)}",
+                request=e.request,
+                response=e.response,
+            )
+        except httpx.RequestError as e:
+            raise httpx.RequestError(f"Failed to {method.upper()} {endpoint}: {str(e)}")
+
+    def _get_headers(self):
+        return {
+            "Accept": "application/vnd.github+json",
+            "Authorization": f"Bearer {self.api_key}",
+            "X-GitHub-Api-Version": "2022-11-28",
+        }
+    
+    async def REPO_TO_TEXT(self, repo_url: str):
+        owner, repo, ref, path = parse_repo_url(repo_url)
+        sha = fetch_repo_sha(owner, repo, ref, path, self.api_key)
+        tree = fetch_repo_tree(owner, repo, sha, self.api_key)
+        blobs = [item for item in tree if item['type'] == 'blob']
+        common_exts = ('.py', '.js', '.ts', '.jsx', '.tsx', '.java', '.cpp', '.html', '.css')
+        selected_files = [item for item in blobs if item['path'].lower().endswith(common_exts)]
+        for item in selected_files:
+            item['url'] = f"https://api.github.com/repos/{owner}/{repo}/contents/{item['path']}?ref={ref}" if ref else f"https://api.github.com/repos/{owner}/{repo}/contents/{item['path']}"
+        contents = fetch_file_contents(selected_files,self. api_key)
+        return format_repo_contents(contents)

pmcp/mcp_server/github/utils/repo_to_text_utils.py

@@ -0,0 +1,83 @@
+import requests
+import re
+import os
+from typing import List, Dict, Tuple
+
+def parse_repo_url(url: str) -> Tuple[str, str, str, str]:
+    url = url.rstrip('/')
+    pattern = r'^https:\/\/github\.com\/([^\/]+)\/([^\/]+)(\/tree\/([^\/]+)(\/(.+))?)?$'
+    match = re.match(pattern, url)
+    if not match:
+        raise ValueError('Invalid GitHub repository URL format.')
+    return match[1], match[2], match[4] or '', match[6] or ''
+
+def fetch_repo_sha(owner: str, repo: str, ref: str, path: str, token: str) -> str:
+    url = f'https://api.github.com/repos/{owner}/{repo}/contents/{path}?ref={ref}' if ref else f'https://api.github.com/repos/{owner}/{repo}/contents/{path}'
+    headers = {'Accept': 'application/vnd.github.object+json'}
+    if token:
+        headers['Authorization'] = f'token {token}'
+    resp = requests.get(url, headers=headers)
+    if resp.status_code == 403 and resp.headers.get('X-RateLimit-Remaining') == '0':
+        raise Exception('GitHub API rate limit exceeded.')
+    if resp.status_code == 404:
+        raise Exception('Repository, branch, or path not found.')
+    if not resp.ok:
+        raise Exception(f'Failed to fetch SHA. Status: {resp.status_code}')
+    return resp.json()['sha']
+
+def fetch_repo_tree(owner: str, repo: str, sha: str, token: str) -> List[Dict]:
+    url = f'https://api.github.com/repos/{owner}/{repo}/git/trees/{sha}?recursive=1'
+    headers = {'Accept': 'application/vnd.github+json'}
+    if token:
+        headers['Authorization'] = f'token {token}'
+    resp = requests.get(url, headers=headers)
+    if resp.status_code == 403 and resp.headers.get('X-RateLimit-Remaining') == '0':
+        raise Exception('GitHub API rate limit exceeded.')
+    if not resp.ok:
+        raise Exception(f'Failed to fetch repo tree. Status: {resp.status_code}')
+    return resp.json()['tree']
+
+def fetch_file_contents(files: List[Dict], token: str) -> List[Dict]:
+    headers = {'Accept': 'application/vnd.github.v3.raw'}
+    if token:
+        headers['Authorization'] = f'token {token}'
+    contents = []
+    for file in files:
+        resp = requests.get(file['url'], headers=headers)
+        if not resp.ok:
+            raise Exception(f'Failed to fetch file: {file["path"]} (status {resp.status_code})')
+        contents.append({'path': file['path'], 'text': resp.text})
+    return contents
+
+def build_tree_structure(paths: List[str]) -> Dict:
+    tree = {}
+    for path in paths:
+        parts = path.split('/')
+        current = tree
+        for i, part in enumerate(parts):
+            if part not in current:
+                current[part] = {} if i < len(parts) - 1 else None
+            current = current[part] if current[part] is not None else {}
+    return tree
+
+def format_tree_index(tree: Dict, prefix: str = '') -> str:
+    output = ''
+    entries = list(tree.items())
+    for i, (name, sub_tree) in enumerate(entries):
+        is_last = i == len(entries) - 1
+        line_prefix = '└── ' if is_last else '├── '
+        child_prefix = '    ' if is_last else '│   '
+        output += f"{prefix}{line_prefix}{name}\n"
+        if sub_tree:
+            output += format_tree_index(sub_tree, prefix + child_prefix)
+    return output
+
+def format_repo_contents(contents: List[Dict]) -> str:
+    contents.sort(key=lambda x: x['path'].lower())
+    paths = [item['path'] for item in contents]
+    tree = build_tree_structure(paths)
+    index = format_tree_index(tree)
+    result = f"Directory Structure:\n\n{index}"
+    for item in contents:
+        result += f"\n\n---\nFile: {item['path']}\n---\n\n{item['text']}\n"
+    return result

pmcp/mcp_server/trello/dtos/update_card.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+
+
+class UpdateCardPayload(BaseModel):
+    """
+    Payload for updating a card.
+
+    Attributes:
+        name (str): The name of the card.
+        desc (str): The description of the card.
+        pos (str | int): The position of the card.
+        closed (bool): Whether the card is closed or not.
+        due (str): The due date of the card in ISO 8601 format.
+        idLabels (str): Comma-separated list of label IDs for the card.
+    """
+
+    name: str | None = None
+    desc: str | None = None
+    pos: str | None = None
+    closed: bool | None = None
+    due: str | None = None
+    idLabels: str | None = None

pmcp/mcp_server/trello/mcp_trello_main.py

@@ -0,0 +1,35 @@
+import logging
+
+from dotenv import load_dotenv
+from mcp.server.fastmcp import FastMCP
+
+from pmcp.mcp_server.trello.tools.tools import register_tools
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+# Load environment variables
+load_dotenv()
+
+
+# Initialize MCP server
+mcp = FastMCP("Trello MCP Server")
+
+# Register tools
+register_tools(mcp)
+
+
+
+if __name__ == "__main__":
+    try:
+        logger.info("Starting Trello MCP Server in Stdio...")
+        mcp.run()
+        logger.info("Github MCP Server started successfully")
+    except KeyboardInterrupt:
+        logger.info("Shutting down server...")
+    except Exception as e:
+        logger.error(f"Server error: {str(e)}")
+        raise

pmcp/mcp_server/trello/models.py

@@ -0,0 +1,47 @@
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class TrelloBoard(BaseModel):
+    """Model representing a Trello board."""
+
+    id: str
+    name: str
+    desc: Optional[str] = None
+    closed: bool = False
+    idOrganization: Optional[str] = None
+    url: str
+
+
+class TrelloList(BaseModel):
+    """Model representing a Trello list."""
+
+    id: str
+    name: str
+    closed: bool = False
+    idBoard: str
+    pos: float
+
+
+class TrelloLabel(BaseModel):
+    """Model representing a Trello label."""
+    
+    id: str
+    name: str
+    color: Optional[str] = None
+
+
+class TrelloCard(BaseModel):
+    """Model representing a Trello card."""
+
+    id: str
+    name: str
+    desc: Optional[str] = None
+    closed: bool = False
+    idList: str
+    idBoard: str
+    url: str
+    pos: float
+    labels: List[TrelloLabel] = []
+    due: Optional[str] = None

pmcp/mcp_server/trello/services/board.py

@@ -0,0 +1,53 @@
+"""
+Service for managing Trello boards in MCP server.
+"""
+
+from typing import List
+
+from pmcp.mcp_server.trello.models import TrelloBoard, TrelloLabel
+from pmcp.mcp_server.trello.utils.trello_api import TrelloClient
+
+
+class BoardService:
+    """
+    Service class for managing Trello boards
+    """
+
+    def __init__(self, client: TrelloClient):
+        self.client = client
+
+    async def get_board(self, board_id: str) -> TrelloBoard:
+        """Retrieves a specific board by its ID.
+
+        Args:
+            board_id (str): The ID of the board to retrieve.
+
+        Returns:
+            TrelloBoard: The board object containing board details.
+        """
+        response = await self.client.GET(f"/boards/{board_id}")
+        return TrelloBoard(**response)
+
+    async def get_boards(self, member_id: str = "me") -> List[TrelloBoard]:
+        """Retrieves all boards for a given member.
+
+        Args:
+            member_id (str): The ID of the member whose boards to retrieve. Defaults to "me" for the authenticated user.
+
+        Returns:
+            List[TrelloBoard]: A list of board objects.
+        """
+        response = await self.client.GET(f"/members/{member_id}/boards")
+        return [TrelloBoard(**board) for board in response]
+
+    async def get_board_labels(self, board_id: str) -> List[TrelloLabel]:
+        """Retrieves all labels for a specific board.
+
+        Args:
+            board_id (str): The ID of the board whose labels to retrieve.
+
+        Returns:
+            List[TrelloLabel]: A list of label objects for the board.
+        """
+        response = await self.client.GET(f"/boards/{board_id}/labels")
+        return [TrelloLabel(**label) for label in response]

pmcp/mcp_server/trello/services/card.py

@@ -0,0 +1,84 @@
+"""
+Service for managing Trello cards in MCP server.
+"""
+
+from typing import Any, Dict, List
+
+from pmcp.mcp_server.trello.models import TrelloCard
+from pmcp.mcp_server.trello.utils.trello_api import TrelloClient
+
+
+class CardService:
+    """
+    Service class for managing Trello cards.
+    """
+
+    def __init__(self, client: TrelloClient):
+        self.client = client
+
+    async def get_card(self, card_id: str) -> TrelloCard:
+        """Retrieves a specific card by its ID.
+
+        Args:
+            card_id (str): The ID of the card to retrieve.
+
+        Returns:
+            TrelloCard: The card object containing card details.
+        """
+        response = await self.client.GET(f"/cards/{card_id}")
+        return TrelloCard(**response)
+
+    async def get_cards(self, list_id: str) -> List[TrelloCard]:
+        """Retrieves all cards in a given list.
+
+        Args:
+            list_id (str): The ID of the list whose cards to retrieve.
+
+        Returns:
+            List[TrelloCard]: A list of card objects.
+        """
+        response = await self.client.GET(f"/lists/{list_id}/cards")
+        return [TrelloCard(**card) for card in response]
+
+    async def create_card(
+        self, list_id: str, name: str, desc: str | None = None
+    ) -> TrelloCard:
+        """Creates a new card in a given list.
+
+        Args:
+            list_id (str): The ID of the list to create the card in.
+            name (str): The name of the new card.
+            desc (str, optional): The description of the new card. Defaults to None.
+
+        Returns:
+            TrelloCard: The newly created card object.
+        """
+        data = {"name": name, "idList": list_id}
+        if desc:
+            data["desc"] = desc
+        response = await self.client.POST("/cards", data=data)
+        return TrelloCard(**response)
+
+    async def update_card(self, card_id: str, **kwargs) -> TrelloCard:
+        """Updates a card's attributes.
+
+        Args:
+            card_id (str): The ID of the card to update.
+            **kwargs: Keyword arguments representing the attributes to update on the card.
+
+        Returns:
+            TrelloCard: The updated card object.
+        """
+        response = await self.client.PUT(f"/cards/{card_id}", data=kwargs)
+        return TrelloCard(**response)
+
+    async def delete_card(self, card_id: str) -> Dict[str, Any]:
+        """Deletes a card.
+
+        Args:
+            card_id (str): The ID of the card to delete.
+
+        Returns:
+            Dict[str, Any]: The response from the delete operation.
+        """
+        return await self.client.DELETE(f"/cards/{card_id}")

pmcp/mcp_server/trello/services/checklist.py

@@ -0,0 +1,162 @@
+import logging
+from typing import Dict, List, Optional
+
+from pmcp.mcp_server.trello.utils.trello_api import TrelloClient
+
+
+class ChecklistService:
+    """
+    Service class for handling Trello checklist operations.
+    """
+
+    def __init__(self, client: TrelloClient):
+        self.client = client
+
+    async def get_checklist(self, checklist_id: str) -> Dict:
+        """
+        Get a specific checklist by ID.
+
+        Args:
+            checklist_id (str): The ID of the checklist to retrieve
+
+        Returns:
+            Dict: The checklist data
+        """
+        return await self.client.GET(f"/checklists/{checklist_id}")
+
+    async def get_card_checklists(self, card_id: str) -> List[Dict]:
+        """
+        Get all checklists for a specific card.
+
+        Args:
+            card_id (str): The ID of the card to get checklists for
+
+        Returns:
+            List[Dict]: List of checklists on the card
+        """
+        return await self.client.GET(f"/cards/{card_id}/checklists")
+
+    async def create_checklist(
+        self, card_id: str, name: str, pos: Optional[str] = None
+    ) -> Dict:
+        """
+        Create a new checklist on a card.
+
+        Args:
+            card_id (str): The ID of the card to create the checklist on
+            name (str): The name of the checklist
+            pos (Optional[str]): The position of the checklist (top, bottom, or a positive number)
+
+        Returns:
+            Dict: The created checklist data
+        """
+        data = {"name": name}
+        if pos:
+            data["pos"] = pos
+        return await self.client.POST(f"/checklists", data={"idCard": card_id, **data})
+
+    async def update_checklist(
+        self, checklist_id: str, name: Optional[str] = None, pos: Optional[str] = None
+    ) -> Dict:
+        """
+        Update an existing checklist.
+
+        Args:
+            checklist_id (str): The ID of the checklist to update
+            name (Optional[str]): New name for the checklist
+            pos (Optional[str]): New position for the checklist
+
+        Returns:
+            Dict: The updated checklist data
+        """
+        data = {}
+        if name:
+            data["name"] = name
+        if pos:
+            data["pos"] = pos
+        return await self.client.PUT(f"/checklists/{checklist_id}", data=data)
+
+    async def delete_checklist(self, checklist_id: str) -> Dict:
+        """
+        Delete a checklist.
+
+        Args:
+            checklist_id (str): The ID of the checklist to delete
+
+        Returns:
+            Dict: The response from the delete operation
+        """
+        return await self.client.DELETE(f"/checklists/{checklist_id}")
+
+    async def add_checkitem(
+        self,
+        checklist_id: str,
+        name: str,
+        checked: bool = False,
+        pos: Optional[str] = None,
+    ) -> Dict:
+        """
+        Add a new item to a checklist.
+
+        Args:
+            checklist_id (str): The ID of the checklist to add the item to
+            name (str): The name of the checkitem
+            checked (bool): Whether the item is checked
+            pos (Optional[str]): The position of the item
+
+        Returns:
+            Dict: The created checkitem data
+        """
+        data = {"name": name, "checked": checked}
+        if pos:
+            data["pos"] = pos
+        return await self.client.POST(
+            f"/checklists/{checklist_id}/checkItems", data=data
+        )
+
+    async def update_checkitem(
+        self,
+        checklist_id: str,
+        checkitem_id: str,
+        name: Optional[str] = None,
+        checked: Optional[bool] = None,
+        pos: Optional[str] = None,
+    ) -> Dict:
+        """
+        Update a checkitem in a checklist.
+
+        Args:
+            checklist_id (str): The ID of the checklist containing the item
+            checkitem_id (str): The ID of the checkitem to update
+            name (Optional[str]): New name for the checkitem
+            checked (Optional[bool]): New checked state
+            pos (Optional[str]): New position for the item
+
+        Returns:
+            Dict: The updated checkitem data
+        """
+        data = {}
+        if name:
+            data["name"] = name
+        if checked is not None:
+            data["checked"] = checked
+        if pos:
+            data["pos"] = pos
+        return await self.client.PUT(
+            f"/checklists/{checklist_id}/checkItems/{checkitem_id}", data=data
+        )
+
+    async def delete_checkitem(self, checklist_id: str, checkitem_id: str) -> Dict:
+        """
+        Delete a checkitem from a checklist.
+
+        Args:
+            checklist_id (str): The ID of the checklist containing the item
+            checkitem_id (str): The ID of the checkitem to delete
+
+        Returns:
+            Dict: The response from the delete operation
+        """
+        return await self.client.DELETE(
+            f"/checklists/{checklist_id}/checkItems/{checkitem_id}"
+        )

pmcp/mcp_server/trello/services/list.py

@@ -0,0 +1,82 @@
+from typing import List
+
+from pmcp.mcp_server.trello.models import TrelloList
+from pmcp.mcp_server.trello.utils.trello_api import TrelloClient
+
+
+class ListService:
+    """
+    Service class for managing Trello lists.
+    """
+
+    def __init__(self, client: TrelloClient):
+        self.client = client
+
+    # Lists
+    async def get_list(self, list_id: str) -> TrelloList:
+        """Retrieves a specific list by its ID.
+
+        Args:
+            list_id (str): The ID of the list to retrieve.
+
+        Returns:
+            TrelloList: The list object containing list details.
+        """
+        response = await self.client.GET(f"/lists/{list_id}")
+        return TrelloList(**response)
+
+    async def get_lists(self, board_id: str) -> List[TrelloList]:
+        """Retrieves all lists on a given board.
+
+        Args:
+            board_id (str): The ID of the board whose lists to retrieve.
+
+        Returns:
+            List[TrelloList]: A list of list objects.
+        """
+        response = await self.client.GET(f"/boards/{board_id}/lists")
+        return [TrelloList(**list_data) for list_data in response]
+
+    async def create_list(
+        self, board_id: str, name: str, pos: str = "bottom"
+    ) -> TrelloList:
+        """Creates a new list on a given board.
+
+        Args:
+            board_id (str): The ID of the board to create the list in.
+            name (str): The name of the new list.
+            pos (str, optional): The position of the new list. Can be "top" or "bottom". Defaults to "bottom".
+
+        Returns:
+            TrelloList: The newly created list object.
+        """
+        data = {"name": name, "idBoard": board_id, "pos": pos}
+        response = await self.client.POST("/lists", data=data)
+        return TrelloList(**response)
+
+    async def update_list(self, list_id: str, name: str) -> TrelloList:
+        """Updates the name of a list.
+
+        Args:
+            list_id (str): The ID of the list to update.
+            name (str): The new name for the list.
+
+        Returns:
+            TrelloList: The updated list object.
+        """
+        response = await self.client.PUT(f"/lists/{list_id}", data={"name": name})
+        return TrelloList(**response)
+
+    async def delete_list(self, list_id: str) -> TrelloList:
+        """Archives a list.
+
+        Args:
+            list_id (str): The ID of the list to close.
+
+        Returns:
+            TrelloList: The archived list object.
+        """
+        response = await self.client.PUT(
+            f"/lists/{list_id}/closed", data={"value": "true"}
+        )
+        return TrelloList(**response)

pmcp/mcp_server/trello/tools/__init__.py

@@ -0,0 +1 @@
+

pmcp/mcp_server/trello/tools/board.py

@@ -0,0 +1,67 @@
+"""
+This module contains tools for managing Trello boards.
+"""
+
+import logging
+from typing import List
+
+from mcp.server.fastmcp import Context
+
+from pmcp.mcp_server.trello.models import TrelloBoard, TrelloLabel
+from pmcp.mcp_server.trello.services.board import BoardService
+from pmcp.mcp_server.trello.trello import client
+
+
+
+service = BoardService(client)
+
+
+async def get_board(ctx: Context, board_id: str) -> TrelloBoard:
+    """Retrieves a specific board by its ID.
+
+    Args:
+        board_id (str): The ID of the board to retrieve.
+
+    Returns:
+        TrelloBoard: The board object containing board details.
+    """
+    try:
+        result = await service.get_board(board_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get board: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def get_boards(ctx: Context) -> List[TrelloBoard]:
+    """Retrieves all boards for the authenticated user.
+
+    Returns:
+        List[TrelloBoard]: A list of board objects.
+    """
+    try:
+        result = await service.get_boards()
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get boards: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def get_board_labels(ctx: Context, board_id: str) -> List[TrelloLabel]:
+    """Retrieves all labels for a specific board.
+
+    Args:
+        board_id (str): The ID of the board whose labels to retrieve.
+
+    Returns:
+        List[TrelloLabel]: A list of label objects for the board.
+    """
+    try:
+        result = await service.get_board_labels(board_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get board labels: {str(e)}"
+        await ctx.error(error_msg)
+        raise

pmcp/mcp_server/trello/tools/card.py

@@ -0,0 +1,115 @@
+"""
+This module contains tools for managing Trello cards.
+"""
+
+import logging
+from typing import List
+
+from mcp.server.fastmcp import Context
+
+from pmcp.mcp_server.trello.models import TrelloCard
+from pmcp.mcp_server.trello.services.card import CardService
+from pmcp.mcp_server.trello.trello import client
+from pmcp.mcp_server.trello.dtos.update_card import UpdateCardPayload
+
+
+service = CardService(client)
+
+
+async def get_card(ctx: Context, card_id: str) -> TrelloCard:
+    """Retrieves a specific card by its ID.
+
+    Args:
+        card_id (str): The ID of the card to retrieve.
+
+    Returns:
+        TrelloCard: The card object containing card details.
+    """
+    try:
+        result = await service.get_card(card_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get card: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def get_cards(ctx: Context, list_id: str) -> List[TrelloCard]:
+    """Retrieves all cards in a given list.
+
+    Args:
+        list_id (str): The ID of the list whose cards to retrieve.
+
+    Returns:
+        List[TrelloCard]: A list of card objects.
+    """
+    try:
+        result = await service.get_cards(list_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get cards: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def create_card(
+    ctx: Context, list_id: str, name: str, desc: str | None = None
+) -> TrelloCard:
+    """Creates a new card in a given list.
+
+    Args:
+        list_id (str): The ID of the list to create the card in.
+        name (str): The name of the new card.
+        desc (str, optional): The description of the new card. Defaults to None.
+
+    Returns:
+        TrelloCard: The newly created card object.
+    """
+    try:
+        result = await service.create_card(list_id, name, desc)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to create card: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def update_card(
+    ctx: Context, card_id: str, payload: UpdateCardPayload
+) -> TrelloCard:
+    """Updates a card's attributes.
+
+    Args:
+        card_id (str): The ID of the card to update.
+        **kwargs: Keyword arguments representing the attributes to update on the card.
+
+    Returns:
+        TrelloCard: The updated card object.
+    """
+    try:
+        result = await service.update_card(
+            card_id, **payload.model_dump(exclude_unset=True)
+        )
+        return result
+    except Exception as e:
+        error_msg = f"Failed to update card: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def delete_card(ctx: Context, card_id: str) -> dict:
+    """Deletes a card.
+
+    Args:
+        card_id (str): The ID of the card to delete.
+
+    Returns:
+        dict: The response from the delete operation.
+    """
+    try:
+        result = await service.delete_card(card_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to delete card: {str(e)}"
+        await ctx.error(error_msg)
+        raise

pmcp/mcp_server/trello/tools/checklist.py

@@ -0,0 +1,139 @@
+"""
+This module contains tools for managing Trello checklists.
+"""
+
+import logging
+from typing import Dict, List, Optional
+
+from pmcp.mcp_server.trello.services.checklist import ChecklistService
+from pmcp.mcp_server.trello.trello import client
+
+service = ChecklistService(client)
+
+
+async def get_checklist(checklist_id: str) -> Dict:
+    """
+    Get a specific checklist by ID.
+
+    Args:
+        checklist_id (str): The ID of the checklist to retrieve
+
+    Returns:
+        Dict: The checklist data
+    """
+    return await service.get_checklist(checklist_id)
+
+
+async def get_card_checklists(card_id: str) -> List[Dict]:
+    """
+    Get all checklists for a specific card.
+
+    Args:
+        card_id (str): The ID of the card to get checklists for
+
+    Returns:
+        List[Dict]: List of checklists on the card
+    """
+    return await service.get_card_checklists(card_id)
+
+
+async def create_checklist(card_id: str, name: str, pos: Optional[str] = None) -> Dict:
+    """
+    Create a new checklist on a card.
+
+    Args:
+        card_id (str): The ID of the card to create the checklist on
+        name (str): The name of the checklist
+        pos (Optional[str]): The position of the checklist (top, bottom, or a positive number)
+
+    Returns:
+        Dict: The created checklist data
+    """
+    return await service.create_checklist(card_id, name, pos)
+
+
+async def update_checklist(
+    checklist_id: str, name: Optional[str] = None, pos: Optional[str] = None
+) -> Dict:
+    """
+    Update an existing checklist.
+
+    Args:
+        checklist_id (str): The ID of the checklist to update
+        name (Optional[str]): New name for the checklist
+        pos (Optional[str]): New position for the checklist
+
+    Returns:
+        Dict: The updated checklist data
+    """
+    return await service.update_checklist(checklist_id, name, pos)
+
+
+async def delete_checklist(checklist_id: str) -> Dict:
+    """
+    Delete a checklist.
+
+    Args:
+        checklist_id (str): The ID of the checklist to delete
+
+    Returns:
+        Dict: The response from the delete operation
+    """
+    return await service.delete_checklist(checklist_id)
+
+
+async def add_checkitem(
+    checklist_id: str, name: str, checked: bool = False, pos: Optional[str] = None
+) -> Dict:
+    """
+    Add a new item to a checklist.
+
+    Args:
+        checklist_id (str): The ID of the checklist to add the item to
+        name (str): The name of the checkitem
+        checked (bool): Whether the item is checked
+        pos (Optional[str]): The position of the item
+
+    Returns:
+        Dict: The created checkitem data
+    """
+    return await service.add_checkitem(checklist_id, name, checked, pos)
+
+
+async def update_checkitem(
+    checklist_id: str,
+    checkitem_id: str,
+    name: Optional[str] = None,
+    checked: Optional[bool] = None,
+    pos: Optional[str] = None,
+) -> Dict:
+    """
+    Update a checkitem in a checklist.
+
+    Args:
+        checklist_id (str): The ID of the checklist containing the item
+        checkitem_id (str): The ID of the checkitem to update
+        name (Optional[str]): New name for the checkitem
+        checked (Optional[bool]): New checked state
+        pos (Optional[str]): New position for the item
+
+    Returns:
+        Dict: The updated checkitem data
+    """
+    return await service.update_checkitem(
+        checklist_id, checkitem_id, name, checked, pos
+    )
+
+
+async def delete_checkitem(checklist_id: str, checkitem_id: str) -> Dict:
+    """
+    Delete a checkitem from a checklist.
+
+    Args:
+        checklist_id (str): The ID of the checklist containing the item
+        checkitem_id (str): The ID of the checkitem to delete
+
+    Returns:
+        Dict: The response from the delete operation
+    """
+    return await service.delete_checkitem(checklist_id, checkitem_id)

pmcp/mcp_server/trello/tools/list.py

@@ -0,0 +1,112 @@
+"""
+This module contains tools for managing Trello lists.
+"""
+
+import logging
+from typing import List
+
+from mcp.server.fastmcp import Context
+
+from pmcp.mcp_server.trello.models import TrelloList
+from pmcp.mcp_server.trello.services.list import ListService
+from pmcp.mcp_server.trello.trello import client
+
+
+
+service = ListService(client)
+
+
+# List Tools
+async def get_list(ctx: Context, list_id: str) -> TrelloList:
+    """Retrieves a specific list by its ID.
+
+    Args:
+        list_id (str): The ID of the list to retrieve.
+
+    Returns:
+        TrelloList: The list object containing list details.
+    """
+    try:
+        result = await service.get_list(list_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get list: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def get_lists(ctx: Context, board_id: str) -> List[TrelloList]:
+    """Retrieves all lists on a given board.
+
+    Args:
+        board_id (str): The ID of the board whose lists to retrieve.
+
+    Returns:
+        List[TrelloList]: A list of list objects.
+    """
+    try:
+        result = await service.get_lists(board_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to get lists: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def create_list(
+    ctx: Context, board_id: str, name: str, pos: str = "bottom"
+) -> TrelloList:
+    """Creates a new list on a given board.
+
+    Args:
+        board_id (str): The ID of the board to create the list in.
+        name (str): The name of the new list.
+        pos (str, optional): The position of the new list. Can be "top" or "bottom". Defaults to "bottom".
+
+    Returns:
+        TrelloList: The newly created list object.
+    """
+    try:
+        result = await service.create_list(board_id, name, pos)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to create list: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def update_list(ctx: Context, list_id: str, name: str) -> TrelloList:
+    """Updates the name of a list.
+
+    Args:
+        list_id (str): The ID of the list to update.
+        name (str): The new name for the list.
+
+    Returns:
+        TrelloList: The updated list object.
+    """
+    try:
+        result = await service.update_list(list_id, name)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to update list: {str(e)}"
+        await ctx.error(error_msg)
+        raise
+
+
+async def delete_list(ctx: Context, list_id: str) -> TrelloList:
+    """Archives a list.
+
+    Args:
+        list_id (str): The ID of the list to close.
+
+    Returns:
+        TrelloList: The archived list object.
+    """
+    try:
+        result = await service.delete_list(list_id)
+        return result
+    except Exception as e:
+        error_msg = f"Failed to delete list: {str(e)}"
+        await ctx.error(error_msg)
+        raise

pmcp/mcp_server/trello/tools/tools.py

@@ -0,0 +1,37 @@
+"""
+This module contains tools for managing Trello boards, lists, and cards.
+"""
+
+from pmcp.mcp_server.trello.tools import board, card, checklist, list
+
+
+def register_tools(mcp):
+    """Register tools with the MCP server."""
+    # Board Tools
+    mcp.add_tool(board.get_board)
+    mcp.add_tool(board.get_boards)
+    mcp.add_tool(board.get_board_labels)
+
+    # List Tools
+    mcp.add_tool(list.get_list)
+    mcp.add_tool(list.get_lists)
+    mcp.add_tool(list.create_list)
+    mcp.add_tool(list.update_list)
+    mcp.add_tool(list.delete_list)
+
+    # Card Tools
+    mcp.add_tool(card.get_card)
+    mcp.add_tool(card.get_cards)
+    mcp.add_tool(card.create_card)
+    mcp.add_tool(card.update_card)
+    mcp.add_tool(card.delete_card)
+
+    # Checklist Tools
+    mcp.add_tool(checklist.get_checklist)
+    mcp.add_tool(checklist.get_card_checklists)
+    mcp.add_tool(checklist.create_checklist)
+    mcp.add_tool(checklist.update_checklist)
+    mcp.add_tool(checklist.delete_checklist)
+    mcp.add_tool(checklist.add_checkitem)
+    mcp.add_tool(checklist.update_checkitem)
+    mcp.add_tool(checklist.delete_checkitem)

pmcp/mcp_server/trello/trello.py

@@ -0,0 +1,56 @@
+import logging
+import os
+
+from dotenv import load_dotenv
+
+from pmcp.mcp_server.trello.utils.trello_api import TrelloClient
+
+
+# Load environment variables
+load_dotenv()
+
+
+# Initialize Trello client and service
+try:
+    api_key = os.getenv("TRELLO_API_KEY")
+    token = os.getenv("TRELLO_TOKEN")
+    if not api_key or not token:
+        raise ValueError(
+            "TRELLO_API_KEY and TRELLO_TOKEN must be set in environment variables"
+        )
+    client = TrelloClient(api_key=api_key, token=token)
+    
+except Exception as e:
+    raise
+
+
+# Add a prompt for common Trello operations
+def trello_help() -> str:
+    """Provides help information about available Trello operations."""
+    return """
+    Available Trello Operations:
+    1. Board Operations:
+       - Get a specific board
+       - List all boards
+    2. List Operations:
+       - Get a specific list
+       - List all lists in a board
+       - Create a new list
+       - Update a list's name
+       - Archive a list
+    3. Card Operations:
+       - Get a specific card
+       - List all cards in a list
+       - Create a new card
+       - Update a card's attributes
+       - Delete a card
+    4. Checklist Operations:
+       - Get a specific checklist
+       - List all checklists in a card
+       - Create a new checklist
+       - Update a checklist
+       - Delete a checklist
+       - Add checkitem to checklist
+       - Update checkitem
+       - Delete checkitem
+    """

pmcp/mcp_server/trello/utils/trello_api.py

@@ -0,0 +1,88 @@
+# trello_api.py
+import logging
+
+import httpx
+
+
+
+
+TRELLO_API_BASE = "https://api.trello.com/1"
+
+
+class TrelloClient:
+    """
+    Client class for interacting with the Trello API over REST.
+    """
+
+    def __init__(self, api_key: str, token: str):
+        self.api_key = api_key
+        self.token = token
+        self.base_url = TRELLO_API_BASE
+        self.client = httpx.AsyncClient(base_url=self.base_url)
+
+    async def close(self):
+        await self.client.aclose()
+
+    async def GET(self, endpoint: str, params: dict = None):
+        all_params = {"key": self.api_key, "token": self.token}
+        if params:
+            all_params.update(params)
+        try:
+            response = await self.client.get(endpoint, params=all_params)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            raise httpx.HTTPStatusError(
+                f"Failed to get {endpoint}: {str(e)}",
+                request=e.request,
+                response=e.response,
+            )
+        except httpx.RequestError as e:
+            raise httpx.RequestError(f"Failed to get {endpoint}: {str(e)}")
+
+    async def POST(self, endpoint: str, data: dict = None):
+        all_params = {"key": self.api_key, "token": self.token}
+        try:
+            response = await self.client.post(endpoint, params=all_params, json=data)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            raise httpx.HTTPStatusError(
+                f"Failed to post to {endpoint}: {str(e)}",
+                request=e.request,
+                response=e.response,
+            )
+        except httpx.RequestError as e:
+            raise httpx.RequestError(f"Failed to post to {endpoint}: {str(e)}")
+
+    async def PUT(self, endpoint: str, data: dict = None):
+        all_params = {"key": self.api_key, "token": self.token}
+        try:
+            response = await self.client.put(endpoint, params=all_params, json=data)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            raise httpx.HTTPStatusError(
+                f"Failed to put to {endpoint}: {str(e)}",
+                request=e.request,
+                response=e.response,
+            )
+        except httpx.RequestError as e:
+            raise httpx.RequestError(f"Failed to put to {endpoint}: {str(e)}")
+
+    async def DELETE(self, endpoint: str, params: dict = None):
+        all_params = {"key": self.api_key, "token": self.token}
+        if params:
+            all_params.update(params)
+        try:
+            response = await self.client.delete(endpoint, params=all_params)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            raise httpx.HTTPStatusError(
+                f"Failed to delete {endpoint}: {str(e)}",
+                request=e.request,
+                response=e.response,
+            )
+        except httpx.RequestError as e:
+            raise httpx.RequestError(f"Failed to delete {endpoint}: {str(e)}")