teahub/mcp/tea_mcp/server.py

"""Green Tea MCP: Teahub's MCP against the Gitea at GITEA_URL.

Every tool acts with the caller's own Gitea authority. Sign-in is Gitea's OAuth (and Gitea's sign-in is GitHub or
nostr), so a login here is a login there: no service token, no allowlist, nothing that outruns the caller's own
permissions. Reads and writes are bounded; nothing here deletes.
"""
from __future__ import annotations

import base64
import binascii
import os
import re
from typing import Any

import httpx
import uvicorn
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from fastmcp.server.auth import cimd as fastmcp_cimd
from fastmcp.server.auth.oauth_proxy import OAuthProxy
from fastmcp.server.auth.oauth_proxy.models import ProxyDCRClient
from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
from fastmcp.server.dependencies import get_access_token
from mcp.server.auth.provider import AccessToken
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route


def required_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


GITEA_URL = required_env("GITEA_URL").rstrip("/")
PUBLIC_URL = required_env("PUBLIC_URL").rstrip("/")
GITEA_OAUTH_CLIENT_ID = required_env("GITEA_OAUTH_CLIENT_ID")
GITEA_OAUTH_CLIENT_SECRET = required_env("GITEA_OAUTH_CLIENT_SECRET")
JWT_SIGNING_KEY = required_env("JWT_SIGNING_KEY")
ALLOWED_HOSTS = [h.strip() for h in os.getenv("ALLOWED_HOSTS", "mcp.teahub.xyz,localhost").split(",") if h.strip()]

OAUTH_SCOPES = ["openid", "profile", "email"]
CLAUDE_CIMD_CLIENT_ID = "https://claude.ai/oauth/mcp-oauth-client-metadata"
CLAUDE_REDIRECT_URI = "https://claude.ai/api/mcp/auth_callback"
MAX_BYTES = 200_000
NAME = re.compile(r"[A-Za-z0-9_.-]{1,100}")
PATH = re.compile(r"[A-Za-z0-9_./ -]{1,500}")

# ---------- auth: Gitea is the identity provider; Claude is a CIMD client ----------

_CIMD_SAFE_FETCH = fastmcp_cimd.ssrf_safe_fetch_response


async def _cimd_fetch(url: str, *, require_path: bool = False, max_size: int = 5120, timeout: float = 10.0,
                      overall_timeout: float = 30.0, request_headers=None, allowed_status_codes=None):
    headers = {"User-Agent": "Green-Tea-MCP/1.0 FastMCP-CIMD", **dict(request_headers or {})}
    return await _CIMD_SAFE_FETCH(url, require_path=require_path, max_size=max_size, timeout=timeout,
                                  overall_timeout=overall_timeout, request_headers=headers,
                                  allowed_status_codes=allowed_status_codes)


fastmcp_cimd.ssrf_safe_fetch_response = _cimd_fetch


class TeaOAuthProxy(OAuthProxy):
    async def get_client(self, client_id: str):
        if client_id == CLAUDE_CIMD_CLIENT_ID:  # exact local fallback for Claude's public client
            return ProxyDCRClient(client_id=CLAUDE_CIMD_CLIENT_ID, client_name="Claude", client_uri="https://claude.ai",
                                  redirect_uris=[CLAUDE_REDIRECT_URI], response_types=["code"],
                                  grant_types=["authorization_code", "refresh_token"], scope="openid profile",
                                  token_endpoint_auth_method="none")
        return await super().get_client(client_id)


class GiteaTokenVerifier(IntrospectionTokenVerifier):
    """Introspect the Gitea token and remember who it belongs to. Any Gitea account is welcome."""

    async def verify_token(self, token: str) -> AccessToken | None:
        access = await super().verify_token(token)
        if access is None:
            return None
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(f"{GITEA_URL}/login/oauth/userinfo", headers={"Authorization": f"Bearer {token}"})
        if r.status_code != 200:
            return None
        login = r.json().get("preferred_username")
        if not login:
            return None
        access.claims = {**(access.claims or {}), "login": login}
        return access


auth = TeaOAuthProxy(
    upstream_authorization_endpoint=f"{GITEA_URL}/login/oauth/authorize",
    upstream_token_endpoint=f"{GITEA_URL}/login/oauth/access_token",
    upstream_client_id=GITEA_OAUTH_CLIENT_ID,
    upstream_client_secret=GITEA_OAUTH_CLIENT_SECRET,
    token_verifier=GiteaTokenVerifier(introspection_url=f"{GITEA_URL}/login/oauth/introspect",
                                      client_id=GITEA_OAUTH_CLIENT_ID, client_secret=GITEA_OAUTH_CLIENT_SECRET,
                                      client_auth_method="client_secret_basic", required_scopes=["openid"]),
    base_url=PUBLIC_URL, resource_base_url=PUBLIC_URL, valid_scopes=OAUTH_SCOPES,
    forward_pkce=True, forward_resource=False, token_endpoint_auth_method="client_secret_basic",
    jwt_signing_key=JWT_SIGNING_KEY, require_authorization_consent="remember", enable_cimd=True,
)

mcp = FastMCP(
    "Teahub",
    version="1.0.0",
    website_url=GITEA_URL,
    instructions=("Teahub tools against " + GITEA_URL + ". Every tool acts as you, with your own Gitea permissions: "
                  "issues, comments, repositories, files, Actions. Reads and writes are bounded; nothing deletes. "
                  "Repositories are named owner/repo."),
    auth=auth,
    mask_error_details=True,
)

# ---------- the caller ----------


class InputError(ToolError):
    pass


def actor() -> str:
    access = get_access_token()
    return str(((access.claims or {}) if access else {}).get("login") or "")


def actor_token() -> str:
    access = get_access_token()
    if access is None or not access.token:
        raise ToolError("not signed in")
    return access.token


async def gitea(method: str, path: str, *, params: dict[str, Any] | None = None, json: Any = None) -> Any:
    async with httpx.AsyncClient(timeout=20) as client:
        r = await client.request(method, f"{GITEA_URL}/api/v1/{path.lstrip('/')}", params=params, json=json,
                                 headers={"Authorization": f"Bearer {actor_token()}", "Accept": "application/json"})
    if r.status_code >= 400:
        detail = ""
        try:
            detail = str(r.json().get("message") or "")[:200]
        except ValueError:
            pass
        if r.status_code < 500:
            raise ToolError(f"Gitea refused ({r.status_code}): {detail or 'no detail'}")
        raise ToolError("Gitea is unavailable right now")
    if r.status_code == 204 or not r.content:
        return {}
    return r.json()


def repo_path(repo: str) -> str:
    owner, _, name = repo.partition("/")
    if not (NAME.fullmatch(owner) and NAME.fullmatch(name)):
        raise InputError("repo must be owner/repo")
    return f"repos/{owner}/{name}"


def file_path(path: str) -> str:
    path = path.strip().strip("/")
    if not PATH.fullmatch(path) or ".." in path.split("/"):
        raise InputError("path must be a plain repository path")
    return path


def bounded(text: str, what: str) -> str:
    if len(text.encode("utf-8")) > MAX_BYTES:
        raise InputError(f"{what} exceeds the {MAX_BYTES}-byte limit")
    return text


def issue_row(x: dict[str, Any]) -> dict[str, Any]:
    return {"number": x.get("number"), "title": x.get("title"), "state": x.get("state"),
            "user": (x.get("user") or {}).get("login"), "labels": [l.get("name") for l in x.get("labels") or []],
            "milestone": (x.get("milestone") or {}).get("title"), "updated_at": x.get("updated_at"),
            "url": x.get("html_url"), "is_pull": bool(x.get("pull_request"))}


# ---------- tools ----------


@mcp.tool(description="Who am I on Teahub, and which repositories can I see.", annotations={"readOnlyHint": True})
async def tea_whoami() -> dict[str, Any]:
    me = await gitea("GET", "user")
    repos = await gitea("GET", "user/repos", params={"limit": 50})
    return {"login": me.get("login"), "admin": me.get("is_admin"), "gitea": GITEA_URL,
            "repos": [r.get("full_name") for r in repos or []]}


@mcp.tool(description="List issues (or pull requests) of a repository, newest first.", annotations={"readOnlyHint": True})
async def tea_issues(repo: str, state: str = "open", kind: str = "issues", labels: str = "", page: int = 1,
                     limit: int = 20) -> dict[str, Any]:
    if state not in {"open", "closed", "all"} or kind not in {"issues", "pulls"} or page < 1 or not 1 <= limit <= 50:
        raise InputError("state open|closed|all, kind issues|pulls, page>=1, limit 1-50")
    rows = await gitea("GET", f"{repo_path(repo)}/issues",
                       params={"state": state, "type": kind, "labels": labels, "page": page, "limit": limit, "sort": "updated"})
    return {"repo": repo, "page": page, "count": len(rows or []), "issues": [issue_row(x) for x in rows or []]}


@mcp.tool(description="Read one issue with its body.", annotations={"readOnlyHint": True})
async def tea_issue_get(repo: str, number: int) -> dict[str, Any]:
    if number < 1:
        raise InputError("number must be positive")
    x = await gitea("GET", f"{repo_path(repo)}/issues/{number}")
    return {**issue_row(x), "body": (x.get("body") or "")[:MAX_BYTES], "repo": repo}


@mcp.tool(description="Open an issue as yourself.", annotations={"destructiveHint": False})
async def tea_issue_create(repo: str, title: str, body: str = "", labels: list[str] | None = None) -> dict[str, Any]:
    title = title.strip()
    if not 1 <= len(title) <= 255:
        raise InputError("title must be 1-255 characters")
    payload: dict[str, Any] = {"title": title, "body": bounded(body, "body")}
    if labels:
        known = {l["name"]: l["id"] for l in await gitea("GET", f"{repo_path(repo)}/labels", params={"limit": 50})}
        payload["labels"] = [known[n] for n in labels if n in known]
    x = await gitea("POST", f"{repo_path(repo)}/issues", json=payload)
    return {"created": True, **issue_row(x)}


@mcp.tool(description="Edit an issue's title, body, state or assignees.", annotations={"destructiveHint": False})
async def tea_issue_edit(repo: str, number: int, title: str | None = None, body: str | None = None,
                         state: str | None = None, assignees: list[str] | None = None) -> dict[str, Any]:
    if number < 1:
        raise InputError("number must be positive")
    patch: dict[str, Any] = {}
    if title is not None:
        title = title.strip()
        if not 1 <= len(title) <= 255:
            raise InputError("title must be 1-255 characters")
        patch["title"] = title
    if body is not None:
        patch["body"] = bounded(body, "body")
    if state is not None:
        if state not in {"open", "closed"}:
            raise InputError("state must be open or closed")
        patch["state"] = state
    if assignees is not None:
        if len(assignees) > 20 or any(not NAME.fullmatch(a) for a in assignees):
            raise InputError("assignees: up to 20 usernames")
        patch["assignees"] = assignees
    if not patch:
        raise InputError("nothing to change")
    x = await gitea("PATCH", f"{repo_path(repo)}/issues/{number}", json=patch)
    return {"updated": True, **issue_row(x)}


@mcp.tool(description="Read the comments on an issue, oldest first.", annotations={"readOnlyHint": True})
async def tea_issue_comments(repo: str, number: int, page: int = 1, limit: int = 20) -> dict[str, Any]:
    if number < 1 or page < 1 or not 1 <= limit <= 50:
        raise InputError("number, page >= 1; limit 1-50")
    rows = await gitea("GET", f"{repo_path(repo)}/issues/{number}/comments", params={"page": page, "limit": limit})
    return {"repo": repo, "issue": number, "page": page, "count": len(rows or []),
            "comments": [{"id": c.get("id"), "user": (c.get("user") or {}).get("login"),
                          "created_at": c.get("created_at"), "body": (c.get("body") or "")[:MAX_BYTES]} for c in rows or []]}


@mcp.tool(description="Comment on an issue as yourself.", annotations={"destructiveHint": False})
async def tea_issue_comment(repo: str, number: int, body: str) -> dict[str, Any]:
    if number < 1:
        raise InputError("number must be positive")
    body = bounded(body.strip(), "body")
    if not body:
        raise InputError("body is empty")
    c = await gitea("POST", f"{repo_path(repo)}/issues/{number}/comments", json={"body": body})
    return {"repo": repo, "issue": number, "comment_id": c.get("id"), "url": c.get("html_url")}


@mcp.tool(description="Describe a repository: default branch, visibility, counts.", annotations={"readOnlyHint": True})
async def tea_repo(repo: str) -> dict[str, Any]:
    r = await gitea("GET", repo_path(repo))
    return {"full_name": r.get("full_name"), "description": r.get("description"), "private": r.get("private"),
            "default_branch": r.get("default_branch"), "mirror": r.get("mirror"), "open_issues": r.get("open_issues_count"),
            "open_pulls": r.get("open_pr_counter"), "updated_at": r.get("updated_at"), "url": r.get("html_url"),
            "permissions": r.get("permissions")}


@mcp.tool(description="Create a repository under your own account.", annotations={"destructiveHint": False})
async def tea_repo_create(name: str, description: str = "", private: bool = True, auto_init: bool = True) -> dict[str, Any]:
    if not NAME.fullmatch(name):
        raise InputError("name: letters, digits, . _ -")
    r = await gitea("POST", "user/repos", json={"name": name, "description": description[:255], "private": private,
                                                 "auto_init": auto_init, "default_branch": "main"})
    return {"created": True, "full_name": r.get("full_name"), "url": r.get("html_url")}


@mcp.tool(description="Create an organisation (you become its owner).", annotations={"destructiveHint": False})
async def tea_org_create(name: str, description: str = "", visibility: str = "public") -> dict[str, Any]:
    if not NAME.fullmatch(name):
        raise InputError("name: letters, digits, . _ -")
    if visibility not in {"public", "limited", "private"}:
        raise InputError("visibility public|limited|private")
    o = await gitea("POST", "orgs", json={"username": name, "description": description[:255], "visibility": visibility,
                                           "repo_admin_change_team_access": True})
    return {"created": True, "org": o.get("username"), "url": f"{GITEA_URL}/{o.get('username')}"}


@mcp.tool(description="Add a user to an organisation team (default: Owners). You must be an owner of the org.",
          annotations={"destructiveHint": False})
async def tea_org_add_member(org: str, user: str, team: str = "Owners") -> dict[str, Any]:
    if not (NAME.fullmatch(org) and NAME.fullmatch(user) and NAME.fullmatch(team)):
        raise InputError("org, user, team: letters, digits, . _ -")
    teams = await gitea("GET", f"orgs/{org}/teams", params={"limit": 50})
    match = next((t for t in teams or [] if t.get("name", "").lower() == team.lower()), None)
    if not match:
        raise InputError(f"no team {team!r} in {org}")
    await gitea("PUT", f"teams/{match['id']}/members/{user}")
    return {"org": org, "team": match["name"], "user": user, "added": True}


@mcp.tool(description="Create a repository under an organisation you own.", annotations={"destructiveHint": False})
async def tea_org_repo_create(org: str, name: str, description: str = "", private: bool = False,
                              auto_init: bool = True) -> dict[str, Any]:
    if not (NAME.fullmatch(org) and NAME.fullmatch(name)):
        raise InputError("org, name: letters, digits, . _ -")
    r = await gitea("POST", f"orgs/{org}/repos", json={"name": name, "description": description[:255], "private": private,
                                                        "auto_init": auto_init, "default_branch": "main"})
    return {"created": True, "full_name": r.get("full_name"), "url": r.get("html_url")}


@mcp.tool(description="Read a file from a repository (text, up to 200 KB).", annotations={"readOnlyHint": True})
async def tea_file_get(repo: str, path: str, ref: str = "") -> dict[str, Any]:
    x = await gitea("GET", f"{repo_path(repo)}/contents/{file_path(path)}", params={"ref": ref} if ref else None)
    if x.get("type") != "file":
        raise InputError("path is not a file")
    try:
        raw = base64.b64decode(x.get("content") or "")
    except (binascii.Error, ValueError):
        raise InputError("file is not decodable")
    if len(raw) > MAX_BYTES:
        raise InputError(f"file exceeds the {MAX_BYTES}-byte limit")
    return {"repo": repo, "path": x.get("path"), "sha": x.get("sha"), "size": x.get("size"),
            "content": raw.decode("utf-8", errors="replace")}


@mcp.tool(description="Create or update one text file on a branch (a commit as yourself). Branch protection still applies.",
          annotations={"destructiveHint": False})
async def tea_file_change(repo: str, path: str, content: str, message: str, branch: str = "") -> dict[str, Any]:
    path = file_path(path)
    message = message.strip()
    if not 1 <= len(message) <= 500:
        raise InputError("message must be 1-500 characters")
    content = bounded(content, "content")
    rp = repo_path(repo)
    if not branch:
        branch = (await gitea("GET", rp)).get("default_branch") or "main"
    payload: dict[str, Any] = {"content": base64.b64encode(content.encode("utf-8")).decode(), "message": message,
                               "branch": branch}
    try:
        current = await gitea("GET", f"{rp}/contents/{path}", params={"ref": branch})
        payload["sha"] = current.get("sha")
        x = await gitea("PUT", f"{rp}/contents/{path}", json=payload)
    except ToolError as e:
        if "404" not in str(e):
            raise
        x = await gitea("POST", f"{rp}/contents/{path}", json=payload)
    commit = (x.get("commit") or {})
    return {"repo": repo, "path": path, "branch": branch, "commit": (commit.get("sha") or "")[:12], "url": commit.get("html_url")}


@mcp.tool(description="List a directory of a repository.", annotations={"readOnlyHint": True})
async def tea_tree(repo: str, path: str = "", ref: str = "") -> dict[str, Any]:
    p = file_path(path) if path.strip("/") else ""
    rows = await gitea("GET", f"{repo_path(repo)}/contents/{p}", params={"ref": ref} if ref else None)
    if isinstance(rows, dict):
        rows = [rows]
    return {"repo": repo, "path": p, "entries": [{"name": e.get("name"), "type": e.get("type"), "size": e.get("size")}
                                                  for e in rows or []][:500]}


@mcp.tool(description="Recent Actions runs of a repository.", annotations={"readOnlyHint": True})
async def tea_actions(repo: str, limit: int = 10) -> dict[str, Any]:
    if not 1 <= limit <= 50:
        raise InputError("limit 1-50")
    x = await gitea("GET", f"{repo_path(repo)}/actions/runs", params={"limit": limit})
    runs = x.get("workflow_runs") if isinstance(x, dict) else x
    return {"repo": repo, "runs": [{"id": r.get("id"), "name": r.get("name") or r.get("display_title"), "status": r.get("status"),
                                    "conclusion": r.get("conclusion"), "branch": r.get("head_branch"),
                                    "updated_at": r.get("updated_at"), "url": r.get("html_url")} for r in runs or []]}


@mcp.tool(description="Run a workflow by file name on a branch (workflow_dispatch).", annotations={"destructiveHint": False})
async def tea_workflow_dispatch(repo: str, workflow: str, ref: str = "main", inputs: dict[str, str] | None = None) -> dict[str, Any]:
    if not re.fullmatch(r"[A-Za-z0-9_.-]+\.ya?ml", workflow):
        raise InputError("workflow must be a .yml file name")
    if not NAME.fullmatch(ref.replace("/", "-")):
        raise InputError("ref must be a branch or tag")
    await gitea("POST", f"{repo_path(repo)}/actions/workflows/{workflow}/dispatches", json={"ref": ref, "inputs": inputs or {}})
    return {"repo": repo, "workflow": workflow, "ref": ref, "dispatched": True}


# ---------- app ----------


async def healthz(_: Request) -> JSONResponse:
    return JSONResponse({"status": "ok", "service": "Green Tea MCP", "gitea": GITEA_URL})


app = mcp.http_app(path="/mcp", transport="streamable-http", stateless_http=True, json_response=True,
                   allowed_hosts=ALLOWED_HOSTS)
app.routes.append(Route("/healthz", healthz, methods=["GET"]))

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")), proxy_headers=True, forwarded_allow_ips="*")