# Copyright 2025 Softwell S.r.l.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""McpEngine — MCP (JSON-RPC 2.0) core over a genro-routes Router.
The engine turns a Router's ``@route`` entries into MCP tools and serves the
protocol methods ``initialize`` / ``tools/list`` / ``tools/call``. It is
transport- and app-agnostic: it holds a Router and a channel to filter on and
never touches HTTP concerns (headers, Origin, 202-for-notifications belong to
the host application). ``dispatch`` receives the parsed JSON-RPC message and
returns the RESULT object — envelope bookkeeping (``id``, ``jsonrpc``) stays
with the transport; protocol failures raise :class:`McpError` carrying the
JSON-RPC code for the transport to render. A list payload is rejected with
-32600: JSON-RPC batching entered the MCP spec in 2025-03-26 and was removed
in 2025-06-18.
Protocol (2025-11-25, the current revision):
- ``initialize`` negotiates the version: the client's requested version is
echoed when it appears in ``SUPPORTED_VERSIONS``, anything else is answered
with the latest supported revision.
- ``tools/list`` walks ``router.nodes(forbidden=False, channel_channel=...)``
so only the entries reachable on the engine's channel are advertised. Tool
names join the router path with ``tool_separator`` (default ``"."``, a
character illegal in Python identifiers, so ``sub.ping`` <-> ``sub/ping``
round-trips losslessly). Descriptors read ONLY the neutral blocks cached by
genro-routes' pydantic plugin at decoration time: ``inputSchema`` from the
entry's ``params.schema`` (aggregate ``request_schema``; fallback: an object
schema assembled from ``params.fields``), ``outputSchema`` from
``result.schema`` (``response_schema``). Nothing is derived from the
callable and pydantic is never imported here.
- ``tools/call`` resolves the tool through ``router.node(path, ...)`` and
reads the ``node.error`` STRING CODE (the stable genro-routes contract —
resolution never raises): ``not_found``/``not_available`` -> -32601,
``not_authorized``/``not_authenticated`` -> -32000, any other code ->
-32603. Execution is delegated to the ``invoke`` callback so a host app can
interpose parameter adaptation and pool dispatch; the default calls the
node directly and the engine awaits an awaitable result (async handlers),
nothing more. Input-validation failures are TOOL EXECUTION errors —
``{"isError": true, "content": [...]}`` results, not JSON-RPC protocol
errors (SEP-1303, enables model self-correction). Validation runs INSIDE
genro-routes (the pydantic plugin validates at call time; the engine never
re-validates). Since genro-routes 0.28.0 every bad-argument error — a
``pydantic.ValidationError`` or an unbindable-argument ``TypeError`` alike —
is channelled through the node's ``errors={"validation_error": ...}`` seam
to a local marker, so this module needs no pydantic import; the bare
``TypeError`` catch remains for an async handler body raising at await time
(a sync body's TypeError is already folded into the marker upstream). Both
become ``isError`` results. ``ping`` answers an empty result (spec MUST).
A dict result is returned BOTH as ``structuredContent`` and as its JSON
text rendering (the unstructured content SHOULD match the declared
``outputSchema``); any other result stays text-only.
"""
from __future__ import annotations
import inspect
import json
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from genro_routes import Router, RouterNode
__all__ = [
"JSONRPC_INTERNAL_ERROR",
"JSONRPC_INVALID_REQUEST",
"JSONRPC_METHOD_NOT_FOUND",
"JSONRPC_NOT_AUTHORIZED",
"McpEngine",
"McpError",
]
JSONRPC_INVALID_REQUEST = -32600
JSONRPC_METHOD_NOT_FOUND = -32601
JSONRPC_INTERNAL_ERROR = -32603
JSONRPC_NOT_AUTHORIZED = -32000
[docs]
class McpError(Exception):
"""Carries a JSON-RPC error code + message for the transport to render."""
def __init__(self, code: int, message: str) -> None:
super().__init__(message)
self.code = code
self.message = message
class _ToolArgumentsInvalid(Exception):
"""Marker raised by the node's ``validation_error`` exception mapping.
genro-routes re-raises an escaping ``pydantic.ValidationError`` as the
class mapped to the ``validation_error`` code, with the original error as
``__cause__`` — the engine catches bad tool arguments through this class
without importing pydantic.
"""
[docs]
class McpEngine:
"""MCP JSON-RPC core over a router.
Args:
router: The genro-routes Router whose entries are exposed as tools.
name / version: server identity returned by ``initialize``.
tool_separator: joins router/method segments into a flat tool name.
channel: channel to filter entries on (visibility per channel).
invoke: callback ``(node, arguments) -> result`` running a resolved
node; the engine awaits an awaitable result. Host applications
pass their own to interpose parameter adaptation (e.g.
``spread_over_params``) and pool dispatch for sync handlers; the
default calls the node directly.
"""
SUPPORTED_VERSIONS: tuple[str, ...] = ("2025-11-25", "2025-06-18", "2025-03-26")
def __init__(
self,
router: Router | None = None,
*,
name: str = "genro-mcp",
version: str = "1.0.0",
tool_separator: str = ".",
channel: str = "mcp",
invoke: Callable[[Any, dict], Any] | None = None,
) -> None:
self.router = router
self.name = name
self.version = version
self.tool_separator = tool_separator
self.channel = channel
self._invoke = invoke or self._default_invoke
def _default_invoke(self, node: RouterNode, arguments: dict) -> Any:
"""Raw invocation, no parameter adaptation; ``dispatch`` awaits it."""
return node(**arguments)
[docs]
async def dispatch(self, payload: Any, auth_tags: Any = None) -> dict:
"""Route a parsed JSON-RPC message to its MCP handler.
Returns the JSON-RPC RESULT object; the transport owns the envelope.
Raises:
McpError: invalid message shape (-32600, batching included) or
unknown method (-32601); ``tools/call`` resolution failures
bubble up from :meth:`handle_tools_call`.
"""
if isinstance(payload, list):
raise McpError(JSONRPC_INVALID_REQUEST, "JSON-RPC batching is not supported")
if not isinstance(payload, dict):
raise McpError(JSONRPC_INVALID_REQUEST, "Invalid JSON-RPC message")
method = payload.get("method")
if not isinstance(method, str):
raise McpError(JSONRPC_INVALID_REQUEST, "Missing method")
params = payload.get("params")
if params is None:
params = {}
if not isinstance(params, dict):
raise McpError(JSONRPC_INVALID_REQUEST, "params must be an object")
if method == "ping":
return {}
if method == "initialize":
return self.handle_initialize(params)
if method == "tools/list":
return self.handle_tools_list()
if method == "tools/call":
return await self.handle_tools_call(params, auth_tags)
raise McpError(JSONRPC_METHOD_NOT_FOUND, f"Method not found: {method}")
[docs]
def handle_initialize(self, params: dict) -> dict:
"""Negotiate the protocol version and return the server capabilities.
The client's requested version is echoed when supported; any other
request is answered with the latest supported revision (spec
negotiation rule).
"""
requested = params.get("protocolVersion")
version = requested if requested in self.SUPPORTED_VERSIONS else self.SUPPORTED_VERSIONS[0]
return {
"protocolVersion": version,
# experimental.push: the host transport's SSE progress channel
# (GET + Mcp-Session-Id); the engine itself stays transport-blind.
"capabilities": {"tools": {}, "experimental": {"push": {}}},
"serverInfo": {"name": self.name, "version": self.version},
}
def _collect_tools(self, nodes: dict, prefix: str, tools: list) -> None:
"""Recursively collect tool descriptors from router nodes."""
sep = self.tool_separator
for name, info in nodes.get("entries", {}).items():
tool_name = name if not prefix else f"{prefix}{sep}{name}"
tools.append(self._build_tool_descriptor(tool_name, info))
for router_name, sub_nodes in nodes.get("routers", {}).items():
sub_prefix = router_name if not prefix else f"{prefix}{sep}{router_name}"
self._collect_tools(sub_nodes, sub_prefix, tools)
def _build_tool_descriptor(self, tool_name: str, info: dict) -> dict:
"""Build an MCP tool descriptor from a nodes() entry info.
Reads only the neutral blocks: ``inputSchema`` from the ``params``
block, ``outputSchema`` from the ``result`` block (present when the
pydantic plugin captured a return-type schema).
"""
metadata = info.get("metadata") or {}
description = metadata.get("meta", {}).get("description") or info.get("doc") or tool_name
descriptor: dict = {
"name": tool_name,
"description": description,
"inputSchema": self._input_schema(info),
}
output_schema = (info.get("result") or {}).get("schema")
if output_schema is not None:
descriptor["outputSchema"] = output_schema
return descriptor
def _input_schema(self, info: dict) -> dict:
"""Tool inputSchema from the neutral ``params`` block, never the callable.
Primary source: the aggregate ``request_schema`` the pydantic plugin
cached (copied before the ``type`` default so the cache is never
mutated). Fallback: an object schema assembled from the cached
per-parameter ``fields`` (untyped and var-parameters carry no schema
and are skipped). No captured params -> empty object schema.
"""
params_block = info.get("params") or {}
schema = params_block.get("schema")
if schema is not None:
schema = dict(schema)
schema.setdefault("type", "object")
return schema
properties: dict[str, Any] = {}
required: list[str] = []
for field in params_block.get("fields") or []:
field_schema = field.get("schema")
if field_schema is None:
continue
properties[field["name"]] = field_schema
if field.get("required"):
required.append(field["name"])
assembled: dict[str, Any] = {"type": "object", "properties": properties}
if required:
assembled["required"] = required
return assembled
def _tool_result(self, result: Any) -> dict:
"""Wrap a handler result as a ``tools/call`` result object.
A dict rides BOTH as ``structuredContent`` and as JSON text content;
anything else is text-only.
"""
payload: dict[str, Any] = {
"content": [{"type": "text", "text": self._serialize_result(result)}]
}
if isinstance(result, dict):
payload["structuredContent"] = result
return payload
def _serialize_result(self, result: Any) -> str:
"""Serialize a result to the text content string."""
if isinstance(result, str):
return result
try:
return json.dumps(result, ensure_ascii=False, default=str)
except (TypeError, ValueError):
return str(result)
if __name__ == "__main__":
import asyncio
from genro_routes import RoutingClass, route
class Demo(RoutingClass):
"""Demo service exposing one MCP tool."""
def __init__(self) -> None:
self.route.plug("pydantic")
self.route.plug("channel")
@route(channel_channels="mcp")
def add(self, x: int, y: int = 0) -> dict:
"""Add two numbers."""
return {"sum": x + y}
engine = McpEngine(Demo().route, name="demo")
init = asyncio.run(engine.dispatch({"method": "initialize", "params": {}}))
assert init["protocolVersion"] == "2025-11-25"
tools = asyncio.run(engine.dispatch({"method": "tools/list"}))
assert [tool["name"] for tool in tools["tools"]] == ["add"]
call = asyncio.run(
engine.dispatch(
{"method": "tools/call", "params": {"name": "add", "arguments": {"x": 2, "y": 3}}}
)
)
assert call["structuredContent"] == {"sum": 5}
bad = asyncio.run(
engine.dispatch(
{"method": "tools/call", "params": {"name": "add", "arguments": {"x": "nope"}}}
)
)
assert bad["isError"] is True