# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""McpApplication - MCP Streamable HTTP transport for genro-asgi.
Implements the MCP (Model Context Protocol) Streamable HTTP transport
using JSON-RPC 2.0 envelopes over standard genro-asgi routing.
The application works inside the Dispatcher via the default_entry
mechanism: when mounted as an app (e.g., under ``/mcp/``), requests
to ``POST /mcp/`` resolve to the ``index`` route handler.
Spec references:
JSON-RPC 2.0: https://www.jsonrpc.org/specification
MCP Streamable HTTP: stateless mode
Protocol summary:
All requests: POST with JSON-RPC 2.0 body (parsed by framework)
Request (has "id"): process and return JSON-RPC result
Notification (no "id"): return HTTP 202, no body
tools/list: enumerate tools from the external router
tools/call: dispatch to router node by path
initialize: return server capabilities
Tool name to route path convention:
"code_search_symbols" -> "code/search_symbols"
Integration with genro-asgi:
McpApplication extends AsgiApplication and uses @route for its
endpoint. The Dispatcher handles request creation, middleware
chain (auth, CORS, errors), and response sending. The handler
accesses the parsed body via ``get_current_request().data``.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from genro_routes import route # type: ignore[import-untyped]
from ...request import get_current_request
from ..asgi_application import AsgiApplication
from .mcp_engine import JSONRPC_INVALID_REQUEST, JSONRPC_INTERNAL_ERROR, McpEngine, McpError
if TYPE_CHECKING:
from genro_routes import Router # type: ignore[import-untyped]
__all__ = ["McpApplication"]
[docs]
class McpApplication(AsgiApplication):
"""MCP Streamable HTTP transport as a genro-asgi application.
Accepts an external genro-routes Router and exposes its @route
entries as MCP tools via JSON-RPC 2.0 over HTTP POST.
Mounts inside the server like any other app and participates in
the full middleware chain (auth, CORS, errors). The ``index``
route handles all JSON-RPC messages via the default_entry mechanism.
Usage::
class SourcererMcpApp(McpApplication):
mcp_name = "sourcerer"
mcp_version = "1.0.0"
def on_init(self, **kwargs):
api = SourcererAPI(...)
self.set_router(api.route)
"""
mcp_name: str = "genro-mcp"
mcp_version: str = "1.0.0"
# Separator joining router/method segments into a flat MCP tool name. Must be a
# character that cannot appear in a Python identifier, so the tool name splits
# back to the route path losslessly — an underscore would be ambiguous with the
# underscores in router/method names (e.g. a router "target_db").
# TODO: this belongs to a future per-app grammar/config (a config.py for the app,
# by analogy with the server's), where the tool-naming scheme is declared rather
# than hard-coded here.
tool_separator: str = "."
[docs]
def __init__(self, **kwargs: Any) -> None:
"""Initialize with optional mcp_name/mcp_version/tool_separator overrides."""
mcp_name = kwargs.pop("mcp_name", None)
mcp_version = kwargs.pop("mcp_version", None)
tool_separator = kwargs.pop("tool_separator", None)
if mcp_name is not None:
self.mcp_name = mcp_name
if mcp_version is not None:
self.mcp_version = mcp_version
if tool_separator is not None:
self.tool_separator = tool_separator
self._engine = McpEngine(
name=self.mcp_name,
version=self.mcp_version,
tool_separator=self.tool_separator,
)
super().__init__(**kwargs)
[docs]
def set_router(self, router: Router) -> None:
"""Set the genro-routes Router to expose as MCP tools.
The engine holds the JSON-RPC core (initialize / tools/list / tools/call);
this app is the HTTP transport. Without a router, initialize still works
and tools/list returns an empty list.
"""
self._engine.router = router
[docs]
@route(media_type="application/json")
def index(self) -> dict | None:
"""JSON-RPC 2.0 endpoint for MCP Streamable HTTP.
Reads the parsed body from ``request.data`` (populated by the framework
via asgi_data) and delegates dispatch to the MCP engine. Notifications
(no "id") get HTTP 202 and no body; requests get a JSON-RPC response.
"""
request = get_current_request()
assert request is not None # guaranteed inside Dispatcher
envelope = request.data
if not isinstance(envelope, dict):
return self._jsonrpc_error(JSONRPC_INVALID_REQUEST, "Invalid request", None)
jsonrpc_id = envelope.get("id")
method = envelope.get("method", "")
params = envelope.get("params") or {}
# Notification: no id → HTTP 202, no body
if "id" not in envelope:
request.response.status_code = 202
return None
try:
result = self._engine.dispatch(method, params, request.auth_tags)
except McpError as exc:
return self._jsonrpc_error(exc.code, exc.message, jsonrpc_id)
except Exception as exc:
return self._jsonrpc_error(JSONRPC_INTERNAL_ERROR, str(exc), jsonrpc_id)
return {"jsonrpc": "2.0", "id": jsonrpc_id, "result": result}
def _jsonrpc_error(self, code: int, message: str, jsonrpc_id: Any) -> dict:
"""Build a JSON-RPC 2.0 error response dict."""
return {
"jsonrpc": "2.0",
"id": jsonrpc_id,
"error": {"code": code, "message": message},
}
if __name__ == "__main__":
app = McpApplication()
print(f"McpApplication: {app.mcp_name} v{app.mcp_version}")