# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""McpOpenApiApplication - one router, two faces: OpenAPI + MCP.
MCP wraps OpenAPI: every MCP tool is, underneath, an OpenAPI endpoint of the same
router. This app inherits the whole OpenAPI machinery (_meta docs, pydantic plug,
make_callable) and adds an MCP face under ``mcp`` that exposes the same router's
entries as JSON-RPC tools.
Two entry shapes, one core:
- ``/{app}/api/endpoint`` REST/OpenAPI (path is the endpoint)
- ``/{app}/mcp`` MCP JSON-RPC (endpoint inside the payload)
The API router comes in two modes:
- **mounted mode** — pass ``routing_class=`` (or ``module=``) and the class is
mounted under ``api_name`` exactly as in OpenApiApplication; the MCP engine
points at its router. Channel visibility is the mounted class's own business
(plug ``channel`` and configure it there).
- **direct mode** — pass nothing: the app's own ``@route`` methods are the API
(endpoints at the app root). The MCP engine points at the app router; the app
plugs ``channel`` with default ``rest``, so a method is REST-only unless its
decorator says otherwise: ``@route(channel_channels="mcp,rest")`` exposes it
on both faces. Nothing becomes an MCP tool undeclared.
Visibility per face is driven by the ``channel`` plugin: the MCP face lists only
entries exposed on channel ``mcp``; the REST face passes the request's own filters.
"""
from __future__ import annotations
from typing import Any
from genro_routes import RoutingClass, route # type: ignore[import-untyped]
from genro_toolbox.smartasync import smartasync # type: ignore[import-untyped]
from ...request import get_current_request
from ..mcp_application.mcp_engine import (
JSONRPC_INVALID_REQUEST,
JSONRPC_INTERNAL_ERROR,
McpEngine,
McpError,
)
from .openapi_application import OpenApiApplication
__all__ = ["McpOpenApiApplication"]
[docs]
class McpOpenApiApplication(OpenApiApplication):
"""OpenApiApplication that also exposes its API router as MCP tools."""
mcp_name: str = "genro-mcp"
mcp_version: str = "1.0.0"
tool_separator: str = "."
mcp_name_segment: str = "mcp"
[docs]
def on_init(
self,
routing_class: RoutingClass | None = None,
module: str | None = None,
**kwargs: Any,
) -> None:
"""Initialize; with no routing_class/module, direct mode (see module doc).
In direct mode the app's own ``@route`` methods are the API: pydantic is
plugged on the app router (signatures for schema and adaptation) and the
channel plugin with default ``rest`` (a method joins the MCP face only
via ``@route(channel_channels="mcp,...")``). The attached MCP face
inherits the pydantic plug from the app router.
"""
super().on_init(routing_class=routing_class, module=module, **kwargs)
if routing_class is None and module is None:
self.route.plug("pydantic")
self.route.plug("channel")
self.route.channel.configure(channels=self.rest_channel)
self._attach_mcp_face(self.route)
def _mount_routing_class(self, routing_class: RoutingClass) -> None:
"""Mount the RoutingClass for OpenAPI, then attach the MCP face on it.
The face's own pydantic plug makes its index() report a (empty)
signature: make_callable then drops the JSON-RPC body_data instead of
passing it. (In direct mode the plug is inherited, see on_init.)
"""
super()._mount_routing_class(routing_class)
face = self._attach_mcp_face(routing_class.route)
face.route.plug("pydantic")
def _attach_mcp_face(self, router: Any) -> "McpFace":
"""Attach the MCP JSON-RPC face whose engine reads ``router``.
The engine invokes nodes through ``spread_over_params`` — the very same
parameter adaptation REST uses in ``make_callable`` — so tools go
through the same logic as REST; MCP wraps OpenAPI, it does not bypass.
"""
engine = McpEngine(
router,
name=self.mcp_name,
version=self.mcp_version,
tool_separator=self.tool_separator,
channel="mcp",
invoke=self._invoke_tool,
)
face = McpFace(self, engine)
self.attach_instance(face, name=self.mcp_name_segment)
return face
def _invoke_tool(self, node: Any, arguments: dict) -> Any:
"""Invoke a resolved node with MCP arguments, adapting like REST.
MCP ``arguments`` is the equivalent of a JSON body: fit it to the
handler's declared parameters via the same ``spread_over_params`` REST
uses (extras dropped; kept whole for ``**kwargs`` or no signature).
"""
return smartasync(node)(**self.spread_over_params(node, arguments))
class McpFace(RoutingClass):
"""The MCP JSON-RPC endpoint of a McpOpenApiApplication, attached under ``mcp``."""
def __init__(self, application: McpOpenApiApplication, engine: McpEngine) -> None:
self.application = application
self.engine = engine
@route(media_type="application/json")
def index(self) -> dict | None:
"""JSON-RPC 2.0 endpoint delegating to the MCP engine."""
request = get_current_request()
assert request is not None # guaranteed inside Dispatcher
envelope = request.data
if not isinstance(envelope, dict):
return self._error(JSONRPC_INVALID_REQUEST, "Invalid request", None)
jsonrpc_id = envelope.get("id")
method = envelope.get("method", "")
params = envelope.get("params") or {}
if "id" not in envelope: # notification
request.response.status_code = 202
return None
try:
result = self.engine.dispatch(method, params, request.auth_tags)
except McpError as exc:
return self._error(exc.code, exc.message, jsonrpc_id)
except Exception as exc:
return self._error(JSONRPC_INTERNAL_ERROR, str(exc), jsonrpc_id)
return {"jsonrpc": "2.0", "id": jsonrpc_id, "result": result}
def _error(self, code: int, message: str, jsonrpc_id: Any) -> dict:
return {"jsonrpc": "2.0", "id": jsonrpc_id, "error": {"code": code, "message": message}}
if __name__ == "__main__":
app = McpOpenApiApplication()
print(f"McpOpenApiApplication: MCP={app.mcp_name} v{app.mcp_version}")