MCP 2.3.0: tea_file_edit, exact-text replacements in one file (#12) (#13)

MCP 2.3.0: tea_file_edit, exact-text replacements in one file (#13)

Closes #12. A commit through the MCP needs the whole file in one call, and one model answer carries about 100 KB at most; tea_file_edit reads a file at the branch, applies exact-text replacements in order (each old text exactly once in the file as it stands), commits one update with the file's sha, and answers with the new size and sha256 to check against the caller's copy. 40 tests pass from a clean checkout on hb-5. Merged on the owner's say-so in the chat.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SKUMBXauq7XVN4BhTXjCSM

mcp/README.md

@@ -4,11 +4,11 @@ Teahub's MCP (Fly app `green-tea-mcp`, `https://mcp.teahub.xyz/mcp`) against the
 
 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 (31), all acting as you. Their parameters are typed, so a bad argument is refused before Gitea is asked anything.
+Tools (32), 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`.
+- Files: `tea_tree`, `tea_file_get`, `tea_file_change`, `tea_file_edit` (exact-text replacements in one file, for a file too long to send whole; it answers with the new file's size and sha256), `tea_commit` (up to 50 files, text or base64, 3 MB in all), `tea_branch_create`.
 - Domains: `tea_domains` (Cloudflare Registrar: exact names checked live, and suggestions, with prices).
 - Actions: `tea_actions` (recent runs), `tea_action_logs` (a run's jobs, steps and log tails), `tea_workflow_dispatch`, `tea_secret_set` (never echoed), `tea_variable_set`.
 

mcp/tea_mcp/server.py

@@ -46,7 +46,7 @@ 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.2.0"
+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"
@@ -143,6 +143,13 @@ class FileChange(BaseModel):
     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
 
@@ -595,6 +602,37 @@ async def commit(repo: str, message: str, files: list[FileChange], branch: str,
             "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]:

mcp/tests/test_server.py

@@ -25,8 +25,8 @@ READS = {"tea_whoami", "tea_issues", "tea_issue_get", "tea_issue_comments", "tea
          "tea_tree", "tea_file_get", "tea_actions", "tea_action_logs", "tea_domains"}
 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"}
+          "tea_org_add_member", "tea_hook_create", "tea_file_change", "tea_file_edit", "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"}
 
@@ -346,6 +346,46 @@ class TeaMCP(unittest.IsolatedAsyncioTestCase):
         self.assertEqual((created["files"][0]["operation"], created["branch"]), ("create", "trunk"))
         self.assertEqual((updated["files"][0]["operation"], updated["branch"]), ("update", "dev"))
 
+    async def test_file_edit_replaces_exact_text_in_order_and_says_what_it_made(self):
+        import hashlib
+        self.gitea.files["notes.md"] = "one\ntwo — deux\nthree\n".encode("utf-8")
+        out = await self.call("tea_file_edit", repo="o/r", path="/notes.md", message=" edit ", edits=[
+            {"old": "two — deux", "new": "two"}, {"old": "two\nthree", "new": "two\nthree\nfour"}, {"old": "one\n"}])
+        payload = self.sent("POST", f"{R}/contents")[0][3]
+        (op,) = payload["files"]
+        self.assertEqual((payload["message"], payload["branch"], "new_branch" in payload), ("edit", "trunk", False))
+        self.assertEqual((op["operation"], op["path"], op["sha"]), ("update", "notes.md", "sha-notes.md"))
+        made = base64.b64decode(op["content"])
+        self.assertEqual(made, b"two\nthree\nfour\n")
+        self.assertEqual((out["path"], out["edits"], out["size"], out["sha256"], out["branch"], out["commit"]),
+                         ("notes.md", 3, len(made), hashlib.sha256(made).hexdigest(), "trunk", "c0ffeec0ffee"))
+        self.assertEqual(self.sent("GET", f"{R}/contents/notes.md")[0][2], {"ref": "trunk"}, "read from the branch it edits")
+        await self.call("tea_file_edit", repo="o/r", path="README.md", message="m", branch="dev", edits=[{"old": "hello", "new": "hi"}])
+        self.assertEqual((self.sent("GET", f"{R}/contents/README.md")[-1][2], self.sent("POST", f"{R}/contents")[-1][3]["branch"]),
+                         ({"ref": "dev"}, "dev"))
+
+    async def test_file_edit_refuses_what_it_cannot_do_exactly(self):
+        self.gitea.files["notes.md"] = b"one two one\n"
+        for edits, expected in [([{"old": "three"}], "edit 1: its old text occurs 0 times in notes.md, not once"),
+                                ([{"old": "one"}], "edit 1: its old text occurs 2 times in notes.md, not once"),
+                                ([{"old": "two", "new": "one"}, {"old": "one"}], "edit 2: its old text occurs 3 times")]:
+            with self.subTest(expected=expected):
+                self.assertIn(expected, await self.refused("tea_file_edit", repo="o/r", path="notes.md", message="m", edits=edits))
+        self.assertIn("is not a file", await self.refused("tea_file_edit", repo="o/r", path="docs", message="m", edits=[{"old": "x"}]))
+        self.assertIn("is not UTF-8 text", await self.refused("tea_file_edit", repo="o/r", path="img/logo.png", message="m",
+                                                              edits=[{"old": "x"}]))
+        self.assertEqual(self.sent("POST"), [], "nothing was committed")
+        for args in [{"edits": []}, {"edits": [{"old": ""}]}, {"edits": [{"new": "x"}]}, {"edits": [{"old": "x"}], "path": "../x"}]:
+            with self.subTest(args=args):
+                await self.refused("tea_file_edit", **{"repo": "o/r", "path": "notes.md", "message": "m", **args})
+        cap, server.MAX_COMMIT_BYTES = server.MAX_COMMIT_BYTES, 10
+        try:
+            self.assertIn("more than 10 bytes", await self.refused("tea_file_edit", repo="o/r", path="notes.md", message="m",
+                                                                   edits=[{"old": "two", "new": "x" * 20}]))
+        finally:
+            server.MAX_COMMIT_BYTES = cap
+        self.assertEqual(self.sent("POST"), [])
+
     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"})