mcp: Teahub MCP 2.0, typed and with the tools agents kept asking for
One file, 29 tools, every one acting with the caller's own Gitea token. Parameters are typed (patterns, ranges, enums), so FastMCP refuses a bad argument before Gitea is asked anything and publishes the limits in each tool's schema. New: tea_pull_get/create/review/merge, tea_labels and tea_label, tea_commit (up to 50 files, text or base64), tea_branch_create, tea_repo_edit (including archived), tea_hook_create (one per URL), tea_secret_set and tea_variable_set. Fixed: - unknown labels are refused instead of silently dropped - a file update is chosen from a 404 on the path, not from matching "404" in an error message; paths with spaces are percent-encoded - tea_org_add_member names its team; there is no default - org repositories are private by default, like user repositories - "more" and "truncated" come from X-Total-Count, so they are true - tea_issue_comments pages client-side: Gitea's endpoint is unpaged - tea_actions reads /actions/tasks: 1.24 has no /actions/runs, so the old tool always got a 404 - a Gitea that does not answer is named as such instead of a masked error Tests run against a fake Gitea (27). The same code was run against teahub.xyz with a temporary token in scratch repositories that were then deleted. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CYHXFvWMCs1GUmRmpySqQ6
This commit is contained in:
teahubxyzandClaude Opus 5.5 committed
1 parent
369bfddaa2
commit
fb8acb0af9
3 files changed
+900
-229
No files matched your search
+12
-3
@@ -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`).
|
||||
+465
-226
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user