pmcp/mcp_server/github_server/tools/contents.py

"""
Commit and file-content tools.
"""
import base64
from typing import Any, List, Dict
from mcp.server.fastmcp import Context
from pmcp.mcp_server.github_server.services.contents import ContentService
from pmcp.mcp_server.github_server.github import github_client

service = ContentService(github_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)
        # return a list of { sha, message }
        items = [
            {"sha": c["sha"], "message": c["commit"]["message"]}
            for c in commits
        ]
        return {"commits": items}
        
    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


async def get_commit_details(
    ctx: Context, owner: str, repo: str, sha: str
) -> Dict[str, Any]:
    """
    Retrieves a commit by SHA, including its commit message and the diff.

    Args:
        ctx:    FastMCP request context.
        owner:  Repository owner.
        repo:   Repository name.
        sha:    Commit SHA.

    Returns:
        {
          "sha": "...",
          "message": "...",
          "files": [
             { "filename": "...", "patch": "diff text..." },
             … 
          ]
        }
    """
    try:
        commit = await service.get_commit_details(owner, repo, sha)
        message = commit["commit"]["message"]
        files = [
            {"filename": f["filename"], "patch": f.get("patch", "")}
            for f in commit.get("files", [])
        ]
        return {"sha": sha, "message": message, "files": files}
    except Exception as exc:
        await ctx.error(str(exc))
        raise