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

# Auto-execute using the project's local virtual environment Python
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
if os.path.exists(venv_python) and sys.executable != venv_python:
    os.execv(venv_python, [venv_python] + sys.argv)

# Ensure the project root is on the path so gitea_auth.py can be imported.
if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)

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

from gitea_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_list_prs(
    state: str = "open",
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> list[dict]:
    """List pull requests on a Gitea repository.

    Args:
        state: State filter — 'open', 'closed', or 'all' (default: 'open').
        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', 'head', 'base', 'url', 'mergeable'.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
    prs = api_request("GET", url, auth) or []
    return [
        {
            "number": pr["number"],
            "title": pr["title"],
            "state": pr["state"],
            "head": pr["head"]["ref"],
            "base": pr["base"]["ref"],
            "url": pr["html_url"],
            "mergeable": pr.get("mergeable"),
        }
        for pr in prs
    ]


@mcp.tool()
def gitea_view_pr(
    pr_number: int,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Get details of a single pull request.

    Args:
        pr_number: The pull request index/number.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict with PR details.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
    pr = api_request("GET", url, auth)
    return {
        "number": pr["number"],
        "title": pr["title"],
        "body": pr.get("body", ""),
        "state": pr["state"],
        "head": pr["head"]["ref"],
        "base": pr["base"]["ref"],
        "url": pr["html_url"],
        "mergeable": pr.get("mergeable"),
        "user": pr.get("user", {}).get("login", ""),
    }


@mcp.tool()
def gitea_merge_pr(
    pr_number: int,
    do: str = "merge",
    title: str | None = None,
    message: str | None = None,
    force: bool = False,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Merge a Gitea pull request.

    Args:
        pr_number: The PR number to merge.
        do: Merge style — 'merge', 'squash', or 'rebase'.
        title: Optional merge title.
        message: Optional merge message.
        force: Force merge, ignoring status checks.
        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' and 'message'.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/merge"
    payload = {
        "Do": do,
        "force_merge": force,
    }
    if title:
        payload["MergeTitleField"] = title
    if message:
        payload["MergeMessageField"] = message
    api_request("POST", url, auth, payload)
    return {"success": True, "message": f"PR #{pr_number} merged via '{do}'."}


@mcp.tool()
def gitea_delete_branch(
    branch: str,
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Delete a remote branch from a Gitea repository.

    Args:
        branch: The remote branch name (e.g. 'feat/branch-name').
        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' and 'message'.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    import urllib.parse
    encoded_branch = urllib.parse.quote(branch, safe="")
    url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}"
    api_request("DELETE", url, auth)
    return {"success": True, "message": f"Remote branch '{branch}' deleted."}


@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_list_labels(
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> list:
    """List all available labels in a Gitea repository.

    Args:
        remote: Known Gitea instance ('dadeschools' or 'prgs').
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        list of labels.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)
    return api_request("GET", f"{base}/labels?limit=100", auth)


@mcp.tool()
def gitea_create_label(
    name: str,
    color: str,
    description: str = "",
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> dict:
    """Create a new label on a Gitea repository.

    Args:
        name: Name of the label (e.g. 'bug', 'epic').
        color: HTML color code (hex, e.g. 'fbca04' or '#fbca04').
        description: Description of the label.
        remote: Known Gitea instance ('dadeschools' or 'prgs').
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        dict containing the created label details.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)

    if color.startswith("#"):
        color = color[1:]

    payload = {
        "name": name,
        "color": color,
        "description": description,
    }
    return api_request("POST", f"{base}/labels", auth, payload)


@mcp.tool()
def gitea_set_issue_labels(
    issue_number: int,
    labels: list[str],
    remote: str = "dadeschools",
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
) -> list:
    """Replace all labels on a Gitea issue with a new list of label names.

    Args:
        issue_number: The issue number.
        labels: List of label names to apply.
        remote: Known Gitea instance ('dadeschools' or 'prgs').
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.

    Returns:
        list of all labels currently applied to the issue.
    """
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)

    # 1. Fetch existing labels on the repo to resolve names -> IDs
    existing = api_request("GET", f"{base}/labels?limit=100", auth)
    name_to_id = {lb["name"]: lb["id"] for lb in existing}

    # 2. Check if any requested labels do not exist, and raise error
    label_ids = []
    missing_labels = []
    for name in labels:
        if name in name_to_id:
            label_ids.append(name_to_id[name])
        else:
            missing_labels.append(name)

    if missing_labels:
        raise RuntimeError(
            f"The following labels do not exist on the repository: {missing_labels}. "
            "Please create them first using gitea_create_label."
        )

    # 3. PUT the labels to the issue
    res = api_request("PUT", f"{base}/issues/{issue_number}/labels", auth, {"labels": label_ids})
    return res


@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")