teahubxyzandClaude Fable 5.1
fcd70f294d
For a file too long to send whole: the edits apply in order, each old text must occur exactly once in the file as it stands, and the answer gives the new file's size and sha256 to check against the caller's copy. Version 2.3.0. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SKUMBXauq7XVN4BhTXjCSM
737 lines
39 KiB
Python
737 lines
39 KiB
Python
"""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 (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. 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 hashlib
|
|
import os
|
|
from typing import Annotated, Any, Literal
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
import uvicorn
|
|
from fastmcp import FastMCP
|
|
from fastmcp.exceptions import ToolError
|
|
from fastmcp.server.auth import cimd as fastmcp_cimd
|
|
from fastmcp.server.auth.oauth_proxy import OAuthProxy
|
|
from fastmcp.server.auth.oauth_proxy.models import ProxyDCRClient
|
|
from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier
|
|
from fastmcp.server.dependencies import get_access_token
|
|
from pydantic import BaseModel, Field
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
from starlette.routing import Route
|
|
|
|
from .domains import Refused, domains_http, look_up
|
|
|
|
|
|
def required_env(name: str) -> str:
|
|
value = os.getenv(name)
|
|
if not value:
|
|
raise RuntimeError(f"Missing required environment variable: {name}")
|
|
return value
|
|
|
|
|
|
GITEA_URL = required_env("GITEA_URL").rstrip("/")
|
|
PUBLIC_URL = required_env("PUBLIC_URL").rstrip("/")
|
|
GITEA_OAUTH_CLIENT_ID = required_env("GITEA_OAUTH_CLIENT_ID")
|
|
GITEA_OAUTH_CLIENT_SECRET = required_env("GITEA_OAUTH_CLIENT_SECRET")
|
|
JWT_SIGNING_KEY = required_env("JWT_SIGNING_KEY")
|
|
ALLOWED_HOSTS = [h.strip() for h in os.getenv("ALLOWED_HOSTS", "mcp.teahub.xyz,localhost").split(",") if h.strip()]
|
|
|
|
VERSION = "2.3.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 # 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)
|
|
LOG_BYTES = 60_000 # all the job logs of one tea_action_logs
|
|
READ = {"readOnlyHint": True}
|
|
WRITE = {"destructiveHint": False}
|
|
AGAIN = {**WRITE, "idempotentHint": True} # the same call twice changes nothing more
|
|
|
|
# ---------- 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
|
|
|
|
|
|
async def _cimd_fetch(url: str, *, require_path: bool = False, max_size: int = 5120, timeout: float = 10.0,
|
|
overall_timeout: float = 30.0, request_headers=None, allowed_status_codes=None):
|
|
headers = {"User-Agent": "Green-Tea-MCP/1.0 FastMCP-CIMD", **dict(request_headers or {})}
|
|
return await _CIMD_SAFE_FETCH(url, require_path=require_path, max_size=max_size, timeout=timeout,
|
|
overall_timeout=overall_timeout, request_headers=headers,
|
|
allowed_status_codes=allowed_status_codes)
|
|
|
|
|
|
fastmcp_cimd.ssrf_safe_fetch_response = _cimd_fetch
|
|
|
|
|
|
class TeaOAuthProxy(OAuthProxy):
|
|
async def get_client(self, client_id: str):
|
|
if client_id == CLAUDE_CIMD_CLIENT_ID:
|
|
return ProxyDCRClient(client_id=CLAUDE_CIMD_CLIENT_ID, client_name="Claude", client_uri="https://claude.ai",
|
|
redirect_uris=[CLAUDE_REDIRECT_URI], response_types=["code"],
|
|
grant_types=["authorization_code", "refresh_token"], scope="openid profile",
|
|
token_endpoint_auth_method="none")
|
|
return await super().get_client(client_id)
|
|
|
|
|
|
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=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,
|
|
)
|
|
|
|
mcp = FastMCP(
|
|
"Teahub",
|
|
version=VERSION,
|
|
website_url=GITEA_URL,
|
|
instructions=("Teahub tools against " + GITEA_URL + ". Every tool acts as you, with your own Gitea permissions: "
|
|
"issues, pull requests and reviews, labels, repositories, files and commits, branches, Actions "
|
|
"(runs, logs, dispatch, secrets, variables), hooks and organisations, and domain search with "
|
|
"Cloudflare Registrar. Reads and writes are bounded and no tool deletes. Repositories are named "
|
|
"owner/repo."),
|
|
auth=auth,
|
|
mask_error_details=True,
|
|
)
|
|
|
|
# ---------- 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 Edit(BaseModel):
|
|
"""One replacement in a text file: `old` occurs exactly once in the file as it stands (after the edits before it),
|
|
and `new` takes its place (empty removes it)."""
|
|
old: Annotated[str, Field(min_length=1, max_length=MAX_BYTES)]
|
|
new: Annotated[str, Field(max_length=MAX_BYTES)] = ""
|
|
|
|
|
|
class InputError(ToolError):
|
|
pass
|
|
|
|
|
|
# ---------- Gitea, as the caller ----------
|
|
|
|
_transport: httpx.AsyncBaseTransport | None = None # tests put a fake Gitea here
|
|
|
|
|
|
def actor_token() -> str:
|
|
access = get_access_token()
|
|
if access is None or not access.token:
|
|
raise ToolError("not signed in")
|
|
return access.token
|
|
|
|
|
|
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 = ""
|
|
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, tail: bool = False, missing_ok: bool = False) -> tuple[str, bool]:
|
|
"""A text answer (a diff, a log): its first `cap` bytes, or its last with `tail`, and whether any were cut."""
|
|
buf, cut = b"", False
|
|
try:
|
|
async with gitea_client() as client, client.stream("GET", path) as r:
|
|
if missing_ok and r.status_code == 404:
|
|
return "", False
|
|
if r.status_code >= 400:
|
|
await r.aread()
|
|
raise refused(r)
|
|
async for chunk in r.aiter_bytes():
|
|
buf += chunk
|
|
if len(buf) > cap:
|
|
cut, buf = True, buf[-cap:] if tail else buf[:cap]
|
|
if not tail:
|
|
break
|
|
except httpx.TransportError as e:
|
|
raise unreachable(e) from None
|
|
return buf.decode("utf-8", "replace"), cut
|
|
|
|
|
|
def repo_path(repo: str) -> str:
|
|
if repo.split("/")[1] in (".", ".."):
|
|
raise InputError("repo must be owner/repo")
|
|
return f"repos/{repo}"
|
|
|
|
|
|
def file_path(path: str) -> str:
|
|
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 "/".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:
|
|
if len(text.encode("utf-8")) > MAX_BYTES:
|
|
raise InputError(f"{what} exceeds the {MAX_BYTES}-byte limit")
|
|
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") or x.get("head"))}
|
|
|
|
|
|
# ---------- you ----------
|
|
|
|
|
|
@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, 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], "truncated": more}
|
|
|
|
|
|
# ---------- issues, labels and pull requests ----------
|
|
|
|
|
|
@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 | Annotated[str, Field(max_length=1000)] | None = None, page: Page = 1,
|
|
limit: Limit = 20) -> dict[str, Any]:
|
|
if isinstance(labels, str): # the comma-separated form that clients holding 1.0's schema still send
|
|
labels = [name.strip() for name in labels.split(",") if name.strip()]
|
|
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 {"repo": repo, **issue_row(x), **clip(x.get("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:
|
|
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 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")
|
|
return {"updated": True, **issue_row(await gitea("PATCH", f"{repo_path(repo)}/issues/{number}", json=patch))}
|
|
|
|
|
|
@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": bounded(body.strip(), "body")})
|
|
return {"repo": repo, "issue": number, "comment_id": c.get("id"), "url": c.get("html_url")}
|
|
|
|
|
|
@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", cap=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"), "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="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}}
|
|
|
|
|
|
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 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:
|
|
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.get("name"), "user": user, "added": True}
|
|
|
|
|
|
@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)
|
|
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:
|
|
return {**meta, "content": base64.b64decode(x.get("content") or "").decode("utf-8")}
|
|
except (binascii.Error, ValueError):
|
|
return {**meta, "binary": True}
|
|
|
|
|
|
@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="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)
|
|
|
|
|
|
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="Edit one text file on a branch by exact-text replacements, as a commit by you: for a file too "
|
|
"long to send whole (tea_file_change takes 200 KB). The edits apply in order; each `old` must "
|
|
"occur exactly once in the file as it stands, and `new` takes its place. The answer gives the new "
|
|
"file's size and sha256, to check against your copy. Branch protection applies.",
|
|
annotations=WRITE)
|
|
async def tea_file_edit(repo: Repo, path: FilePath, edits: Annotated[list[Edit], Field(min_length=1, max_length=50)],
|
|
message: Message, branch: Ref = "") -> dict[str, Any]:
|
|
rp, p = repo_path(repo), file_path(path)
|
|
branch = branch or await default_branch(rp)
|
|
x = await gitea("GET", contents(rp, p), params={"ref": branch})
|
|
if not isinstance(x, dict) or x.get("type") != "file":
|
|
raise InputError(f"{p} is not a file")
|
|
try:
|
|
text = base64.b64decode(x.get("content") or "").decode("utf-8")
|
|
except (binascii.Error, ValueError):
|
|
raise InputError(f"{p} is not UTF-8 text") from None
|
|
for n, e in enumerate(edits, 1):
|
|
found = text.count(e.old)
|
|
if found != 1:
|
|
raise InputError(f"edit {n}: its old text occurs {found} times in {p}, not once")
|
|
text = text.replace(e.old, e.new, 1)
|
|
data = text.encode("utf-8")
|
|
if len(data) > MAX_COMMIT_BYTES:
|
|
raise InputError(f"{p} would come to more than {MAX_COMMIT_BYTES} bytes")
|
|
payload = {"files": [{"operation": "update", "path": p, "sha": x.get("sha"),
|
|
"content": base64.b64encode(data).decode()}], "message": message.strip(), "branch": branch}
|
|
c = (await gitea("POST", f"{rp}/contents", json=payload)).get("commit") or {}
|
|
return {"repo": repo, "branch": branch, "path": p, "edits": len(edits), "size": len(data),
|
|
"sha256": hashlib.sha256(data).hexdigest(), "commit": (c.get("sha") or "")[:12], "url": c.get("html_url")}
|
|
|
|
|
|
@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 runs of a repository, newest first. tea_action_logs reads a run's jobs and 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/runs", params={"page": page, "limit": limit}) or {}
|
|
runs = []
|
|
for r in x.get("workflow_runs") or []:
|
|
workflow, _, ref = (r.get("path") or "").partition("@")
|
|
runs.append({"id": r.get("id"), "number": r.get("run_number"), "title": r.get("display_title"),
|
|
"workflow": workflow, "branch": r.get("head_branch") or ref, "event": r.get("event"),
|
|
"status": r.get("status"), "conclusion": r.get("conclusion"),
|
|
"sha": (r.get("head_sha") or "")[:12], "started_at": r.get("started_at"),
|
|
"completed_at": r.get("completed_at"), "url": r.get("html_url")})
|
|
return {"repo": repo, "page": page, "more": page * limit < (x.get("total_count") or 0), "runs": runs}
|
|
|
|
|
|
@mcp.tool(description="The jobs of one Actions run (its id from tea_actions), their steps, and the last lines of each "
|
|
"job's log.", annotations=READ)
|
|
async def tea_action_logs(repo: Repo, run: Annotated[int, Field(ge=1, description="run id, from tea_actions")],
|
|
tail: Annotated[int, Field(ge=10, le=400)] = 120) -> dict[str, Any]:
|
|
rp = repo_path(repo)
|
|
jobs = (await gitea("GET", f"{rp}/actions/runs/{run}/jobs", params={"limit": 50}) or {}).get("jobs") or []
|
|
out = []
|
|
for j in jobs:
|
|
log, cut = await gitea_text(f"{rp}/actions/jobs/{j.get('id')}/logs", cap=LOG_BYTES // len(jobs), tail=True,
|
|
missing_ok=True)
|
|
lines = log.splitlines()[1 if cut else 0:] # a cut log starts mid-line
|
|
out.append({"id": j.get("id"), "name": j.get("name"), "status": j.get("status"),
|
|
"conclusion": j.get("conclusion"),
|
|
"steps": [{"name": s.get("name"), "conclusion": s.get("conclusion")} for s in j.get("steps") or []],
|
|
"log_tail": "\n".join(lines[-tail:])})
|
|
return {"repo": repo, "run": run, "jobs": out}
|
|
|
|
|
|
@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}
|
|
|
|
|
|
# ---------- domains ----------
|
|
|
|
|
|
@mcp.tool(description="Look up domain names with Cloudflare Registrar: a domain as given, or a name or phrase under "
|
|
"common endings (com, net, org, co, io, xyz, app, dev, ai, me), checked live, plus suggestions, "
|
|
"each with availability and Cloudflare's at-cost price. Nothing is registered or bought.",
|
|
annotations=READ)
|
|
async def tea_domains(search: Annotated[str, Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9 ._-]{0,99}$",
|
|
description="a name, a phrase or a domain")],
|
|
limit: Limit = 24) -> dict[str, Any]:
|
|
who = "mcp:" + hashlib.sha256(actor_token().encode()).hexdigest()[:16]
|
|
try:
|
|
return await look_up(search, limit, who)
|
|
except Refused as no:
|
|
raise ToolError(str(no)) from None
|
|
|
|
|
|
# ---------- app ----------
|
|
|
|
|
|
async def healthz(_: Request) -> JSONResponse:
|
|
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,
|
|
allowed_hosts=ALLOWED_HOSTS)
|
|
app.routes.append(Route("/healthz", healthz, methods=["GET"]))
|
|
app.routes.append(Route("/domains", domains_http, methods=["GET", "OPTIONS"])) # the dashboard's Domains tab
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")), proxy_headers=True, forwarded_allow_ips="*")
|