Gitea-Tools/mcp_server.py

#!/usr/bin/env python3
"""Gitea MCP Server — exposes Gitea operations as MCP tools.

Runs over stdio. All tools authenticate via macOS keychain (git credential fill).

Usage (standalone test):
    python3 mcp_server.py

Configuration (mcp_config.json):
    "gitea-tools": {
        "command": "/Users/jasonwalker/Development/Gitea-Tools/venv/bin/python3",
        "args": ["/Users/jasonwalker/Development/Gitea-Tools/mcp_server.py"],
        "env": {}
    }
"""
import os
import sys
import subprocess

# Ensure the project root is on the path so auth.py can be imported.
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)

from mcp.server.fastmcp import FastMCP  # noqa: E402

from auth import (  # noqa: E402
    REMOTES,
    get_credentials,
    get_auth_header,
    api_request,
    repo_api_url,
)

mcp = FastMCP("gitea-tools", instructions=(
    "Gitea issue tracker and PR management for dadeschools and prgs instances. "
    "Use the gitea_ prefixed tools to create issues, PRs, list issues, etc."
))


# ── Helpers ───────────────────────────────────────────────────────────────────

def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
    """Resolve remote + overrides to (host, org, repo)."""
    if remote not in REMOTES:
        raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
    profile = REMOTES[remote]
    return (
        host or profile["host"],
        org or profile["org"],
        repo or profile["repo"],
    )


def _auth(host: str) -> str:
    """Get auth header, raise if unavailable."""
    header = get_auth_header(host)
    if header is None:
        raise RuntimeError(
            f"No credentials for {host}. "
            "Ensure you've logged in via HTTPS at least once."
        )
    return header


# ── Tools ─────────────────────────────────────────────────────────────────────

@mcp.tool()
def gitea_create_issue(
    title: str,
    body: str = "",
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Create a new issue on a Gitea repository.

    Args:
        title: Issue title (required).
        body: Issue body text.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict with 'number' and 'url' of the created issue.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/issues"
    data = api_request("POST", url, auth, {"title": title, "body": body})
    return {"number": data["number"], "url": data["html_url"]}


@mcp.tool()
def gitea_create_pr(
    title: str,
    head: str,
    base: str = "main",
    body: str = "",
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Create a pull request on a Gitea repository.

    Args:
        title: PR title (required).
        head: Source branch name (required).
        base: Target branch (default: 'main').
        body: PR description.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict with 'number' and 'url' of the created PR.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls"
    payload = {"title": title, "body": body, "head": head, "base": base}
    data = api_request("POST", url, auth, payload)
    return {"number": data["number"], "url": data["html_url"]}


@mcp.tool()
def gitea_close_issue(
    issue_number: int,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Close an issue by setting its state to 'closed'.

    Args:
        issue_number: The issue number to close.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict with 'success' boolean and 'message'.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
    api_request("PATCH", url, auth, {"state": "closed"})
    return {"success": True, "message": f"Issue #{issue_number} closed."}


@mcp.tool()
def gitea_list_issues(
    state: str = "open",
    label: str | None = None,
    limit: int = 50,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> list[dict]:
    """List issues on a Gitea repository with optional filters.

    Args:
        state: Filter by state — 'open', 'closed', or 'all'.
        label: Filter by label name (e.g. 'important').
        limit: Max number of issues to return (default: 50).
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        List of dicts with 'number', 'title', 'state', 'labels', 'assignee'.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    params = f"state={state}&limit={limit}&type=issues"
    if label:
        params += f"&labels={label}"
    url = f"{repo_api_url(h, o, r)}/issues?{params}"
    issues = api_request("GET", url, auth)
    return [
        {
            "number": i["number"],
            "title": i["title"],
            "state": i["state"],
            "labels": [lb["name"] for lb in i.get("labels", [])],
            "assignee": (i.get("assignee") or {}).get("login", ""),
        }
        for i in issues
    ]


@mcp.tool()
def gitea_view_issue(
    issue_number: int,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Get full details of a single issue.

    Args:
        issue_number: The issue number to view.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict with 'number', 'title', 'body', 'state', 'labels', 'assignee', 'url'.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
    i = api_request("GET", url, auth)
    return {
        "number": i["number"],
        "title": i["title"],
        "body": i.get("body", ""),
        "state": i["state"],
        "labels": [lb["name"] for lb in i.get("labels", [])],
        "assignee": (i.get("assignee") or {}).get("login", ""),
        "url": i["html_url"],
    }


@mcp.tool()
def gitea_mark_issue(
    issue_number: int,
    action: str,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Claim or release an issue via the status:in-progress label.

    This is the cross-agent lock mechanism. Check before starting work.

    Args:
        issue_number: The issue number.
        action: 'start' to claim (add label) or 'done' to release (remove label).
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict with 'success' boolean and 'message'.
    """
    if action not in ("start", "done"):
        raise ValueError(f"action must be 'start' or 'done', got '{action}'")

    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)

    # Find the status:in-progress label id
    labels = api_request("GET", f"{base}/labels?limit=100", auth)
    label_id = None
    for lb in labels:
        if lb["name"] == "status:in-progress":
            label_id = lb["id"]
            break

    if label_id is None:
        raise RuntimeError(
            "Label 'status:in-progress' not found. "
            "Run manage_labels.py to create it first."
        )

    if action == "start":
        api_request("POST", f"{base}/issues/{issue_number}/labels", auth,
                     {"labels": [label_id]})
        return {"success": True, "message": f"Issue #{issue_number} claimed."}
    else:
        api_request("DELETE",
                     f"{base}/issues/{issue_number}/labels/{label_id}", auth)
        return {"success": True, "message": f"Issue #{issue_number} released."}


@mcp.tool()
def gitea_mirror_refs(
    apply: bool = False,
    force: bool = False,
) -> dict:
    """Mirror branches and tags between dadeschools and prgs Timesheet repos.

    Additive only — never deletes branches or tags. Diverged branches are
    skipped unless force is True.

    Args:
        apply: If True, actually push. If False (default), dry-run only.
        force: If True, force-push diverged branches.

    Returns:
        dict with 'output' (script stdout) and 'return_code'.
    """
    script = os.path.join(PROJECT_ROOT, "mirror_refs.sh")
    args = [script]
    if apply:
        args.append("--apply")
    if force:
        args.append("--force")

    result = subprocess.run(
        args, capture_output=True, text=True, timeout=120,
    )
    return {
        "output": result.stdout + result.stderr,
        "return_code": result.returncode,
    }


# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    mcp.run(transport="stdio")