Derive the /mcp agent endpoint from the agent-tagged routes

backend/druks/api/app.py

@@ -1,13 +1,15 @@
 import logging
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
+from collections.abc import AsyncIterator, Mapping
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from pathlib import Path
 from typing import Any
 
 from fastapi import Depends, FastAPI, HTTPException, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
+from fastmcp.utilities.lifespan import combine_lifespans
 from starlette.datastructures import MutableHeaders
+from starlette.routing import Route
 
 from druks.accounts.dependencies import current_account
 from druks.accounts.routes import router as auth_router
@@ -112,7 +114,18 @@ async def _release_db_session() -> AsyncIterator[None]:
         db_session.remove()
 
 
-app = FastAPI(title="Druks", lifespan=lifespan, dependencies=[Depends(_release_db_session)])
+def _mcp_lifespan(app: FastAPI) -> AbstractAsyncContextManager[Mapping[str, Any] | None]:
+    # FastAPI never runs a plain route's lifespan; the endpoint's builds its
+    # session manager. Late-bound: the endpoint derives from the assembled
+    # app further down.
+    return mcp_app.lifespan(app)
+
+
+app = FastAPI(
+    title="Druks",
+    lifespan=combine_lifespans(lifespan, _mcp_lifespan),
+    dependencies=[Depends(_release_db_session)],
+)
 
 
 # Exception handlers — uniform JSON envelope.
@@ -201,6 +214,17 @@ async def _unhandled_exception_handler(
 app.include_router(artifacts_router, dependencies=_session_gate)
 load(app)
 
+# Tools derive from every route tagged "agent", so the endpoint composes
+# after load() — an extension's tagged routes join tools/list too.
+from druks.mcp.app import create_mcp_app  # noqa: E402
+
+# A bare Route at exactly /mcp (a Mount would 307 the no-slash path); PATs
+# authenticate it, so it sits outside the session gate.
+mcp_app = create_mcp_app(app)
+app.router.routes.append(
+    Route("/mcp", mcp_app, methods=["POST", "DELETE"], include_in_schema=False)
+)
+
 
 # Unknown /api/* paths return a JSON 404 across every method instead of
 # falling through to the SPA index.html, which would mislead API consumers
@@ -215,6 +239,17 @@ async def api_not_found(path: str) -> None:
     raise HTTPException(status_code=404, detail=f"Unknown API path: /api/{path}")
 
 
+# The edge forwards /mcp/* but the endpoint owns only the exact path; an
+# unowned subpath must not fall through to the SPA's index.html.
+@app.api_route(
+    "/mcp/{path:path}",
+    methods=["GET", "POST", "PATCH", "PUT", "DELETE"],
+    include_in_schema=False,
+)
+async def mcp_not_found(path: str) -> None:
+    raise HTTPException(status_code=404, detail=f"Unknown MCP path: /mcp/{path}")
+
+
 # Repo root in a checkout, /app in the backend image — both put the built SPA
 # at <root>/dist (vite's outDir).
 _SPA_DIST = Path(__file__).resolve().parents[3] / "dist"

backend/druks/mcp/app.py

@@ -0,0 +1,109 @@
+# The inbound /mcp endpoint ("server" stays reserved for the registry rows).
+# Its tools are derived from the routes tagged "agent": the route is an
+# operation's single declaration — schema, docstring, operation_id — and a
+# tagged extension route joins the surface the same way.
+from collections.abc import Generator
+
+import httpx
+from fastapi import FastAPI
+from fastmcp import FastMCP
+from fastmcp.server.auth import AccessToken, TokenVerifier
+from fastmcp.server.dependencies import get_http_request
+from fastmcp.server.http import StarletteWithLifespan
+from fastmcp.server.providers.openapi import MCPType, OpenAPIProvider, OpenAPITool, RouteMap
+from mcp.types import ToolAnnotations
+
+from druks.accounts.exceptions import InvalidPatError
+from druks.accounts.models import PersonalAccessToken
+from druks.database import db_session
+
+_INSTRUCTIONS = """\
+Druks coordinates durable agent runs over shared work items. This surface
+answers gates: get_gate returns a parked run's ask, a bounded artifact
+chunk, and parkedAt; answer_gate must echo that parkedAt unchanged — it
+names the exact question being answered, and a repeat answer reports
+already_answered. get_agent_call returns bounded transcript and stderr
+tails, never full payloads. cancel_run records its reason as the run's
+failure. get_usage is the caller's quota and today's spend. There is no
+push channel; poll. Tool failures embed {code, message, retryable} from the
+HTTP surface.
+"""
+
+# FastMCP logs component-fn errors instead of raising, so the tools/list
+# test is the guard: an unmapped tagged route ships unannotated and fails CI.
+_TOOL_ANNOTATIONS = {
+    "get_gate": ToolAnnotations(readOnlyHint=True),
+    "answer_gate": ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=True),
+    "get_agent_call": ToolAnnotations(readOnlyHint=True),
+    "cancel_run": ToolAnnotations(readOnlyHint=False, destructiveHint=True, idempotentHint=True),
+    "get_usage": ToolAnnotations(readOnlyHint=True),
+}
+
+
+class PatTokenVerifier(TokenVerifier):
+    async def verify_token(self, token: str) -> AccessToken | None:
+        # Auth middleware runs outside the request session boundary, so this
+        # owns one — authenticate stamps last_used_at.
+        try:
+            pat = PersonalAccessToken.authenticate(token)
+            access = AccessToken(
+                token=token,
+                client_id=pat.token_prefix,
+                scopes=[],
+                claims={"account_id": pat.account_id, "pat_id": pat.id},
+            )
+            db_session().commit()
+            return access
+        except InvalidPatError:
+            db_session().rollback()
+            return
+        finally:
+            db_session.remove()
+
+
+class CallerPat(httpx.Auth):
+    # The derivation strips authorization when replaying inbound headers;
+    # the caller's PAT re-enters here, so each route runs as that account.
+    def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
+        try:
+            incoming = get_http_request()
+        except RuntimeError:
+            incoming = None
+        bearer = incoming.headers.get("authorization") if incoming else None
+        if bearer:
+            request.headers["Authorization"] = bearer
+        yield request
+
+
+def _annotate(route: object, component: object) -> None:
+    if isinstance(component, OpenAPITool):
+        component.annotations = _TOOL_ANNOTATIONS[component.name]
+
+
+def create_mcp_app(api: FastAPI) -> StarletteWithLifespan:
+    # Built directly rather than via from_fastapi, which owns the transport:
+    # raise_app_exceptions=False makes an app crash reach the tool as the
+    # app's sanitized 500, so no masking is needed and the taxonomy travels.
+    client = httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=api, raise_app_exceptions=False),
+        base_url="http://druks",
+        auth=CallerPat(),
+    )
+    provider = OpenAPIProvider(
+        openapi_spec=api.openapi(),
+        client=client,
+        route_maps=[
+            RouteMap(tags={"agent"}, mcp_type=MCPType.TOOL),
+            RouteMap(mcp_type=MCPType.EXCLUDE),
+        ],
+        mcp_component_fn=_annotate,
+    )
+    server = FastMCP(
+        name="druks",
+        providers=[provider],
+        instructions=_INSTRUCTIONS,
+        auth=PatTokenVerifier(),
+    )
+    # Derivation primed app.openapi()'s cache mid-assembly; drop it.
+    api.openapi_schema = None
+    return server.http_app(path="/mcp", stateless_http=True, json_response=False)

backend/druks/mcp/gateway/routes.py

@@ -6,7 +6,9 @@
 from druks.accounts.models import Account
 from druks.mcp.gateway import schemas, services
 
-router = APIRouter(prefix="/mcp", tags=["mcp"])
+# Docstrings here are the derived tool descriptions and operation_id is the
+# tool name — renaming one is a break, never a refactor side effect.
+router = APIRouter(prefix="/api", tags=["agent"])
 
 
 @router.get(
@@ -16,6 +18,8 @@
     response_model_by_alias=True,
 )
 async def get_gate(run_id: str) -> schemas.GateResponse:
+    """A parked run's open gate: the ask, a bounded artifact chunk, and
+    parkedAt — echo parkedAt unchanged to answer_gate."""
     return services.get_gate(run_id)
 
 
@@ -26,6 +30,9 @@ async def get_gate(run_id: str) -> schemas.GateResponse:
     response_model_by_alias=True,
 )
 async def answer_gate(run_id: str, body: schemas.AnswerGateRequest) -> schemas.GateAnswerResponse:
+    """Answer the gate get_gate showed, resuming the run. parkedAt must echo
+    get_gate's value unchanged; a repeat answer to the same parkedAt reports
+    already_answered."""
     return await services.answer_gate(
         run_id,
         parked_at=body.parked_at,
@@ -42,6 +49,8 @@ async def answer_gate(run_id: str, body: schemas.AnswerGateRequest) -> schemas.G
     response_model_by_alias=True,
 )
 async def get_agent_call(call_id: str) -> schemas.AgentCallDetailResponse:
+    """One agent call's metadata with bounded transcript and stderr tails and
+    an artifact chunk."""
     return services.get_agent_call(call_id)
 
 
@@ -54,14 +63,18 @@ async def get_agent_call(call_id: str) -> schemas.AgentCallDetailResponse:
 async def cancel_run(
     run_id: str, reason: Annotated[str, Body(embed=True, min_length=1, max_length=500)]
 ) -> schemas.CancelRunResponse:
+    """Cancel an active run, recording the reason as its failure; a repeat
+    cancel reports already_cancelled."""
     return await services.cancel_run(run_id, reason=reason)
 
 
 @router.get(
-    "/usage",
+    "/usage/summary",
     operation_id="get_usage",
     response_model=schemas.AgentUsageResponse,
     response_model_by_alias=True,
 )
 async def get_usage(account: Account = Depends(current_account)) -> schemas.AgentUsageResponse:
+    """The caller's harness quota snapshot and today's spend. Pure read — it
+    never triggers a scrape."""
     return services.get_usage(account)

backend/tests/test_agent_routes.py

@@ -6,10 +6,12 @@
     configure_app_for_test,
     make_settings,
     make_test_work_item,
+    seed_agent_run,
     seed_build_run,
     seed_run,
 )
 from druks.accounts.models import Account
+from druks.api.app import app
 from druks.durable.models import Run
 from druks.durable.reads import read_transcript_chunk
 from druks.mcp.gateway import services
@@ -22,11 +24,11 @@
 }
 
 _MCP_ROUTES = {
-    ("get", "/mcp/gates/{run_id}"): "get_gate",
-    ("post", "/mcp/gates/{run_id}/answer"): "answer_gate",
-    ("get", "/mcp/agent-calls/{call_id}"): "get_agent_call",
-    ("post", "/mcp/runs/{run_id}/cancel"): "cancel_run",
-    ("get", "/mcp/usage"): "get_usage",
+    ("get", "/api/gates/{run_id}"): "get_gate",
+    ("post", "/api/gates/{run_id}/answer"): "answer_gate",
+    ("get", "/api/agent-calls/{call_id}"): "get_agent_call",
+    ("post", "/api/runs/{run_id}/cancel"): "cancel_run",
+    ("get", "/api/usage/summary"): "get_usage",
 }
 
 
@@ -69,27 +71,25 @@ def _park(db_session, item_id):
 
 
 def test_openapi_pins_the_five_agent_routes(client: TestClient):
-    from druks.api.app import app
-
     schema = app.openapi()
     found = {
         (method, path): operation
         for path, operations in schema["paths"].items()
         for method, operation in operations.items()
-        if operation.get("tags") == ["mcp"]
+        if operation.get("tags") == ["agent"]
     }
     assert {key: op["operationId"] for key, op in found.items()} == _MCP_ROUTES
 
 
 def test_agent_routes_sit_behind_the_gate(tmp_path, db_session):
     app = configure_app_for_test(settings=make_settings(tmp_path), authenticated=False)
     with TestClient(app) as anonymous:
-        assert anonymous.get("/mcp/gates/x").status_code == 401
-        assert anonymous.get("/mcp/usage").status_code == 401
+        assert anonymous.get("/api/gates/x").status_code == 401
+        assert anonymous.get("/api/usage/summary").status_code == 401
 
 
 def test_agent_errors_share_one_shape(client: TestClient, db_session):
-    missing = client.get("/mcp/gates/no-such-run")
+    missing = client.get("/api/gates/no-such-run")
     assert missing.status_code == 404
     assert missing.json() == {
         "code": "RUN_NOT_FOUND",
@@ -100,7 +100,7 @@ def test_agent_errors_share_one_shape(client: TestClient, db_session):
     item = make_test_work_item(repo="o/r", title="t")
     run = _park(db_session, item.id)
     stale = client.post(
-        f"/mcp/gates/{run.id}/answer",
+        f"/api/gates/{run.id}/answer",
         json={"parkedAt": "2020-01-01T00:00:00+00:00", "control": "approve"},
     )
     assert stale.status_code == 409
@@ -113,13 +113,13 @@ def test_get_gate_then_answer_roundtrip(client: TestClient, db_session, resume_s
     item = make_test_work_item(repo="o/r", title="t")
     run = _park(db_session, item.id)
 
-    view = client.get(f"/mcp/gates/{run.id}")
+    view = client.get(f"/api/gates/{run.id}")
     assert view.status_code == 200
     data = view.json()
     assert data == services.get_gate(run.id).model_dump(mode="json", by_alias=True)
 
     answered = client.post(
-        f"/mcp/gates/{run.id}/answer",
+        f"/api/gates/{run.id}/answer",
         json={"parkedAt": data["parkedAt"], "control": "approve", "note": "ship it"},
     )
     assert answered.status_code == 200
@@ -138,7 +138,7 @@ def test_answer_gate_reads_already_answered_off_the_receipt(
     db_session.flush()
 
     response = client.post(
-        f"/mcp/gates/{run.id}/answer",
+        f"/api/gates/{run.id}/answer",
         json={"parkedAt": parked_at.isoformat(), "control": "approve"},
     )
 
@@ -152,7 +152,7 @@ def test_answer_gate_requires_an_aware_parked_at(client: TestClient, db_session)
     run = _park(db_session, item.id)
 
     naive = client.post(
-        f"/mcp/gates/{run.id}/answer",
+        f"/api/gates/{run.id}/answer",
         json={"parkedAt": "2026-07-19T10:00:00", "control": "approve"},
     )
 
@@ -168,12 +168,12 @@ def test_cancel_run_route(client: TestClient, db_session):
     run.input_requested_at = run.utc_now()
     db_session.flush()
 
-    unbounded = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": "r" * 501})
+    unbounded = client.post(f"/api/runs/{run.id}/cancel", json={"reason": "r" * 501})
     assert unbounded.status_code == 422
-    blank = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": ""})
+    blank = client.post(f"/api/runs/{run.id}/cancel", json={"reason": ""})
     assert blank.status_code == 422
 
-    cancelled = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": "wrong branch"})
+    cancelled = client.post(f"/api/runs/{run.id}/cancel", json={"reason": "wrong branch"})
     assert cancelled.status_code == 200
     assert cancelled.json() == {"runId": run.id, "result": "cancelled"}
 
@@ -183,14 +183,12 @@ def test_cancel_run_route(client: TestClient, db_session):
     assert not run.input_gate
     assert run.failure == "wrong branch"
 
-    again = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": "wrong branch"})
+    again = client.post(f"/api/runs/{run.id}/cancel", json={"reason": "wrong branch"})
     assert again.status_code == 200
     assert again.json()["result"] == "already_cancelled"
 
 
 def test_transcript_route_matches_the_read_machinery(client: TestClient, db_session, db_engine):
-    from conftest import seed_agent_run
-
     call = seed_agent_run()
     call_dir = call.call_dir
     call_dir.mkdir(parents=True, exist_ok=True)
@@ -202,8 +200,8 @@ def test_transcript_route_matches_the_read_machinery(client: TestClient, db_sess
     assert response.status_code == 200
     chunk = read_transcript_chunk(db_engine, call.id, "stdout", offset=0, limit=7)
     assert response.json() == chunk.model_dump(mode="json", by_alias=True)
-    # The 7-byte cut lands mid-é; the served chunk ends on a character boundary.
-    assert response.json()["text"] == "hello "
+    # The 7-byte cut lands mid-é; the window serves the seam's one �.
+    assert response.json()["text"] == "hello �"
 
 
 def test_resume_route_contract_is_preserved(client: TestClient, db_session, resume_spy):
@@ -257,7 +255,7 @@ def test_usage_agent_route_matches_the_service(client: TestClient, db_session, a
     )
     db_session.flush()
 
-    response = client.get("/mcp/usage")
+    response = client.get("/api/usage/summary")
     assert response.status_code == 200
     body = response.json()
     assert body == services.get_usage(account).model_dump(mode="json", by_alias=True)