src/genro_asgi/applications/mcp.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """MCP applications: the stateless Streamable HTTP transport over ``McpEngine``.16 17 Two ready-to-mount apps expose a genro-routes router as MCP tools over JSON-RPC18 2.0. Both delegate the HTTP shell to a :class:`McpTransport` — the helper that19 owns everything the transport-agnostic :class:`McpEngine` does not:20 method/header/Origin gating, the JSON-RPC envelope, the 202-for-notifications21 rule, and the sync/async invoke callback (async handlers stay on the loop, sync22 handlers go through the server pool via ``run_sync`` — the Macro 1 protocol,23 replacing the old ``smartasync``). The transport holds its owning application as24 ``self.application`` (dual-parent) and reaches its ``spread_over_params`` and25 ``server`` through it.26 27 - :class:`McpApplication` — the whole app is one MCP endpoint. It holds an28 engine over an EXTERNAL router (``routing_class=`` or ``module=``); every29 request is a JSON-RPC message. Without a router ``initialize`` still answers30 and ``tools/list`` is empty.31 - :class:`McpOpenApiApplication` — one router, two faces: it inherits the whole32 OpenApiApplication machinery (``_meta`` docs, pydantic plug, REST dispatch)33 and adds an MCP face under ``mcp_name_segment`` (default ``"mcp"``). The34 ``channel`` plugin drives per-face visibility: a method is an MCP tool only on35 channel ``"mcp"`` (``@route(channel_channels="mcp,rest")`` for a dual method);36 undeclared methods default to REST-only (``channels=rest_channel``).37 38 Transport conformance (MCP Streamable HTTP): a JSON-RPC POST answers with a39 JSON response; a notification (no ``id``) answers HTTP 202 with an empty body; a40 GET opens the SSE push stream (below); any other method answers 405; an41 ``MCP-Protocol-Version`` header that is present but unsupported answers 400 (an42 absent header is assumed ``2025-03-26`` per the spec's backwards-compat rule);43 an ``Origin`` header present and not in ``allowed_origins`` answers 403 (the44 ``allowed_origins`` option defaults to ``None`` — no restriction, a dev-mode45 default: production fronting owns the Origin gate). The transport gates raise46 core HTTP exceptions answered by the server's ``ErrorMiddleware``.47 48 The push half (core 1e, the option-B commitment honored): a GET opens a49 ``text/event-stream`` keyed by ``Mcp-Session-Id`` — echoed when the client50 supplies one, minted otherwise (``secrets.token_urlsafe``, decoupled from the51 cookie session: MCP clients carry no cookie) — and follows the server's task52 hub live (``server.tasks.hub``, the A<->C bridge). A ``Last-Event-ID`` header53 replays the session's current ``progress.json`` snapshots first54 (snapshot-baseline resumability — no durable event log, ratified). A server55 composed without tasks answers GET with 405, the 1c stateless behavior. The56 engine stays untouched apart from advertising the capability.57 """58 59 from __future__ import annotations60 61 import asyncio62 import secrets63 from collections.abc import AsyncIterator64 from typing import TYPE_CHECKING, Any, ClassVar65 66 from ..exceptions import HTTPBadRequest, HTTPException, HTTPForbidden67 from ..mcp import JSONRPC_INTERNAL_ERROR, McpEngine, McpError68 from ..request import Request69 from ..routed_application import RoutedApplication70 from ..sse import SseStream71 from .openapi import OpenApiApplication72 73 if TYPE_CHECKING:74 from genro_routes import Router, RouterNode, RoutingClass75 76 from ..types import Receive, Scope, Send77 78 __all__ = ["McpApplication", "McpOpenApiApplication"]79 80 81 class McpTransport:82 """Stateless MCP Streamable HTTP transport bound to a ``RoutedApplication``.83 84 Owns the HTTP shell (method/header/Origin gating, JSON-RPC envelope,85 202-for-notifications) and the sync/async invoke callback, driving an86 :class:`McpEngine` built over a router the host application supplies. The87 host is held as ``self.application`` (dual-parent): the transport reads its88 ``spread_over_params`` and ``server`` through it.89 """90 91 def __init__(92 self,93 application: RoutedApplication,94 *,95 name: str,96 version: str,97 tool_separator: str,98 channel: str,99 allowed_origins: list[str] | None,100 ) -> None:101 self.application = application102 self.channel = channel103 self.allowed_origins = allowed_origins104 self._name = name105 self._version = version106 self._tool_separator = tool_separator107 self._engine: McpEngine | None = None108 109 @property110 def engine(self) -> McpEngine | None:111 """The MCP engine driving the tool surface (``None`` until built)."""112 return self._engine113 114 def build_engine(self, router: Router | None) -> None:115 """Build the engine over ``router``, ensuring pydantic and auth.116 117 The pydantic plugin caches the neutral ``params``/``result`` blocks the118 engine reads for ``inputSchema``/``outputSchema``; the auth plugin119 enforces any ``auth_rule`` the router's handlers declare — without it120 a ruled tool would be listed and callable by anonymous clients. Each121 is plugged only if absent (plugging is not idempotent).122 """123 if router is not None:124 self.plug_if_absent(router, "pydantic")125 self.plug_if_absent(router, "auth")126 self._engine = McpEngine(127 router,128 name=self._name,129 version=self._version,130 tool_separator=self._tool_separator,131 channel=self.channel,132 invoke=self._invoke_tool,133 )134 135 def plug_if_absent(self, router: Router, name: str, **options: Any) -> None:136 """Plug ``name`` on ``router`` unless already attached (guards double-plug)."""137 if name not in {plugin.name for plugin in router.iter_plugins()}:138 router.plug(name, **options)139 140 def _invoke_tool(self, node: RouterNode, arguments: dict) -> Any:141 """Run a resolved tool node, adapting arguments and picking the vehicle.142 143 MCP ``arguments`` is the equivalent of a JSON body: fit it to the144 handler's declared parameters through the same ``spread_over_params``145 REST uses (extras dropped) — MCP wraps the API, it does not bypass it.146 An async handler stays on the loop; a sync handler goes through the147 server pool. Either way the returned awaitable is awaited by the engine.148 """149 app = self.application150 kwargs = app.spread_over_params(node, arguments)151 if asyncio.iscoroutinefunction(node):152 return node(**kwargs)153 server = app.server154 assert server is not None # a request is in flight, so the app is owned155 return server.run_sync(lambda: node(**kwargs))156 157 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:158 """Answer one MCP request over the Streamable HTTP transport.159 160 Gates the request (403/400), then routes by method: GET opens the SSE161 push stream, POST dispatches the JSON-RPC message through the engine162 and answers with the JSON-RPC envelope (a notification — no ``id`` —163 answers HTTP 202 with an empty body), anything else answers 405.164 """165 app = self.application166 request = Request(scope, receive, server=app.server, application=app)167 await request.init()168 origin = request.headers.get("origin")169 if self.allowed_origins is not None and origin is not None and origin not in self.allowed_origins:170 raise HTTPForbidden(f"Origin not allowed: {origin}")171 version = request.headers.get("mcp-protocol-version")172 if version is not None and version not in McpEngine.SUPPORTED_VERSIONS:173 raise HTTPBadRequest(f"Unsupported MCP-Protocol-Version: {version}")174 if request.method == "GET":175 await self.open_stream(request, scope, receive, send)176 return177 if request.method != "POST":178 raise HTTPException(405, "Method Not Allowed", headers=[(b"allow", b"GET, POST")])179 180 response = request.response181 envelope = request.data182 if isinstance(envelope, dict) and "id" not in envelope: # notification183 response.status_code = 202184 await response(scope, receive, send)185 return186 jsonrpc_id = envelope.get("id") if isinstance(envelope, dict) else None187 assert self._engine is not None # built at construction188 try:189 result = await self._engine.dispatch(envelope, request.auth_tags)190 except McpError as exc:191 payload = self._jsonrpc_error(exc.code, exc.message, jsonrpc_id)192 except Exception as exc: # noqa: BLE001 — any handler failure becomes a JSON-RPC error193 payload = self._jsonrpc_error(JSONRPC_INTERNAL_ERROR, str(exc), jsonrpc_id)194 else:195 payload = {"jsonrpc": "2.0", "id": jsonrpc_id, "result": result}196 response.set_result(payload)197 await response(scope, receive, send)198 199 async def open_stream(self, request: Request, scope: Scope, receive: Receive, send: Send) -> None:200 """Open the SSE push stream for one MCP session (the GET branch).201 202 ``Mcp-Session-Id`` is echoed when the client supplies one, minted203 otherwise (``secrets.token_urlsafe`` — the ``MemorySessionStore.create``204 pattern) and always returned in the response headers. A ``Last-Event-ID``205 header asks for the snapshot baseline before the live feed. A server206 composed without tasks has no hub: GET answers 405 (the 1c stateless207 behavior).208 """209 server = self.application.server210 if not getattr(server, "tasks_enabled", False):211 raise HTTPException(405, "Method Not Allowed", headers=[(b"allow", b"POST")])212 session_id = request.headers.get("mcp-session-id") or secrets.token_urlsafe(32)213 last_event_id = request.headers.get("last-event-id")214 stream = SseStream(self._event_source(server, session_id, last_event_id))215 response = stream.response()216 response.set_header("mcp-session-id", session_id)217 await response(scope, receive, send)218 219 async def _event_source(220 self, server: Any, session_id: str, last_event_id: str | None221 ) -> AsyncIterator[dict[str, Any]]:222 """The session's events: snapshot baseline on reconnect, then live.223 224 The baseline (only when the client presented ``Last-Event-ID``) replays225 the current ``progress.json`` of the session's pending/active tasks —226 spool reads off the loop via ``run_sync``. The live half follows a hub227 subscription; the queue is unsubscribed when the stream closes (client228 gone / task cancelled).229 """230 manager = server.tasks231 if last_event_id is not None:232 for event in await server.run_sync(lambda: self._baseline(manager, session_id)):233 yield self._sse_event(event)234 queue = manager.hub.subscribe(session_id)235 try:236 while True:237 yield self._sse_event(await queue.get())238 finally:239 manager.hub.unsubscribe(session_id, queue)240 241 def _sse_event(self, event: Any) -> dict[str, Any]:242 """Adapt one hub event into the SSE event dict.243 244 The WHOLE hub event is the ``data:`` payload (the client gets245 ``type``/``task_id`` back intact); its ``type`` doubles as the SSE246 ``event:`` field so clients can listen per kind.247 """248 kind = event.get("type") if isinstance(event, dict) else None249 return {"event": kind, "data": event}250 251 def _baseline(self, manager: Any, session_id: str) -> list[dict[str, Any]]:252 """The current progress snapshots of the session's live tasks (sync, pool).253 254 Settled tasks are not replayed (their result is fetched, not streamed —255 no durable event log, ratified); tasks without a snapshot yield nothing.256 """257 events: list[dict[str, Any]] = []258 spool = manager.spool259 for descriptor in spool.list_pending() + spool.list_active(manager.worker_id):260 if descriptor.get("session_id") != session_id:261 continue262 snapshot = spool.read_progress(descriptor["task_id"])263 if snapshot is not None:264 events.append(265 {"type": "progress", "task_id": descriptor["task_id"], "data": snapshot}266 )267 return events268 269 def _jsonrpc_error(self, code: int, message: str, jsonrpc_id: Any) -> dict:270 """Build a JSON-RPC 2.0 error envelope."""271 return {"jsonrpc": "2.0", "id": jsonrpc_id, "error": {"code": code, "message": message}}272 273 274 class McpApplication(RoutedApplication):275 """Standalone MCP transport: the whole app is one JSON-RPC endpoint.276 277 Supply the tool surface as an external ``RoutingClass`` (``routing_class=``278 instance, or ``module="pkg.mod:Class"`` imported and instantiated with no279 arguments); every route on it becomes a tool. The ``channel`` filter only280 bites when the external router plugs the ``channel`` plugin, so a plain281 router exposes all its routes. Without a router the app still answers282 ``initialize`` and lists no tools. Class attributes carry the MCP identity283 defaults so a subclass can set them declaratively.284 """285 286 mcp_name: ClassVar[str] = "genro-mcp"287 mcp_version: ClassVar[str] = "1.0.0"288 tool_separator: ClassVar[str] = "."289 mcp_channel: ClassVar[str] = "mcp"290 291 def __init__(self, **kwargs: Any) -> None:292 cls = type(self)293 name = kwargs.pop("mcp_name", cls.mcp_name)294 version = kwargs.pop("mcp_version", cls.mcp_version)295 separator = kwargs.pop("tool_separator", cls.tool_separator)296 allowed_origins = kwargs.pop("allowed_origins", None)297 routing_class: RoutingClass | None = kwargs.pop("routing_class", None)298 module: str | None = kwargs.pop("module", None)299 super().__init__(**kwargs)300 self._transport = McpTransport(301 self,302 name=name,303 version=version,304 tool_separator=separator,305 channel=cls.mcp_channel,306 allowed_origins=allowed_origins,307 )308 self._transport.build_engine(self._resolve_router(routing_class, module))309 310 @property311 def mcp_engine(self) -> McpEngine | None:312 """The MCP engine driving this app's tool surface (``None`` if unbuilt)."""313 return self._transport.engine314 315 def _resolve_router(316 self, routing_class: RoutingClass | None, module: str | None317 ) -> Router | None:318 """Resolve the external router from a ``routing_class`` or a ``module`` path."""319 if routing_class is None and module:320 routing_class = self._import_routing_class(module)321 return routing_class.route if routing_class is not None else None322 323 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:324 """Every request is an MCP JSON-RPC message on the single endpoint."""325 await self._transport.handle(scope, receive, send)326 327 328 class McpOpenApiApplication(OpenApiApplication):329 """OpenApiApplication that also exposes its API router as MCP tools.330 331 The REST/OpenAPI faces work exactly as in :class:`OpenApiApplication`; the332 MCP face answers under ``mcp_name_segment`` (default ``"mcp"``) via the same333 engine, pointed at the API router (the app's own in direct mode, the mounted334 class's in mounted mode). Visibility is the ``channel`` plugin's job: the335 MCP face lists only channel-``"mcp"`` entries; undeclared methods default to336 REST-only (``channels=rest_channel``).337 """338 339 mcp_name: ClassVar[str] = "genro-mcp"340 mcp_version: ClassVar[str] = "1.0.0"341 tool_separator: ClassVar[str] = "."342 mcp_channel: ClassVar[str] = "mcp"343 rest_channel: ClassVar[str] = "rest"344 345 def __init__(self, **kwargs: Any) -> None:346 cls = type(self)347 self._mcp_segment: str = kwargs.pop("mcp_name_segment", "mcp")348 name = kwargs.pop("mcp_name", cls.mcp_name)349 version = kwargs.pop("mcp_version", cls.mcp_version)350 separator = kwargs.pop("tool_separator", cls.tool_separator)351 allowed_origins = kwargs.pop("allowed_origins", None)352 self._mounted_router: Router | None = None353 # OpenApiApplication consumes routing_class/module to mount; the engine354 # is pointed at the app's own router (direct) or the mounted one.355 mounted = kwargs.get("routing_class") is not None or kwargs.get("module") is not None356 super().__init__(**kwargs)357 self._transport = McpTransport(358 self,359 name=name,360 version=version,361 tool_separator=separator,362 channel=cls.mcp_channel,363 allowed_origins=allowed_origins,364 )365 if mounted:366 self._transport.build_engine(self._mounted_router)367 else:368 self._transport.plug_if_absent(self.route, "channel")369 self.route.channel.configure(channels=cls.rest_channel)370 self._transport.build_engine(self.route)371 372 @property373 def mcp_engine(self) -> McpEngine | None:374 """The MCP engine driving this app's tool surface (``None`` if unbuilt)."""375 return self._transport.engine376 377 @property378 def mcp_name_segment(self) -> str:379 """Path segment under which the MCP JSON-RPC face is served."""380 return self._mcp_segment381 382 def auth_filters(self, scope: Scope) -> dict[str, str]:383 """Node-resolution filters: the base auth tags plus the REST channel.384 385 The API router carries the ``channel`` plugin (a method is an MCP tool386 only on channel ``"mcp"``), so the REST face must resolve on the REST387 channel; the MCP face passes ``"mcp"`` through the engine. The filter is388 harmless when no router on the path plugs ``channel`` (it is ignored).389 """390 filters = super().auth_filters(scope)391 filters["channel_channel"] = self.rest_channel392 return filters393 394 def schema_filters(self) -> dict[str, Any]:395 """Build the OpenAPI schema over the REST channel (the channel plugin is armed)."""396 return {"channel_channel": self.rest_channel}397 398 def _mount_routing_class(self, routing_class: RoutingClass) -> None:399 """Mount the API for OpenAPI, then remember its router for the MCP engine."""400 super()._mount_routing_class(routing_class)401 self._mounted_router = routing_class.route402 403 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:404 """Route the MCP segment to the JSON-RPC face; everything else to REST."""405 path = str(scope.get("path", "/")).lstrip("/")406 segment = path.partition("/")[0]407 if segment == self._mcp_segment:408 await self._transport.handle(scope, receive, send)409 else:410 await super().__call__(scope, receive, send)