diff --git a/mcp/README.md b/mcp/README.md index 526b5b7..831537b 100644 --- a/mcp/README.md +++ b/mcp/README.md @@ -1,7 +1,16 @@ -# Green Tea MCP +# Teahub MCP -Teahub's MCP against the Gitea at `GITEA_URL`. Sign-in is Gitea's OAuth (Gitea signs people in with GitHub or nostr), and every tool acts with the caller's own Gitea token. No service token, no allowlist, nothing deletes. +Teahub's MCP (Fly app `green-tea-mcp`, `https://mcp.teahub.xyz/mcp`) against the Gitea at `GITEA_URL`. Sign-in is Gitea's OAuth (Gitea signs people in with GitHub or nostr), and every tool acts with the caller's own Gitea token: no service token, no allowlist, nothing that outruns the caller's own permissions. Reads and writes are bounded, and no tool deletes a repository, branch, file, issue or comment. Env: `GITEA_URL`, `PUBLIC_URL`, `ALLOWED_HOSTS`; secrets `GITEA_OAUTH_CLIENT_ID`, `GITEA_OAUTH_CLIENT_SECRET` (an OAuth2 application on the Gitea with redirect `PUBLIC_URL/auth/callback`), `JWT_SIGNING_KEY`. State: `/data` (FastMCP's OAuth store). -Tools: `tea_whoami`, `tea_issues`, `tea_issue_get/create/edit`, `tea_issue_comments/comment`, `tea_repo`, `tea_repo_create`, `tea_org_create`, `tea_org_add_member`, `tea_org_repo_create`, `tea_tree`, `tea_file_get/change`, `tea_actions`, `tea_workflow_dispatch`. Repos are `owner/repo`. +Tools (29), all acting as you. Their parameters are typed, so a bad argument is refused before Gitea is asked anything. + +- Issues and pull requests: `tea_issues`, `tea_issue_get/create/edit`, `tea_issue_comments/comment`, `tea_labels`, `tea_label` (add or remove by name), `tea_pull_get` (branches, mergeability, checks, diff), `tea_pull_create`, `tea_pull_review`, `tea_pull_merge` (keeps the head branch). +- Repositories and organisations: `tea_whoami`, `tea_repo`, `tea_repo_edit` (description, website, visibility, default branch, archived), `tea_repo_create`, `tea_org_create`, `tea_org_repo_create`, `tea_org_add_member` (name the team), `tea_hook_create` (one hook per URL). +- Files: `tea_tree`, `tea_file_get`, `tea_file_change`, `tea_commit` (up to 50 files, text or base64, 3 MB in all), `tea_branch_create`. +- Actions: `tea_actions` (recent jobs, each linking its run), `tea_workflow_dispatch`, `tea_secret_set` (never echoed), `tea_variable_set`. + +New repositories are private unless asked otherwise, and labels must already exist. Gitea 1.24 has no API that lists a run's jobs, so there is no log tool; `tea_actions` links each run's page, where the logs are. + +Tests: from `mcp/`, `python -m unittest discover -s tests` (a fake Gitea; needs `requirements.txt`). diff --git a/mcp/tea_mcp/server.py b/mcp/tea_mcp/server.py index 043a303..2d41543 100644 --- a/mcp/tea_mcp/server.py +++ b/mcp/tea_mcp/server.py @@ -1,16 +1,18 @@ -"""Green Tea MCP: Teahub's MCP against the Gitea at GITEA_URL. +"""Teahub MCP (Fly app green-tea-mcp): Teahub's tools 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 +Every tool acts with the caller's own Gitea authority. Sign-in is Gitea's OAuth (Gitea signs people in with 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. +permissions. An owner can archive a repository, add a hook or set a secret because Gitea lets owners do that, not +because this server does. Reads and writes are bounded, and no tool deletes a repository, branch, file, issue or +comment. """ from __future__ import annotations import base64 import binascii import os -import re -from typing import Any +from typing import Annotated, Any, Literal +from urllib.parse import quote import httpx import uvicorn @@ -21,7 +23,7 @@ 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 pydantic import BaseModel, Field from starlette.requests import Request from starlette.responses import JSONResponse from starlette.routing import Route @@ -41,14 +43,21 @@ 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()] +VERSION = "2.0.0" 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}") +MAX_BYTES = 200_000 # one text file, body, comment or diff +MAX_COMMIT_BYTES = 3_000_000 # all files of one commit, decoded (MCP requests themselves stop at 4 MiB) +READ = {"readOnlyHint": True} +WRITE = {"destructiveHint": False} +AGAIN = {**WRITE, "idempotentHint": True} # the same call twice changes nothing more -# ---------- auth: Gitea is the identity provider; Claude is a CIMD client ---------- +# ---------- sign-in: Gitea is the identity provider; Claude is a CIMD client ---------- +# Two shims, both for Claude's connector. Claude names itself by a client-metadata URL (CIMD), and fetching that +# document without a User-Agent is refused upstream, so the fetch below adds one. If the fetch fails anyway, +# TeaOAuthProxy.get_client answers for Claude's public client from the values Claude publishes. Re-check both when +# fastmcp is upgraded: one patches a module function, the other overrides a method. _CIMD_SAFE_FETCH = fastmcp_cimd.ssrf_safe_fetch_response @@ -66,7 +75,7 @@ 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 + if client_id == CLAUDE_CIMD_CLIENT_ID: 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", @@ -74,32 +83,15 @@ class TeaOAuthProxy(OAuthProxy): 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"]), + token_verifier=IntrospectionTokenVerifier(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, @@ -107,25 +99,52 @@ auth = TeaOAuthProxy( mcp = FastMCP( "Teahub", - version="1.0.0", + version=VERSION, 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."), + "issues, pull requests and reviews, labels, repositories, files and commits, branches, Actions " + "(jobs, dispatch, secrets, variables), hooks and organisations. Reads and writes are bounded and no " + "tool deletes. Repositories are named owner/repo."), auth=auth, mask_error_details=True, ) -# ---------- the caller ---------- +# ---------- parameter types: FastMCP checks these and publishes them in each tool's schema ---------- + +OWNER = "[A-Za-z0-9][A-Za-z0-9_.-]{0,99}" +Owner = Annotated[str, Field(pattern=f"^{OWNER}$")] +Repo = Annotated[str, Field(pattern=f"^{OWNER}/[A-Za-z0-9_.-]{{1,100}}$", description="owner/repo")] +RepoName = Annotated[str, Field(pattern=r"^[A-Za-z0-9_.-]{1,100}$")] +Number = Annotated[int, Field(ge=1, description="issue or pull request number")] +Page = Annotated[int, Field(ge=1, le=1000)] +Limit = Annotated[int, Field(ge=1, le=50)] +Ref = Annotated[str, Field(pattern=r"^[A-Za-z0-9_./+-]{0,200}$", + description="branch, tag or commit; empty means the default branch")] +Branch = Annotated[str, Field(pattern=r"^[A-Za-z0-9_./+-]{1,200}$")] +Title = Annotated[str, Field(min_length=1, max_length=255)] +Short = Annotated[str, Field(max_length=255)] +Message = Annotated[str, Field(min_length=1, max_length=5000)] +Text = Annotated[str, Field(max_length=MAX_BYTES)] +FilePath = Annotated[str, Field(pattern=r"^[^\x00-\x1f\x7f\\]{1,500}$", description="path in the repository")] +Labels = Annotated[list[Annotated[str, Field(min_length=1, max_length=100)]], Field(max_length=20)] +VarName = Annotated[str, Field(pattern=r"^[A-Za-z_][A-Za-z0-9_]{0,99}$")] +Value = Annotated[str, Field(min_length=1, max_length=65536)] + + +class FileChange(BaseModel): + """One file of a commit: exactly one of `content` (UTF-8 text) or `content_base64` (any bytes).""" + path: FilePath + content: str | None = None + content_base64: str | None = None class InputError(ToolError): pass -def actor() -> str: - access = get_access_token() - return str(((access.claims or {}) if access else {}).get("login") or "") +# ---------- Gitea, as the caller ---------- + +_transport: httpx.AsyncBaseTransport | None = None # tests put a fake Gitea here def actor_token() -> str: @@ -135,36 +154,96 @@ def actor_token() -> str: 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: +def gitea_client() -> httpx.AsyncClient: + return httpx.AsyncClient(base_url=f"{GITEA_URL}/api/v1/", timeout=60, transport=_transport, + headers={"Authorization": f"Bearer {actor_token()}", "Accept": "application/json"}) + + +def refused(r: httpx.Response) -> ToolError: + if r.status_code >= 500: + return ToolError("Gitea is unavailable right now") + try: + detail = str(r.json().get("message") or "")[:300] + except (ValueError, AttributeError): 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() + return ToolError(f"Gitea refused ({r.status_code}): {detail or 'no detail'}") + + +def unreachable(e: Exception) -> ToolError: + return ToolError(f"Gitea did not answer ({type(e).__name__})") + + +async def request(method: str, path: str, **kwargs: Any) -> httpx.Response: + try: + async with gitea_client() as client: + return await client.request(method, path, **kwargs) + except httpx.TransportError as e: + raise unreachable(e) from None + + +async def gitea(method: str, path: str, *, params: dict[str, Any] | None = None, json: Any = None, + missing_ok: bool = False) -> Any: + """One Gitea API call: its JSON answer, or None for a 404 when `missing_ok`. Any other refusal raises.""" + r = await request(method, path, params=params, json=json) + if missing_ok and r.status_code == 404: + return None + if r.status_code >= 400: + raise refused(r) + return r.json() if r.content and r.status_code != 204 else {} + + +async def listing(path: str, page: int, limit: int, params: dict[str, Any] | None = None) -> tuple[list[Any], bool]: + """One page of a Gitea list, and whether more follow (from X-Total-Count when Gitea sends it).""" + r = await request("GET", path, params={**(params or {}), "page": page, "limit": limit}) + if r.status_code >= 400: + raise refused(r) + rows, total = r.json() or [], r.headers.get("x-total-count", "") + return rows, page * limit < int(total) if total.isdigit() else len(rows) == limit + + +async def pages(path: str, params: dict[str, Any] | None = None, most: int = 200) -> tuple[list[Any], bool]: + """Up to `most` rows of a Gitea list, and whether there were more.""" + rows: list[Any] = [] + for page in range(1, most // 50 + 1): + batch, more = await listing(path, page, 50, params) + rows += batch + if not more: + return rows, False + return rows, True + + +async def gitea_text(path: str, cap: int) -> tuple[str, bool]: + """A text answer such as a diff: its first `cap` bytes, and whether there was more.""" + buf = b"" + try: + async with gitea_client() as client, client.stream("GET", path) as r: + if r.status_code >= 400: + await r.aread() + raise refused(r) + async for chunk in r.aiter_bytes(): + buf += chunk + if len(buf) > cap: + break + except httpx.TransportError as e: + raise unreachable(e) from None + return buf[:cap].decode("utf-8", "replace"), len(buf) > cap def repo_path(repo: str) -> str: - owner, _, name = repo.partition("/") - if not (NAME.fullmatch(owner) and NAME.fullmatch(name)): + if repo.split("/")[1] in (".", ".."): raise InputError("repo must be owner/repo") - return f"repos/{owner}/{name}" + return f"repos/{repo}" def file_path(path: str) -> str: - path = path.strip().strip("/") - if not PATH.fullmatch(path) or ".." in path.split("/"): + parts = [p for p in path.strip().split("/") if p] + if not parts or {".", ".."} & set(parts): raise InputError("path must be a plain repository path") - return path + return "/".join(parts) + + +def contents(rp: str, path: str) -> str: + return f"{rp}/contents/{quote(path)}" if path else f"{rp}/contents" def bounded(text: str, what: str) -> str: @@ -173,231 +252,391 @@ def bounded(text: str, what: str) -> str: return text +def clip(text: str | None, key: str = "body") -> dict[str, Any]: + text = text or "" + return {key: text[:MAX_BYTES], **({f"{key}_truncated": True} if len(text) > MAX_BYTES else {})} + + +async def default_branch(rp: str) -> str: + return (await gitea("GET", rp)).get("default_branch") or "main" + + +async def label_ids(rp: str, names: list[str]) -> list[int]: + known = {l.get("name"): l.get("id") for l in (await pages(f"{rp}/labels"))[0]} + unknown = [n for n in names if n not in known] + if unknown: + has = ", ".join(sorted(known)) or "none" + raise InputError(f"unknown labels: {', '.join(unknown)} (the repository has: {has})") + return [known[n] for n in names] + + 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"))} + "url": x.get("html_url"), "is_pull": bool(x.get("pull_request") or x.get("head"))} -# ---------- tools ---------- +# ---------- you ---------- -@mcp.tool(description="Who am I on Teahub, and which repositories can I see.", annotations={"readOnlyHint": True}) +@mcp.tool(description="Who am I on Teahub, and which repositories can I see (up to 200).", annotations=READ) async def tea_whoami() -> dict[str, Any]: me = await gitea("GET", "user") - repos = await gitea("GET", "user/repos", params={"limit": 50}) + repos, more = await pages("user/repos") return {"login": me.get("login"), "admin": me.get("is_admin"), "gitea": GITEA_URL, - "repos": [r.get("full_name") for r in repos or []]} + "repos": [r.get("full_name") for r in repos], "truncated": more} -@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 []]} +# ---------- issues, labels and pull requests ---------- -@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") +@mcp.tool(description="List issues or pull requests of a repository, newest first. `labels` keeps only those " + "carrying all of them.", annotations=READ) +async def tea_issues(repo: Repo, state: Literal["open", "closed", "all"] = "open", + kind: Literal["issues", "pulls"] = "issues", labels: Labels | None = None, page: Page = 1, + limit: Limit = 20) -> dict[str, Any]: + params = {"state": state, "type": kind, **({"labels": ",".join(labels)} if labels else {})} + rows, more = await listing(f"{repo_path(repo)}/issues", page, limit, params) + return {"repo": repo, "page": page, "more": more, "issues": [issue_row(x) for x in rows]} + + +@mcp.tool(description="Read one issue or pull request with its body.", annotations=READ) +async def tea_issue_get(repo: Repo, number: Number) -> dict[str, Any]: x = await gitea("GET", f"{repo_path(repo)}/issues/{number}") - return {**issue_row(x), "body": (x.get("body") or "")[:MAX_BYTES], "repo": repo} + return {"repo": repo, **issue_row(x), **clip(x.get("body"))} -@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")} +@mcp.tool(description="Read the comments on an issue or pull request, oldest first.", annotations=READ) +async def tea_issue_comments(repo: Repo, number: Number, page: Page = 1, limit: Limit = 20) -> dict[str, Any]: + rows = await gitea("GET", f"{repo_path(repo)}/issues/{number}/comments") or [] # Gitea sends them all, unpaged + start = (page - 1) * limit + return {"repo": repo, "issue": number, "page": page, "more": len(rows) > start + limit, + "comments": [{"id": c.get("id"), "user": (c.get("user") or {}).get("login"), + "created_at": c.get("created_at"), **clip(c.get("body"))} for c in rows[start:start + limit]]} + + +@mcp.tool(description="Open an issue as yourself. Labels are names the repository already has (see tea_labels).", + annotations=WRITE) +async def tea_issue_create(repo: Repo, title: Title, body: Text = "", labels: Labels | None = None) -> dict[str, Any]: + rp = repo_path(repo) + payload: dict[str, Any] = {"title": title.strip(), "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)} + payload["labels"] = await label_ids(rp, labels) + return {"created": True, **issue_row(await gitea("POST", f"{rp}/issues", json=payload))} -@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 +@mcp.tool(description="Edit an issue's or pull request's title, body, state or assignees.", annotations=AGAIN) +async def tea_issue_edit(repo: Repo, number: Number, title: Title | None = None, body: Text | None = None, + state: Literal["open", "closed"] | None = None, + assignees: Annotated[list[Owner], Field(max_length=20)] | None = None) -> dict[str, Any]: + bounded(body or "", "body") + patch = {k: v for k, v in {"title": title and title.strip(), "body": body, "state": state, + "assignees": assignees}.items() if v is not None} 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)} + return {"updated": True, **issue_row(await gitea("PATCH", f"{repo_path(repo)}/issues/{number}", json=patch))} -@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: +@mcp.tool(description="Comment on an issue or pull request as yourself.", annotations=WRITE) +async def tea_issue_comment(repo: Repo, number: Number, body: Text) -> dict[str, Any]: + if not body.strip(): raise InputError("body is empty") - c = await gitea("POST", f"{repo_path(repo)}/issues/{number}/comments", json={"body": body}) + c = await gitea("POST", f"{repo_path(repo)}/issues/{number}/comments", json={"body": bounded(body.strip(), "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]: +@mcp.tool(description="The labels a repository has.", annotations=READ) +async def tea_labels(repo: Repo) -> dict[str, Any]: + rows, more = await pages(f"{repo_path(repo)}/labels") + return {"repo": repo, "truncated": more, + "labels": [{k: l.get(k) for k in ("name", "color", "description")} for l in rows]} + + +@mcp.tool(description="Add and/or remove labels on an issue or pull request, by name.", annotations=AGAIN) +async def tea_label(repo: Repo, number: Number, add: Labels | None = None, + remove: Labels | None = None) -> dict[str, Any]: + add, remove = add or [], remove or [] + if not (add or remove): + raise InputError("nothing to change") + rp = repo_path(repo) + ids = await label_ids(rp, add + remove) + if add: + await gitea("POST", f"{rp}/issues/{number}/labels", json={"labels": ids[:len(add)]}) + for label_id in ids[len(add):]: + await gitea("DELETE", f"{rp}/issues/{number}/labels/{label_id}", missing_ok=True) + x = await gitea("GET", f"{rp}/issues/{number}") + return {"repo": repo, "number": number, "labels": [l.get("name") for l in x.get("labels") or []]} + + +@mcp.tool(description="Read a pull request: its branches, whether it can merge, the checks on its head, and its diff " + "(first 200 KB).", annotations=READ) +async def tea_pull_get(repo: Repo, number: Number, diff: bool = True) -> dict[str, Any]: + rp = repo_path(repo) + p = await gitea("GET", f"{rp}/pulls/{number}") + head, base = p.get("head") or {}, p.get("base") or {} + sha = head.get("sha") + checks = (await gitea("GET", f"{rp}/commits/{sha}/status", missing_ok=True) if sha else None) or {} + out = {"repo": repo, **issue_row(p), **clip(p.get("body")), "head": head.get("ref"), "base": base.get("ref"), + "head_sha": (sha or "")[:12], "mergeable": p.get("mergeable"), "merged": p.get("merged"), + "checks": {"state": checks.get("state"), + "statuses": [{k: s.get(k) for k in ("context", "status", "description", "target_url")} + for s in checks.get("statuses") or []]}} + if diff: + out["diff"], out["diff_truncated"] = await gitea_text(f"{rp}/pulls/{number}.diff", MAX_BYTES) + return out + + +@mcp.tool(description="Open a pull request from a branch as yourself (base defaults to the default branch).", + annotations=WRITE) +async def tea_pull_create(repo: Repo, head: Branch, title: Title, base: Ref = "", body: Text = "") -> dict[str, Any]: + rp = repo_path(repo) + p = await gitea("POST", f"{rp}/pulls", json={"head": head, "base": base or await default_branch(rp), + "title": title.strip(), "body": bounded(body, "body")}) + return {"created": True, **issue_row(p)} + + +@mcp.tool(description="Review a pull request as yourself: APPROVED, REQUEST_CHANGES or COMMENT (the last two need a " + "body). Gitea does not let you approve your own pull request.", annotations=WRITE) +async def tea_pull_review(repo: Repo, number: Number, event: Literal["APPROVED", "REQUEST_CHANGES", "COMMENT"], + body: Text = "") -> dict[str, Any]: + r = await gitea("POST", f"{repo_path(repo)}/pulls/{number}/reviews", + json={"event": event, "body": bounded(body, "body")}) + return {"repo": repo, "number": number, "review_id": r.get("id"), "state": r.get("state")} + + +@mcp.tool(description="Merge a pull request as yourself. Branch protection and required checks still apply, and the " + "head branch is kept.", annotations=WRITE) +async def tea_pull_merge(repo: Repo, number: Number, + style: Literal["merge", "squash", "rebase", "rebase-merge", "fast-forward-only"] = "squash", + message: Annotated[str, Field(max_length=5000)] = "") -> dict[str, Any]: + rp = repo_path(repo) + form: dict[str, Any] = {"Do": style, "delete_branch_after_merge": False} + if message.strip(): + form["MergeMessageField"] = message.strip() + await gitea("POST", f"{rp}/pulls/{number}/merge", json=form) + p = await gitea("GET", f"{rp}/pulls/{number}") + return {"repo": repo, "number": number, "style": style, "merged": p.get("merged"), + "merge_commit": (p.get("merge_commit_sha") or "")[:12]} + + +# ---------- repositories, organisations and hooks ---------- + + +@mcp.tool(description="Describe a repository: default branch, visibility, archived, counts and your permissions.", + annotations=READ) +async def tea_repo(repo: Repo) -> 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"), + return {"full_name": r.get("full_name"), "description": r.get("description"), "website": r.get("website"), + "private": r.get("private"), "archived": r.get("archived"), "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="Change a repository's description, website, visibility, default branch or archived state " + "(you need admin rights on it).", annotations=AGAIN) +async def tea_repo_edit(repo: Repo, description: Short | None = None, website: Short | None = None, + private: bool | None = None, archived: bool | None = None, + default_branch: Branch | None = None) -> dict[str, Any]: + patch = {k: v for k, v in {"description": description, "website": website, "private": private, + "archived": archived, "default_branch": default_branch}.items() if v is not None} + if not patch: + raise InputError("nothing to change") + r = await gitea("PATCH", repo_path(repo), json=patch) + return {"updated": True, "full_name": r.get("full_name"), **{k: r.get(k) for k in patch}} -@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, +async def new_repo(where: str, name: str, description: str, private: bool, auto_init: bool) -> dict[str, Any]: + r = await gitea("POST", where, json={"name": name, "description": description, "private": private, + "auto_init": auto_init, "default_branch": "main"}) + return {"created": True, "full_name": r.get("full_name"), "private": r.get("private"), "url": r.get("html_url")} + + +@mcp.tool(description="Create a repository under your own account (private unless you say otherwise).", + annotations=WRITE) +async def tea_repo_create(name: RepoName, description: Short = "", private: bool = True, + auto_init: bool = True) -> dict[str, Any]: + return await new_repo("user/repos", name, description, private, auto_init) + + +@mcp.tool(description="Create a repository under an organisation (private unless you say otherwise).", + annotations=WRITE) +async def tea_org_repo_create(org: Owner, name: RepoName, description: Short = "", private: bool = True, + auto_init: bool = True) -> dict[str, Any]: + return await new_repo(f"orgs/{org}/repos", name, description, private, auto_init) + + +@mcp.tool(description="Create an organisation; you become its owner.", annotations=WRITE) +async def tea_org_create(name: Owner, description: Short = "", + visibility: Literal["public", "limited", "private"] = "public") -> dict[str, Any]: + o = await gitea("POST", "orgs", json={"username": name, "description": description, "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) +@mcp.tool(description="Add a user to a team of an organisation you own. Name the team (for example Owners); there is " + "no default.", annotations=AGAIN) +async def tea_org_add_member(org: Owner, user: Owner, + team: Annotated[str, Field(min_length=1, max_length=255)]) -> dict[str, Any]: + teams, _ = await pages(f"orgs/{org}/teams") + match = next((t for t in teams if (t.get("name") or "").lower() == team.lower()), None) if not match: - raise InputError(f"no team {team!r} in {org}") + names = ", ".join(t.get("name") or "" for t in teams) or "none" + raise InputError(f"{org} has no team {team!r} (its teams: {names})") await gitea("PUT", f"teams/{match['id']}/members/{user}") - return {"org": org, "team": match["name"], "user": user, "added": True} + return {"org": org, "team": match.get("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") +@mcp.tool(description="Add a webhook to a repository, such as a pages host's push hook. Adding a URL the repository " + "already has changes nothing.", annotations=AGAIN) +async def tea_hook_create(repo: Repo, url: Annotated[str, Field(pattern=r"^https?://\S{1,500}$")], + events: Annotated[list[Literal["push", "create", "pull_request", "issues", "issue_comment", + "release"]], Field(min_length=1, max_length=6)] | None = None, + branch_filter: Short = "", secret: Short = "") -> dict[str, Any]: 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} + hooks, _ = await pages(f"{rp}/hooks") + same = next((h for h in hooks if (h.get("config") or {}).get("url") == url), None) + if same: + return {"repo": repo, "hook_id": same.get("id"), "created": False, "events": same.get("events")} + config = {"url": url, "content_type": "json", **({"secret": secret} if secret else {})} + h = await gitea("POST", f"{rp}/hooks", json={"type": "gitea", "active": True, "events": events or ["push"], + "branch_filter": branch_filter, "config": config}) + return {"repo": repo, "hook_id": h.get("id"), "created": True, "events": h.get("events")} + + +# ---------- files, commits and branches ---------- + + +@mcp.tool(description="List a directory of a repository (up to 500 entries).", annotations=READ) +async def tea_tree(repo: Repo, path: Annotated[str, Field(pattern=r"^[^\x00-\x1f\x7f\\]{0,500}$")] = "", + ref: Ref = "") -> dict[str, Any]: + p = file_path(path) if path.strip(" /") else "" + rows = await gitea("GET", contents(repo_path(repo), p), params={"ref": ref} if ref else None) + rows = [rows] if isinstance(rows, dict) else rows or [] + return {"repo": repo, "path": p, "truncated": len(rows) > 500, + "entries": [{k: e.get(k) for k in ("name", "type", "size")} for e in rows[:500]]} + + +@mcp.tool(description="Read a text file from a repository (up to 200 KB). Larger and binary files report only their " + "size and sha.", annotations=READ) +async def tea_file_get(repo: Repo, path: FilePath, ref: Ref = "") -> dict[str, Any]: + p = file_path(path) + x = await gitea("GET", contents(repo_path(repo), p), params={"ref": ref} if ref else None) + if not isinstance(x, dict) or x.get("type") != "file": + raise InputError(f"{p} is not a file") + meta = {"repo": repo, "path": x.get("path"), "sha": x.get("sha"), "size": x.get("size")} + if (x.get("size") or 0) > MAX_BYTES: + return {**meta, "too_large": True} 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")} + return {**meta, "content": base64.b64decode(x.get("content") or "").decode("utf-8")} + except (binascii.Error, ValueError): + return {**meta, "binary": True} -@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="Create or update one text file on a branch, as a commit by you. Branch protection applies.", + annotations=WRITE) +async def tea_file_change(repo: Repo, path: FilePath, content: Text, message: Message, + branch: Ref = "") -> dict[str, Any]: + return await commit(repo, message, [FileChange(path=path, content=content)], branch, "") -@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="Commit up to 50 files at once, as you. Each file is UTF-8 `content` or `content_base64` (images " + "and other binaries), 3 MB in all; files are created or updated, never deleted. `new_branch` " + "starts a branch from `branch` for the commit.", annotations=WRITE) +async def tea_commit(repo: Repo, message: Message, + files: Annotated[list[FileChange], Field(min_length=1, max_length=50)], branch: Ref = "", + new_branch: Ref = "") -> dict[str, Any]: + return await commit(repo, message, files, branch, new_branch) -@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 {}}) +async def commit(repo: str, message: str, files: list[FileChange], branch: str, new_branch: str) -> dict[str, Any]: + rp = repo_path(repo) + blobs: dict[str, bytes] = {} + for f in files: # check every file before asking Gitea anything + path = file_path(f.path) + if path in blobs: + raise InputError(f"{path} appears twice") + if (f.content is None) == (f.content_base64 is None): + raise InputError(f"{path}: give exactly one of content or content_base64") + try: + blobs[path] = (f.content.encode("utf-8") if f.content is not None + else base64.b64decode(f.content_base64, validate=True)) + except (binascii.Error, ValueError): + raise InputError(f"{path}: content_base64 is not valid base64") from None + if sum(map(len, blobs.values())) > MAX_COMMIT_BYTES: + raise InputError(f"the files come to more than {MAX_COMMIT_BYTES} bytes") + branch = branch or await default_branch(rp) + ops = [] + for path, data in blobs.items(): + current = await gitea("GET", contents(rp, path), params={"ref": branch}, missing_ok=True) + if current is not None and (not isinstance(current, dict) or current.get("type") != "file"): + raise InputError(f"{path} exists and is not a file") + op = {"operation": "create", "path": path, "content": base64.b64encode(data).decode()} + if current is not None: + op.update(operation="update", sha=current.get("sha")) + ops.append(op) + payload = {"files": ops, "message": message.strip(), "branch": branch, + **({"new_branch": new_branch} if new_branch else {})} + c = (await gitea("POST", f"{rp}/contents", json=payload)).get("commit") or {} + return {"repo": repo, "branch": new_branch or branch, "commit": (c.get("sha") or "")[:12], "url": c.get("html_url"), + "files": {o["path"]: o["operation"] for o in ops}} + + +@mcp.tool(description="Create a branch from another branch, a tag or a commit (default: the default branch).", + annotations=WRITE) +async def tea_branch_create(repo: Repo, name: Branch, from_ref: Ref = "") -> dict[str, Any]: + rp = repo_path(repo) + b = await gitea("POST", f"{rp}/branches", json={"new_branch_name": name, + "old_ref_name": from_ref or await default_branch(rp)}) + return {"repo": repo, "branch": b.get("name"), "commit": ((b.get("commit") or {}).get("id") or "")[:12]} + + +# ---------- Actions ---------- + + +@mcp.tool(description="Recent Actions jobs of a repository, newest first: each with its run number, status and a " + "link to the run, where its logs are. (This Gitea's API has no endpoint for job logs.)", + annotations=READ) +async def tea_actions(repo: Repo, page: Page = 1, limit: Limit = 10) -> dict[str, Any]: + x = await gitea("GET", f"{repo_path(repo)}/actions/tasks", params={"page": page, "limit": limit}) or {} + return {"repo": repo, "page": page, "more": page * limit < (x.get("total_count") or 0), "jobs": [ + {"id": t.get("id"), "run": t.get("run_number"), "job": t.get("name"), "title": t.get("display_title"), + "workflow": t.get("workflow_id"), "event": t.get("event"), "status": t.get("status"), + "branch": t.get("head_branch"), "sha": (t.get("head_sha") or "")[:12], "started_at": t.get("run_started_at"), + "updated_at": t.get("updated_at"), "url": t.get("url")} for t in x.get("workflow_runs") or []]} + + +@mcp.tool(description="Run a workflow by file name (workflow_dispatch) on a branch, by default the default branch.", + annotations=WRITE) +async def tea_workflow_dispatch(repo: Repo, workflow: Annotated[str, Field(pattern=r"^[A-Za-z0-9_.-]+\.ya?ml$")], + ref: Ref = "", inputs: dict[str, str] | None = None) -> dict[str, Any]: + rp = repo_path(repo) + ref = ref or await default_branch(rp) + await gitea("POST", f"{rp}/actions/workflows/{workflow}/dispatches", json={"ref": ref, "inputs": inputs or {}}) return {"repo": repo, "workflow": workflow, "ref": ref, "dispatched": True} +@mcp.tool(description="Set a repository Actions secret. Gitea stores the value and never returns it.", + annotations=AGAIN) +async def tea_secret_set(repo: Repo, name: VarName, value: Value) -> dict[str, Any]: + await gitea("PUT", f"{repo_path(repo)}/actions/secrets/{name.upper()}", json={"data": value}) + return {"repo": repo, "secret": name.upper(), "set": True} + + +@mcp.tool(description="Set a repository Actions variable, creating it if needed.", annotations=AGAIN) +async def tea_variable_set(repo: Repo, name: VarName, value: Value) -> dict[str, Any]: + path = f"{repo_path(repo)}/actions/variables/{name.upper()}" + exists = await gitea("GET", path, missing_ok=True) is not None + await gitea("PUT" if exists else "POST", path, json={"value": value}) + return {"repo": repo, "variable": name.upper(), "created": not exists} + + # ---------- app ---------- async def healthz(_: Request) -> JSONResponse: - return JSONResponse({"status": "ok", "service": "Green Tea MCP", "gitea": GITEA_URL}) + return JSONResponse({"status": "ok", "service": "Teahub MCP", "version": VERSION, "gitea": GITEA_URL}) app = mcp.http_app(path="/mcp", transport="streamable-http", stateless_http=True, json_response=True, diff --git a/mcp/tests/test_server.py b/mcp/tests/test_server.py new file mode 100644 index 0000000..a547dc5 --- /dev/null +++ b/mcp/tests/test_server.py @@ -0,0 +1,423 @@ +"""The Teahub MCP against a fake Gitea. From mcp/: python -m unittest discover -s tests""" +from __future__ import annotations + +import base64 +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +os.environ.update({"GITEA_URL": "https://gitea.test", "PUBLIC_URL": "https://mcp.test", "GITEA_OAUTH_CLIENT_ID": "id", + "GITEA_OAUTH_CLIENT_SECRET": "not-a-secret", "JWT_SIGNING_KEY": "k" * 32, + "FASTMCP_HOME": tempfile.mkdtemp()}) + +import httpx # noqa: E402 +from fastmcp import Client # noqa: E402 +from fastmcp.exceptions import ToolError # noqa: E402 + +from tea_mcp import server # noqa: E402 + +R = "repos/o/r" +READS = {"tea_whoami", "tea_issues", "tea_issue_get", "tea_issue_comments", "tea_labels", "tea_pull_get", "tea_repo", + "tea_tree", "tea_file_get", "tea_actions"} +WRITES = {"tea_issue_create", "tea_issue_edit", "tea_issue_comment", "tea_label", "tea_pull_create", "tea_pull_review", + "tea_pull_merge", "tea_repo_edit", "tea_repo_create", "tea_org_repo_create", "tea_org_create", + "tea_org_add_member", "tea_hook_create", "tea_file_change", "tea_commit", "tea_branch_create", + "tea_workflow_dispatch", "tea_secret_set", "tea_variable_set"} +AGAIN = {"tea_issue_edit", "tea_label", "tea_repo_edit", "tea_org_add_member", "tea_hook_create", "tea_secret_set", + "tea_variable_set"} + + +def b64(data: bytes) -> str: + return base64.b64encode(data).decode() + + +def answer(data=None, status=200, total=None) -> httpx.Response: + return httpx.Response(status, json=data, headers={"x-total-count": str(total)} if total is not None else None) + + +class FakeGitea: + """Just enough of Gitea's API. Every call is kept as (method, path, query, json body).""" + + def __init__(self): + self.calls: list[tuple[str, str, dict, object]] = [] + self.raw_paths: list[bytes] = [] + self.repo_count = 230 + self.files = {"README.md": b"hello\n", "img/logo.png": b"\x89PNG\r\n\x1a\n\xff", "docs/a b.md": b"spaced\n"} + self.labels = [{"id": 1, "name": "bug", "color": "ee0701", "description": "broken"}, + {"id": 2, "name": "docs", "color": "0075ca", "description": ""}] + self.on_issue = [self.labels[0]] + self.comments = [{"id": i, "user": {"login": "tea"}, "body": f"c{i}"} for i in range(1, 46)] + self.variables = {"EXISTING": "1"} + self.hooks = [{"id": 7, "config": {"url": "https://pages.test/hook"}, "events": ["push"]}] + self.merged = False + self.diff = "diff --git a/x b/x\n" + + def __call__(self, request: httpx.Request) -> httpx.Response: + m, p, q = request.method, request.url.path.removeprefix("/api/v1/"), dict(request.url.params) + body = json.loads(request.content) if request.content else None + self.calls.append((m, p, q, body)) + self.raw_paths.append(request.url.raw_path) + if request.headers.get("authorization") != "Bearer tok": + return answer({"message": "token is required"}, 401) + page, limit = int(q.get("page", 1)), int(q.get("limit", 50)) + window = slice((page - 1) * limit, page * limit) + if (m, p) == ("GET", "user"): + return answer({"login": "tea", "is_admin": False}) + if (m, p) == ("GET", "user/repos"): + return answer([{"full_name": f"tea/r{i}"} for i in range(self.repo_count)][window], total=self.repo_count) + if (m, p) in (("POST", "user/repos"), ("POST", "orgs/acme/repos")): + owner = "tea" if p == "user/repos" else "acme" + return answer({"full_name": f"{owner}/{body['name']}", "private": body["private"]}, 201) + if (m, p) == ("POST", "orgs"): + return answer({"username": body["username"]}, 201) + if (m, p) == ("GET", "orgs/acme/teams"): + return answer([{"id": 3, "name": "Owners"}, {"id": 4, "name": "Writers"}], total=2) + if m == "PUT" and p.startswith("teams/"): + return answer(status=204) + if (m, p) == ("GET", R): + return answer({"full_name": "o/r", "default_branch": "trunk", "private": True, "archived": False}) + if (m, p) == ("PATCH", R): + return answer({"full_name": "o/r", **body}) + if (m, p) == ("GET", "repos/o/forbidden"): + return answer({"message": "user should have a permission"}, 403) + if (m, p) == ("GET", "repos/o/broken"): + return answer({"message": "boom: secret internals"}, 500) + if (m, p) == ("GET", f"{R}/labels"): + return answer(self.labels, total=len(self.labels)) + if (m, p) == ("GET", f"{R}/issues"): + return answer([{"number": n, "title": f"t{n}", "state": "open"} for n in range(25, 0, -1)][window], total=25) + if (m, p) == ("POST", f"{R}/issues"): + return answer({"number": 26, "title": body["title"], + "labels": [l for l in self.labels if l["id"] in body.get("labels", [])]}, 201) + if (m, p) == ("GET", f"{R}/issues/1"): + return answer({"number": 1, "title": "t1", "body": "x" * (server.MAX_BYTES + 5), "labels": self.on_issue}) + if (m, p) == ("PATCH", f"{R}/issues/1"): + return answer({"number": 1, "title": body.get("title", "t1"), "state": body.get("state", "open")}, 201) + if (m, p) == ("GET", f"{R}/issues/1/comments"): + return answer(self.comments, total=len(self.comments)) + if (m, p) == ("POST", f"{R}/issues/1/comments"): + return answer({"id": 99, "html_url": "https://gitea.test/o/r/issues/1#issuecomment-99"}, 201) + if (m, p) == ("POST", f"{R}/issues/1/labels"): + self.on_issue = self.on_issue + [l for l in self.labels if l["id"] in body["labels"]] + return answer(self.on_issue) + if m == "DELETE" and p.startswith(f"{R}/issues/1/labels/"): + self.on_issue = [l for l in self.on_issue if str(l["id"]) != p.rsplit("/", 1)[1]] + return answer(status=204) + if (m, p) == ("GET", f"{R}/pulls/2"): + return answer({"number": 2, "title": "p", "body": "why", "head": {"ref": "feature", "sha": "abcdef1234567890"}, + "base": {"ref": "trunk"}, "mergeable": True, "merged": self.merged, + "merge_commit_sha": "fedcba9876543210" if self.merged else None}) + if (m, p) == ("GET", f"{R}/pulls/2.diff"): + return httpx.Response(200, text=self.diff) + if (m, p) == ("GET", f"{R}/commits/abcdef1234567890/status"): + return answer({"state": "success", "statuses": [{"context": "ci / test", "status": "success", + "description": "ok", "target_url": "https://ci.test/1"}]}) + if (m, p) == ("POST", f"{R}/pulls"): + return answer({"number": 3, "title": body["title"], "head": {"ref": body["head"]}}, 201) + if (m, p) == ("POST", f"{R}/pulls/2/reviews"): + return answer({"id": 11, "state": body["event"]}) + if (m, p) == ("POST", f"{R}/pulls/2/merge"): + self.merged = True + return answer() + if (m, p) == ("GET", f"{R}/contents"): + return answer([{"name": n, "type": "file" if "." in n else "dir", "size": 1} for n in ("README.md", "img", "docs")]) + if m == "GET" and p.startswith(f"{R}/contents/"): + path = p.removeprefix(f"{R}/contents/") + if path in ("img", "docs"): + return answer([{"name": k.split("/")[1], "type": "file", "size": len(v)} + for k, v in self.files.items() if k.startswith(path + "/")]) + if path not in self.files: + return answer({"message": "object does not exist"}, 404) + data = self.files[path] + return answer({"type": "file", "path": path, "sha": f"sha-{path}", "size": len(data), "content": b64(data)}) + if (m, p) == ("POST", f"{R}/contents"): + return answer({"commit": {"sha": "c0ffee" * 7, "html_url": "https://gitea.test/o/r/commit/c0ffee"}}, 201) + if (m, p) == ("POST", f"{R}/branches"): + return answer({"name": body["new_branch_name"], "commit": {"id": "b" * 40}}, 201) + if (m, p) == ("GET", f"{R}/hooks"): + return answer(self.hooks, total=len(self.hooks)) + if (m, p) == ("POST", f"{R}/hooks"): + return answer({"id": 8, "events": body["events"]}, 201) + if (m, p) == ("GET", f"{R}/actions/tasks"): + tasks = [{"id": 40 - i, "run_number": 9 - i // 2, "name": ("test", "lint")[i % 2], "display_title": "deploy", + "workflow_id": "deploy.yml", "event": "push", "status": "success", "head_branch": "trunk", + "head_sha": "1234567890abcdef", "url": f"https://gitea.test/o/r/actions/runs/{9 - i // 2}"} + for i in range(15)] + return answer({"total_count": len(tasks), "workflow_runs": tasks[window]}) + if (m, p) == ("POST", f"{R}/actions/workflows/ci.yml/dispatches"): + return answer(status=204) + if m == "PUT" and p.startswith(f"{R}/actions/secrets/"): + return answer(status=201) + if p.startswith(f"{R}/actions/variables/"): + name = p.rsplit("/", 1)[1] + if m == "GET": + return answer({"name": name, "data": self.variables[name]}) if name in self.variables else \ + answer({"message": "not found"}, 404) + self.variables[name] = body["value"] + return answer(status=201 if m == "POST" else 204) + return answer({"message": f"no route {m} {p}"}, 404) + + +class TeaMCP(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.gitea = FakeGitea() + server._transport = httpx.MockTransport(self.gitea) + self.real_token, server.actor_token = server.actor_token, lambda: "tok" + self.client = Client(server.mcp) + await self.client.__aenter__() + + async def asyncTearDown(self): + await self.client.__aexit__(None, None, None) + server.actor_token, server._transport = self.real_token, None + + async def call(self, tool: str, **args) -> dict: + return (await self.client.call_tool(tool, args)).structured_content + + async def refused(self, tool: str, **args) -> str: + with self.assertRaises(ToolError) as caught: + await self.client.call_tool(tool, args) + return str(caught.exception) + + def sent(self, method: str, path: str | None = None) -> list[tuple]: + return [c for c in self.gitea.calls if c[0] == method and (path is None or c[1] == path)] + + async def test_tools_and_their_hints(self): + tools = {t.name: t for t in await self.client.list_tools()} + self.assertEqual(set(tools), READS | WRITES) + for name, tool in tools.items(): + with self.subTest(tool=name): + hints = tool.annotations + self.assertEqual(bool(hints.read_only_hint), name in READS) + self.assertIs(hints.destructive_hint, None if name in READS else False) + self.assertEqual(bool(hints.idempotent_hint), name in AGAIN) + self.assertEqual(tools["tea_issues"].input_schema["properties"]["limit"]["maximum"], 50) + + async def test_bad_arguments_never_reach_gitea(self): + for tool, args in [("tea_issues", {"repo": "no-slash"}), ("tea_issues", {"repo": "o/r", "limit": 51}), + ("tea_issues", {"repo": "o/r", "state": "merged"}), ("tea_repo", {"repo": "../x"}), + ("tea_repo", {"repo": "o/.."}), ("tea_file_get", {"repo": "o/r", "path": "a\\b"}), + ("tea_secret_set", {"repo": "o/r", "name": "1BAD", "value": "v"}), + ("tea_commit", {"repo": "o/r", "message": "m", "files": []}), + ("tea_pull_review", {"repo": "o/r", "number": 2, "event": "LGTM"}), + ("tea_workflow_dispatch", {"repo": "o/r", "workflow": "../x.yml"}), + ("tea_org_add_member", {"org": "acme", "user": "kim"})]: + with self.subTest(tool=tool, args=args): + await self.refused(tool, **args) + self.assertEqual(self.gitea.calls, []) + + async def test_whoami_says_truthfully_whether_it_stopped_short(self): + me = await self.call("tea_whoami") + self.assertEqual((me["login"], len(me["repos"]), me["truncated"]), ("tea", 200, True)) + self.gitea.repo_count = 200 + me = await self.call("tea_whoami") + self.assertEqual((len(me["repos"]), me["truncated"]), (200, False)) + + async def test_issues_pages(self): + first = await self.call("tea_issues", repo="o/r", limit=20) + last = await self.call("tea_issues", repo="o/r", page=2, limit=20) + self.assertEqual((len(first["issues"]), first["more"], len(last["issues"]), last["more"]), (20, True, 5, False)) + self.assertEqual(self.sent("GET", f"{R}/issues")[0][2], {"state": "open", "type": "issues", "page": "1", "limit": "20"}) + await self.call("tea_issues", repo="o/r", kind="pulls", labels=["bug", "docs"]) + self.assertEqual(self.sent("GET", f"{R}/issues")[-1][2]["labels"], "bug,docs") + + async def test_unknown_labels_are_refused(self): + message = await self.refused("tea_issue_create", repo="o/r", title="x", labels=["bug", "nope"]) + self.assertIn("unknown labels: nope", message) + self.assertIn("bug, docs", message) + self.assertEqual(self.sent("POST"), []) + made = await self.call("tea_issue_create", repo="o/r", title=" x ", labels=["docs"]) + self.assertEqual(self.sent("POST", f"{R}/issues")[0][3], {"title": "x", "body": "", "labels": [2]}) + self.assertEqual((made["number"], made["labels"]), (26, ["docs"])) + + async def test_label_adds_and_removes(self): + out = await self.call("tea_label", repo="o/r", number=1, add=["docs"], remove=["bug"]) + self.assertEqual(self.sent("POST", f"{R}/issues/1/labels")[0][3], {"labels": [2]}) + self.assertEqual([c[1] for c in self.sent("DELETE")], [f"{R}/issues/1/labels/1"]) + self.assertEqual(out["labels"], ["docs"]) + self.assertIn("nothing to change", await self.refused("tea_label", repo="o/r", number=1)) + + async def test_long_bodies_are_clipped_and_comments_paged(self): + issue = await self.call("tea_issue_get", repo="o/r", number=1) + self.assertEqual((len(issue["body"]), issue["body_truncated"]), (server.MAX_BYTES, True)) + page2 = await self.call("tea_issue_comments", repo="o/r", number=1, page=2, limit=20) + page3 = await self.call("tea_issue_comments", repo="o/r", number=1, page=3, limit=20) + self.assertEqual([c["id"] for c in page2["comments"]], list(range(21, 41))) + self.assertEqual((page2["more"], len(page3["comments"]), page3["more"]), (True, 5, False)) + self.assertNotIn("body_truncated", page2["comments"][0]) + + async def test_issue_edit_and_comment(self): + out = await self.call("tea_issue_edit", repo="o/r", number=1, state="closed", assignees=["tea"]) + self.assertEqual((self.sent("PATCH")[0][3], out["state"]), ({"state": "closed", "assignees": ["tea"]}, "closed")) + self.assertIn("nothing to change", await self.refused("tea_issue_edit", repo="o/r", number=1)) + comment = await self.call("tea_issue_comment", repo="o/r", number=1, body=" hi ") + self.assertEqual((self.sent("POST")[0][3], comment["comment_id"]), ({"body": "hi"}, 99)) + self.assertIn("body is empty", await self.refused("tea_issue_comment", repo="o/r", number=1, body=" ")) + + async def test_pull_read_review_merge(self): + self.gitea.diff = "+" * (server.MAX_BYTES + 10) + pr = await self.call("tea_pull_get", repo="o/r", number=2) + self.assertEqual((pr["head"], pr["base"], pr["head_sha"], pr["is_pull"]), ("feature", "trunk", "abcdef123456", True)) + self.assertEqual((pr["checks"]["state"], pr["checks"]["statuses"][0]["context"]), ("success", "ci / test")) + self.assertEqual((len(pr["diff"]), pr["diff_truncated"]), (server.MAX_BYTES, True)) + self.assertNotIn("diff", await self.call("tea_pull_get", repo="o/r", number=2, diff=False)) + review = await self.call("tea_pull_review", repo="o/r", number=2, event="APPROVED") + self.assertEqual(review["state"], "APPROVED") + merged = await self.call("tea_pull_merge", repo="o/r", number=2) + self.assertEqual(self.sent("POST", f"{R}/pulls/2/merge")[0][3], {"Do": "squash", "delete_branch_after_merge": False}) + self.assertEqual((merged["merged"], merged["merge_commit"]), (True, "fedcba987654")) + + async def test_pull_create_targets_the_default_branch(self): + made = await self.call("tea_pull_create", repo="o/r", head="feature", title="p") + self.assertEqual(self.sent("POST", f"{R}/pulls")[0][3], {"head": "feature", "base": "trunk", "title": "p", "body": ""}) + self.assertEqual((made["number"], made["is_pull"]), (3, True)) + + async def test_file_reads(self): + text = await self.call("tea_file_get", repo="o/r", path="/README.md") + binary = await self.call("tea_file_get", repo="o/r", path="img/logo.png") + spaced = await self.call("tea_file_get", repo="o/r", path="docs/a b.md") + self.assertEqual((text["content"], text["sha"]), ("hello\n", "sha-README.md")) + self.assertEqual((binary.get("binary"), "content" in binary), (True, False)) + self.assertEqual(spaced["content"], "spaced\n") + self.assertIn(b"/api/v1/repos/o/r/contents/docs/a%20b.md", self.gitea.raw_paths) + self.gitea.files["big.txt"] = b"x" * (server.MAX_BYTES + 1) + self.assertTrue((await self.call("tea_file_get", repo="o/r", path="big.txt"))["too_large"]) + self.assertIn("not a file", await self.refused("tea_file_get", repo="o/r", path="docs")) + self.assertIn("plain repository path", await self.refused("tea_file_get", repo="o/r", path="docs/../x")) + + async def test_tree(self): + root = await self.call("tea_tree", repo="o/r") + self.assertEqual(([e["name"] for e in root["entries"]], root["truncated"]), (["README.md", "img", "docs"], False)) + self.assertEqual(self.gitea.calls[-1][1], f"{R}/contents") + docs = await self.call("tea_tree", repo="o/r", path="docs/", ref="trunk") + self.assertEqual(([e["name"] for e in docs["entries"]], docs["path"]), (["a b.md"], "docs")) + self.assertEqual(self.gitea.calls[-1][2], {"ref": "trunk"}) + + async def test_commit_creates_updates_and_carries_binaries(self): + png = b"\x89PNG\r\n\x1a\n\x00\x01" + out = await self.call("tea_commit", repo="o/r", message=" add logo ", new_branch="art", files=[ + {"path": "README.md", "content": "hello again\n"}, {"path": "img/new.png", "content_base64": b64(png)}]) + payload = self.sent("POST", f"{R}/contents")[0][3] + self.assertEqual((payload["message"], payload["branch"], payload["new_branch"]), ("add logo", "trunk", "art")) + readme, image = payload["files"] + self.assertEqual((readme["operation"], readme["sha"], base64.b64decode(readme["content"])), + ("update", "sha-README.md", b"hello again\n")) + self.assertEqual((image["operation"], base64.b64decode(image["content"]), "sha" in image), ("create", png, False)) + self.assertEqual(out["files"], {"README.md": "update", "img/new.png": "create"}) + self.assertEqual((out["branch"], out["commit"]), ("art", "c0ffeec0ffee")) + self.assertTrue(all(c[2] == {"ref": "trunk"} for c in self.gitea.calls if c[1].startswith(f"{R}/contents/"))) + + async def test_commit_refusals(self): + for files, expected in [([{"path": "a.txt"}], "exactly one"), + ([{"path": "a.txt", "content": "x", "content_base64": b64(b"x")}], "exactly one"), + ([{"path": "a.bin", "content_base64": "not base64!"}], "not valid base64"), + ([{"path": "../a.txt", "content": "x"}], "plain repository path"), + ([{"path": "a", "content": "x"}, {"path": "/a", "content": "y"}], "appears twice")]: + with self.subTest(expected=expected): + self.assertIn(expected, await self.refused("tea_commit", repo="o/r", message="m", files=files)) + cap, server.MAX_COMMIT_BYTES = server.MAX_COMMIT_BYTES, 10 + try: + self.assertIn("more than 10 bytes", await self.refused("tea_commit", repo="o/r", message="m", + files=[{"path": "a", "content": "x" * 11}])) + finally: + server.MAX_COMMIT_BYTES = cap + self.assertEqual(self.gitea.calls, []) # all of that was refused before asking Gitea anything + self.assertIn("is not a file", await self.refused("tea_commit", repo="o/r", message="m", + files=[{"path": "docs", "content": "x"}])) + self.assertEqual(self.sent("POST"), []) + + async def test_file_change_creates_or_updates_without_guessing(self): + await self.call("tea_file_change", repo="o/r", path="new.md", content="n", message="m") + await self.call("tea_file_change", repo="o/r", path="README.md", content="u", message="m", branch="dev") + created, updated = [c[3] for c in self.sent("POST", f"{R}/contents")] + self.assertEqual((created["files"][0]["operation"], created["branch"]), ("create", "trunk")) + self.assertEqual((updated["files"][0]["operation"], updated["branch"]), ("update", "dev")) + + async def test_branch_create(self): + branch = await self.call("tea_branch_create", repo="o/r", name="feature/x") + self.assertEqual(self.sent("POST", f"{R}/branches")[0][3], {"new_branch_name": "feature/x", "old_ref_name": "trunk"}) + self.assertEqual(branch["commit"], "b" * 12) + + async def test_actions_lists_jobs_newest_first(self): + first = await self.call("tea_actions", repo="o/r") + last = await self.call("tea_actions", repo="o/r", page=2) + job = first["jobs"][0] + self.assertEqual((job["id"], job["run"], job["job"], job["workflow"], job["sha"]), + (40, 9, "test", "deploy.yml", "1234567890ab")) + self.assertEqual((len(first["jobs"]), first["more"], len(last["jobs"]), last["more"]), (10, True, 5, False)) + + async def test_workflow_dispatch_defaults_to_the_default_branch(self): + out = await self.call("tea_workflow_dispatch", repo="o/r", workflow="ci.yml", inputs={"why": "test"}) + self.assertEqual((self.sent("POST")[0][3], out["dispatched"]), ({"ref": "trunk", "inputs": {"why": "test"}}, True)) + + async def test_a_secret_is_never_echoed(self): + result = await self.client.call_tool("tea_secret_set", {"repo": "o/r", "name": "deploy_key", "value": "s3cr3t-v"}) + shown = json.dumps(result.structured_content) + "".join(getattr(c, "text", "") for c in result.content) + self.assertNotIn("s3cr3t-v", shown) + self.assertEqual(self.sent("PUT")[0][1:], (f"{R}/actions/secrets/DEPLOY_KEY", {}, {"data": "s3cr3t-v"})) + + async def test_variable_set_creates_or_updates(self): + fresh = await self.call("tea_variable_set", repo="o/r", name="fresh", value="1") + existing = await self.call("tea_variable_set", repo="o/r", name="existing", value="2") + self.assertEqual((fresh["created"], existing["created"]), (True, False)) + self.assertEqual([(c[0], c[1].rsplit("/", 1)[1]) for c in self.gitea.calls if c[0] in ("POST", "PUT")], + [("POST", "FRESH"), ("PUT", "EXISTING")]) + + async def test_hook_create_is_idempotent(self): + same = await self.call("tea_hook_create", repo="o/r", url="https://pages.test/hook") + self.assertEqual((same["created"], same["hook_id"], self.sent("POST")), (False, 7, [])) + new = await self.call("tea_hook_create", repo="o/r", url="https://ci.test/hook", events=["push", "release"], + secret="shh") + posted = self.sent("POST", f"{R}/hooks")[0][3] + self.assertEqual((posted["events"], posted["config"]), + (["push", "release"], {"url": "https://ci.test/hook", "content_type": "json", "secret": "shh"})) + self.assertNotIn("shh", json.dumps(new)) + + async def test_repositories_are_private_unless_asked(self): + mine = await self.call("tea_repo_create", name="lab") + org = await self.call("tea_org_repo_create", org="acme", name="site", private=False) + self.assertEqual([c[3]["private"] for c in self.sent("POST")], [True, False]) + self.assertEqual((mine["full_name"], org["full_name"]), ("tea/lab", "acme/site")) + + async def test_org_add_member_needs_a_real_team(self): + self.assertIn("Owners, Writers", await self.refused("tea_org_add_member", org="acme", user="kim", team="Admins")) + added = await self.call("tea_org_add_member", org="acme", user="kim", team="writers") + self.assertEqual((self.sent("PUT")[0][1], added["team"]), ("teams/4/members/kim", "Writers")) + + async def test_repo_edit_can_archive(self): + out = await self.call("tea_repo_edit", repo="o/r", archived=True, description="old") + self.assertEqual(self.sent("PATCH", R)[0][3], {"description": "old", "archived": True}) + self.assertEqual((out["archived"], out["description"]), (True, "old")) + self.assertIn("nothing to change", await self.refused("tea_repo_edit", repo="o/r")) + + async def test_an_unreachable_gitea_is_named(self): + def flaky(request): + if request.url.path.endswith((".diff", "/contents")): + raise httpx.ConnectError("refused", request=request) + return self.gitea(request) + server._transport = httpx.MockTransport(flaky) + self.assertIn("Gitea did not answer (ConnectError)", await self.refused("tea_tree", repo="o/r")) + self.assertIn("Gitea did not answer (ConnectError)", await self.refused("tea_pull_get", repo="o/r", number=2)) + + async def test_gitea_refusals_pass_through_but_its_internals_do_not(self): + self.assertIn("Gitea refused (403): user should have a permission", await self.refused("tea_repo", repo="o/forbidden")) + message = await self.refused("tea_repo", repo="o/broken") + self.assertIn("Gitea is unavailable", message) + self.assertNotIn("secret internals", message) + + +class App(unittest.IsolatedAsyncioTestCase): + async def test_health_and_sign_in_are_wired(self): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url="http://localhost") as web: + health = await web.get("/healthz") + anonymous = await web.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, + headers={"Accept": "application/json, text/event-stream"}) + metadata = await web.get("/.well-known/oauth-authorization-server") + self.assertEqual((health.json()["version"], anonymous.status_code), (server.VERSION, 401)) + self.assertEqual(metadata.json()["issuer"].rstrip("/"), "https://mcp.test") + + +if __name__ == "__main__": + unittest.main()