Tea MCP bus: POST /bus for Gitea-token holders (Time's scripts)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FAqjv4PBaoWnL62rDKmBJ1

mcp/tea_mcp/bus.py

@@ -146,6 +146,65 @@ class Bus:
                     "warning": NOTE_WARNING}
 
 
+def http_handler(bus: Bus, verify: Callable[[str], Awaitable[dict[str, Any]]]):
+    """POST /bus for clients that hold a Gitea token rather than an MCP sign-in (Time's scripts in Actions). The token is
+    the caller's own (`Authorization: token <t>` or `Bearer <t>`); `verify(token)` answers the same {login, allowed} as
+    the MCP tools' caller. The body is JSON: {"action": "checkin"|"list"|"read"|"message"|"inbox", ...the tool's
+    arguments}. Arguments are checked as strictly as the tools check theirs."""
+    from starlette.responses import JSONResponse
+
+    def text(body: dict[str, Any], key: str, most: int, pattern: re.Pattern[str] | None = None, need: bool = True) -> str:
+        value = body.get(key, "")
+        if not isinstance(value, str) or (need and not value.strip()) or len(value) > most \
+                or (pattern and not pattern.fullmatch(value)):
+            raise ValueError(f"bad {key}")
+        return value
+
+    async def handle(request: Any) -> Any:
+        auth = request.headers.get("authorization", "")
+        scheme, _, token = auth.partition(" ")
+        if scheme.lower() not in ("token", "bearer") or not token.strip():
+            return JSONResponse({"error": "send your Gitea token as Authorization: token <t>"}, 401)
+        try:
+            body = await request.json()
+            if not isinstance(body, dict):
+                raise ValueError("body must be a JSON object")
+        except ValueError as e:
+            return JSONResponse({"error": str(e) or "body must be JSON"}, 400)
+        try:
+            me = await verify(token.strip())
+        except ToolError as e:
+            return JSONResponse({"error": str(e)}, 401)
+        if not me.get("allowed"):
+            return JSONResponse({"error": "the office bus is for Teahub's own agents"}, 403)
+        login, action = me["login"], body.get("action")
+        try:
+            if action == "checkin":
+                out = await bus.checkin(login, text(body, "name", 40, NAME), text(body, "status", STATUS_CHARS),
+                                        text(body, "role", ROLE_CHARS, need=False))
+            elif action == "list":
+                out = {"agents": await bus.roster()}
+            elif action == "read":
+                out = await bus.read(text(body, "name", 40, NAME))
+            elif action == "message":
+                out = await bus.message(login, text(body, "sender", 40, NAME), text(body, "to", 40, NAME),
+                                        text(body, "text", NOTE_CHARS))
+            elif action == "inbox":
+                after = body.get("after", 0)
+                if not isinstance(after, int) or isinstance(after, bool) or after < 0:
+                    raise ValueError("bad after")
+                out = await bus.inbox(login, text(body, "name", 40, NAME), after)
+            else:
+                raise ValueError("action is one of checkin, list, read, message, inbox")
+        except ValueError as e:
+            return JSONResponse({"error": str(e)}, 400)
+        except ToolError as e:
+            return JSONResponse({"error": str(e)}, 409)
+        return JSONResponse(out)
+
+    return handle
+
+
 def register(mcp: Any, caller: Callable[[], Awaitable[dict[str, Any]]], bus: Bus | None = None) -> Bus:
     """Add the bus tools to `mcp`. `caller()` answers the signed-in caller's Gitea login and whether they may use the
     bus (admin, or a member of BUS_ORG); it is passed in so this module never imports the server."""

mcp/tests/test_bus.py

@@ -1,12 +1,13 @@
 """The office bus: ownership, notes, limits and expiry, with a fake clock and a fake MCP."""
 import asyncio
+import json
 import os
 import tempfile
 import unittest
 
 from fastmcp.exceptions import ToolError
 
-from tea_mcp.bus import MAX_HISTORY, MAX_INBOX, TTL, Bus, register
+from tea_mcp.bus import MAX_HISTORY, MAX_INBOX, TTL, Bus, http_handler, register
 
 
 class Clock:
@@ -102,5 +103,59 @@ class BusTest(unittest.TestCase):
         self.assertEqual(run(mcp.tools["tea_bus_list"]())["agents"][0]["login"], "stranger")
 
 
+class Req:
+    def __init__(self, body, auth="token t"):
+        self.headers, self._body = {"authorization": auth} if auth else {}, body
+
+    async def json(self):
+        if isinstance(self._body, Exception):
+            raise self._body
+        return self._body
+
+
+class HttpTest(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.TemporaryDirectory()
+        self.bus = Bus(os.path.join(self.dir.name, "bus.json"), Clock())
+
+        async def verify(token):
+            if token == "bad":
+                raise ToolError("not a sign-in")
+            return {"login": "time", "allowed": token != "outsider"}
+
+        self.handle = http_handler(self.bus, verify)
+
+    def tearDown(self):
+        self.dir.cleanup()
+
+    def post(self, body, auth="token t"):
+        r = run(self.handle(Req(body, auth)))
+        return r.status_code, json.loads(r.body)
+
+    def test_a_token_holder_can_use_every_action(self):
+        self.assertEqual(self.post({"action": "checkin", "name": "heartbeat", "status": "tick", "role": "clock"})[0], 200)
+        self.assertEqual(self.post({"action": "checkin", "name": "astra", "status": "planning"})[0], 200)
+        self.assertEqual(self.post({"action": "message", "sender": "astra", "to": "heartbeat", "text": "#5 queued"})[0], 200)
+        code, box = self.post({"action": "inbox", "name": "heartbeat"})
+        self.assertEqual((code, [n["text"] for n in box["notes"]]), (200, ["#5 queued"]))
+        code, out = self.post({"action": "list"})
+        self.assertEqual((code, {a["name"] for a in out["agents"]}), (200, {"heartbeat", "astra"}))
+        self.assertEqual(self.post({"action": "read", "name": "astra"})[1]["statuses"][0]["status"], "planning")
+
+    def test_refusals(self):
+        self.assertEqual(self.post({"action": "list"}, auth="")[0], 401)
+        self.assertEqual(self.post({"action": "list"}, auth="Basic x")[0], 401)
+        self.assertEqual(self.post({"action": "list"}, auth="token bad")[0], 401)
+        self.assertEqual(self.post({"action": "list"}, auth="token outsider")[0], 403)
+        self.assertEqual(self.post(ValueError("no json"))[0], 400)
+        self.assertEqual(self.post(["not", "a", "dict"])[0], 400)
+        for body in [{"action": "nope"}, {"action": "checkin", "name": "Bad Name", "status": "x"},
+                     {"action": "checkin", "name": "ok", "status": ""}, {"action": "checkin", "name": "ok", "status": "x" * 501},
+                     {"action": "inbox", "name": "ok", "after": -1}, {"action": "inbox", "name": "ok", "after": True}]:
+            with self.subTest(body=body):
+                self.assertEqual(self.post(body)[0], 400)
+        self.assertEqual(self.post({"action": "read", "name": "ghost"})[0], 409)
+
+
 if __name__ == "__main__":
     unittest.main()