pmcp/mcp_server/github_server/tools/pull_requests.py

"""
Pull-request utilities (read-only).
"""
from typing import List, Dict
from mcp.server.fastmcp import Context
from pmcp.mcp_server.github_server.services.pull_requests import PullRequestService
from pmcp.mcp_server.github_server.github import github_client

service = PullRequestService(github_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