"""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="*")