tests/core/test_mcp_application.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 application tests (Macro 4 Phase 8).16 17 Requests drive a REAL ``AsgiServer`` composition at the ASGI level (no uvicorn),18 so the middleware chain (errors) turns the transport gates (405/400/403) into19 responses. ``McpApplication`` is mounted at ``/mcp`` over an empty primary;20 ``McpOpenApiApplication`` runs in direct mode as the primary. The GET-side21 helpers come from ``tests/conftest.py``; ``drive``/``mcp_post`` (local) add a22 body and custom headers.23 """24 25 from __future__ import annotations26 27 import json28 import threading29 from typing import Any, Callable30 31 import pytest32 from genro_routes import RoutingClass, route33 34 from genro_asgi import AsgiServer, McpApplication, McpOpenApiApplication, RoutedApplication35 from genro_asgi.types import Message, Scope36 37 38 class Empty(RoutedApplication):39 """A do-nothing primary so an MCP app can mount as a secondary."""40 41 42 class Calc(RoutingClass):43 """External tool surface: pydantic plugged, one sync and one async tool."""44 45 def __init__(self) -> None:46 self.route.plug("pydantic")47 self.threads: list[int] = []48 49 @route()50 def add(self, x: int, y: int = 0) -> dict:51 """Add two numbers."""52 self.threads.append(threading.get_ident())53 return {"sum": x + y}54 55 @route()56 async def greet(self, name: str) -> str:57 """Greet someone."""58 return f"hi {name}"59 60 61 @pytest.fixture62 def drive() -> Callable[..., object]:63 """Fixture: drive one request through a server at the ASGI level."""64 65 async def _drive(66 server: object,67 path: str,68 *,69 method: str = "GET",70 query: bytes = b"",71 headers: list[tuple[bytes, bytes]] | None = None,72 body: bytes = b"",73 ) -> list[Message]:74 scope: Scope = {75 "type": "http",76 "method": method,77 "path": path,78 "query_string": query,79 "headers": list(headers or []),80 }81 sent: list[Message] = []82 83 async def receive() -> Message:84 return {"type": "http.request", "body": body, "more_body": False}85 86 async def send(message: Message) -> None:87 sent.append(message)88 89 await server(scope, receive, send) # type: ignore[operator]90 return sent91 92 return _drive93 94 95 @pytest.fixture96 def mcp_post(drive: Callable[..., Any]) -> Callable[..., object]:97 """Fixture: POST a JSON-RPC envelope (application/json) to an MCP endpoint."""98 99 async def _mcp_post(100 server: object,101 path: str,102 envelope: dict,103 headers: list[tuple[bytes, bytes]] | None = None,104 ) -> list[Message]:105 hdrs = [(b"content-type", b"application/json"), *(headers or [])]106 return await drive(server, path, method="POST", headers=hdrs, body=json.dumps(envelope).encode())107 108 return _mcp_post109 110 111 def result_of(response_body: Callable[[list[Message]], bytes], sent: list[Message]) -> dict:112 """Parse the JSON-RPC envelope carried by the response body."""113 return json.loads(response_body(sent))114 115 116 def mcp_server(app: McpApplication) -> AsgiServer:117 """A server with ``app`` mounted at its ``mount`` over an empty root app."""118 server = AsgiServer(applications=[Empty(mount=""), app])119 return server120 121 122 class TestMcpApplication:123 async def test_initialize_negotiates_version(124 self, mcp_post, response_status, response_body125 ) -> None:126 app = McpApplication(code="mcp", routing_class=Calc())127 sent = await mcp_post(128 mcp_server(app), "/mcp", {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}129 )130 assert response_status(sent) == 200131 envelope = result_of(response_body, sent)132 assert envelope["id"] == 1133 assert envelope["result"]["protocolVersion"] == "2025-11-25"134 135 async def test_mcp_name_names_the_engine_the_client_sees(136 self, mcp_post, response_body137 ) -> None:138 app = McpApplication(code="mcp", mcp_name="demo", routing_class=Calc())139 assert app.mcp_engine.name == "demo"140 sent = await mcp_post(141 mcp_server(app), "/mcp", {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}142 )143 server_info = result_of(response_body, sent)["result"]["serverInfo"]144 assert server_info["name"] == "demo"145 146 async def test_tools_list_enumerates_external_router(147 self, mcp_post, response_status, response_body148 ) -> None:149 app = McpApplication(code="mcp", routing_class=Calc())150 sent = await mcp_post(mcp_server(app), "/mcp", {"jsonrpc": "2.0", "id": 2, "method": "tools/list"})151 assert response_status(sent) == 200152 tools = result_of(response_body, sent)["result"]["tools"]153 assert {tool["name"] for tool in tools} == {"add", "greet"}154 155 async def test_tools_call_sync_tool(self, mcp_post, response_body) -> None:156 app = McpApplication(code="mcp", routing_class=Calc())157 sent = await mcp_post(158 mcp_server(app),159 "/mcp",160 {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "add", "arguments": {"x": 2, "y": 3}}},161 )162 result = result_of(response_body, sent)["result"]163 assert result["structuredContent"] == {"sum": 5}164 165 async def test_tools_call_async_tool(self, mcp_post, response_body) -> None:166 app = McpApplication(code="mcp", routing_class=Calc())167 sent = await mcp_post(168 mcp_server(app),169 "/mcp",170 {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "greet", "arguments": {"name": "bob"}}},171 )172 result = result_of(response_body, sent)["result"]173 assert result["content"] == [{"type": "text", "text": "hi bob"}]174 175 async def test_sync_tool_runs_off_the_loop_thread(self, mcp_post) -> None:176 calc = Calc()177 app = McpApplication(code="mcp", routing_class=calc)178 await mcp_post(179 mcp_server(app),180 "/mcp",181 {"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "add", "arguments": {"x": 1}}},182 )183 assert calc.threads and calc.threads[0] != threading.get_ident()184 185 async def test_notification_answers_202_empty(186 self, mcp_post, response_status, response_body187 ) -> None:188 app = McpApplication(code="mcp", routing_class=Calc())189 sent = await mcp_post(mcp_server(app), "/mcp", {"jsonrpc": "2.0", "method": "initialize"})190 assert response_status(sent) == 202191 assert response_body(sent) == b""192 193 async def test_other_method_is_method_not_allowed(self, drive, response_status) -> None:194 # GET is the push stream since core 1e (test_mcp_push.py); DELETE keeps195 # the 405 gate covered.196 app = McpApplication(code="mcp", routing_class=Calc())197 sent = await drive(mcp_server(app), "/mcp", method="DELETE")198 assert response_status(sent) == 405199 200 async def test_unsupported_protocol_version_is_400(201 self, mcp_post, response_status202 ) -> None:203 app = McpApplication(code="mcp", routing_class=Calc())204 sent = await mcp_post(205 mcp_server(app),206 "/mcp",207 {"jsonrpc": "2.0", "id": 6, "method": "tools/list"},208 headers=[(b"mcp-protocol-version", b"1999-01-01")],209 )210 assert response_status(sent) == 400211 212 async def test_supported_protocol_version_passes(213 self, mcp_post, response_status214 ) -> None:215 app = McpApplication(code="mcp", routing_class=Calc())216 sent = await mcp_post(217 mcp_server(app),218 "/mcp",219 {"jsonrpc": "2.0", "id": 7, "method": "tools/list"},220 headers=[(b"mcp-protocol-version", b"2025-11-25")],221 )222 assert response_status(sent) == 200223 224 async def test_disallowed_origin_is_403(self, mcp_post, response_status) -> None:225 app = McpApplication(code="mcp", routing_class=Calc(), allowed_origins=["https://ok.example"])226 sent = await mcp_post(227 mcp_server(app),228 "/mcp",229 {"jsonrpc": "2.0", "id": 8, "method": "tools/list"},230 headers=[(b"origin", b"https://evil.example")],231 )232 assert response_status(sent) == 403233 234 async def test_allowed_origin_passes(self, mcp_post, response_status) -> None:235 app = McpApplication(code="mcp", routing_class=Calc(), allowed_origins=["https://ok.example"])236 sent = await mcp_post(237 mcp_server(app),238 "/mcp",239 {"jsonrpc": "2.0", "id": 9, "method": "tools/list"},240 headers=[(b"origin", b"https://ok.example")],241 )242 assert response_status(sent) == 200243 244 async def test_unknown_method_is_a_jsonrpc_error(self, mcp_post, response_status, response_body) -> None:245 app = McpApplication(code="mcp", routing_class=Calc())246 sent = await mcp_post(247 mcp_server(app), "/mcp", {"jsonrpc": "2.0", "id": 10, "method": "resources/list"}248 )249 # A protocol error rides a JSON-RPC error envelope with HTTP 200.250 assert response_status(sent) == 200251 envelope = result_of(response_body, sent)252 assert envelope["error"]["code"] == -32601253 assert envelope["id"] == 10254 255 async def test_without_router_lists_no_tools(self, mcp_post, response_body) -> None:256 app = McpApplication(code="mcp")257 sent = await mcp_post(mcp_server(app), "/mcp", {"jsonrpc": "2.0", "id": 11, "method": "tools/list"})258 assert result_of(response_body, sent)["result"] == {"tools": []}259 260 async def test_bad_arguments_are_a_tool_error(self, mcp_post, response_status, response_body) -> None:261 # A sync tool with an invalid annotated argument: the error travels back262 # through the run_sync future and lands as an isError result (SEP-1303),263 # never a JSON-RPC protocol error.264 app = McpApplication(code="mcp", routing_class=Calc())265 sent = await mcp_post(266 mcp_server(app),267 "/mcp",268 {"jsonrpc": "2.0", "id": 12, "method": "tools/call", "params": {"name": "add", "arguments": {"x": "nope"}}},269 )270 assert response_status(sent) == 200271 result = result_of(response_body, sent)["result"]272 assert result["isError"] is True273 274 async def test_non_object_body_is_invalid_request(self, drive, response_status, response_body) -> None:275 # A JSON body that is not an object is rejected by the engine with -32600276 # and rendered as a JSON-RPC error envelope (HTTP 200, id null).277 app = McpApplication(code="mcp", routing_class=Calc())278 sent = await drive(279 mcp_server(app),280 "/mcp",281 method="POST",282 headers=[(b"content-type", b"application/json")],283 body=b'"not-an-object"',284 )285 assert response_status(sent) == 200286 envelope = result_of(response_body, sent)287 assert envelope["error"]["code"] == -32600288 assert envelope["id"] is None289 290 291 class BridgeApi(McpOpenApiApplication):292 """Direct-mode bridge: a dual method (REST + MCP) and a REST-only method."""293 294 openapi_info = {"title": "Bridge", "version": "1.0.0", "description": "dual-face"}295 296 @route(channel_channels="mcp,rest")297 def echo(self, msg: str = "hi") -> dict:298 """Echo a message."""299 return {"echo": msg}300 301 @route(channel_channels="rest")302 def only_rest(self) -> dict:303 """REST-only endpoint, never an MCP tool."""304 return {"rest": True}305 306 307 def bridge_server() -> AsgiServer:308 """A server whose plugin config arms openapi on the bridge app."""309 return AsgiServer(applications=[BridgeApi(mount="")], plugins={"openapi": True})310 311 312 class TestMcpOpenApiApplication:313 async def test_dual_method_same_result_on_both_faces(314 self, drive, mcp_post, response_body315 ) -> None:316 server = bridge_server()317 rest = await drive(server, "/echo", method="GET", query=b"msg=hey")318 assert json.loads(response_body(rest)) == {"echo": "hey"}319 320 mcp = await mcp_post(321 server,322 "/mcp",323 {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "echo", "arguments": {"msg": "hey"}}},324 )325 result = result_of(response_body, mcp)["result"]326 assert result["structuredContent"] == {"echo": "hey"}327 328 async def test_rest_only_absent_from_tools_list(self, mcp_post, response_body) -> None:329 server = bridge_server()330 sent = await mcp_post(server, "/mcp", {"jsonrpc": "2.0", "id": 2, "method": "tools/list"})331 names = {tool["name"] for tool in result_of(response_body, sent)["result"]["tools"]}332 assert "echo" in names333 assert "only_rest" not in names334 335 async def test_openapi_schema_still_serves(336 self, http_request, response_status, response_body337 ) -> None:338 sent = await http_request(bridge_server(), "/_meta/schema_json")339 assert response_status(sent) == 200340 doc = json.loads(response_body(sent))341 assert doc["openapi"] == "3.1.0"342 assert doc["info"]["title"] == "Bridge"343 assert "/echo" in doc["paths"]344 345 async def test_mcp_other_method_is_method_not_allowed(self, drive, response_status) -> None:346 # GET on the mcp segment is the push stream since core 1e (test_mcp_push.py)347 sent = await drive(bridge_server(), "/mcp", method="DELETE")348 assert response_status(sent) == 405349 350 351 class TestMcpOpenApiMountedMode:352 async def test_mounted_router_tools_and_rest(self, drive, mcp_post, response_body) -> None:353 class SubApi(RoutingClass):354 """External API mounted into the bridge under ``api_name``."""355 356 openapi_info = {"title": "Sub", "version": "2.0.0"}357 358 def __init__(self) -> None:359 self.route.plug("channel")360 self.route.channel.configure(channels="rest")361 362 @route(channel_channels="mcp,rest")363 def ping(self, name: str = "x") -> dict:364 """Ping."""365 return {"pong": name}366 367 app = McpOpenApiApplication(code="mount", routing_class=SubApi())368 server = AsgiServer(applications=[Empty(mount=""), app], plugins={"openapi": True})369 370 rest = await drive(server, "/mount/api/ping", method="GET", query=b"name=z")371 assert json.loads(response_body(rest)) == {"pong": "z"}372 373 mcp = await mcp_post(374 server,375 "/mount/mcp",376 {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "ping", "arguments": {"name": "z"}}},377 )378 assert result_of(response_body, mcp)["result"]["structuredContent"] == {"pong": "z"}379 380 381 class TestExternalRouterAuthEnforcement:382 """``build_engine`` plugs auth: a ruled tool on an external router is enforced."""383 384 def _app(self) -> McpApplication:385 class Ruled(RoutingClass):386 @route()387 def open_tool(self) -> dict:388 """Unruled tool."""389 return {"ok": True}390 391 @route(auth_rule="admin")392 def secret(self) -> dict:393 """Admin-only tool."""394 return {"secret": True}395 396 return McpApplication(code="mcp", routing_class=Ruled())397 398 async def test_ruled_tool_hidden_from_anonymous_tools_list(399 self, mcp_post, response_body400 ) -> None:401 sent = await mcp_post(402 mcp_server(self._app()),403 "/mcp",404 {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},405 )406 tools = {t["name"] for t in result_of(response_body, sent)["result"]["tools"]}407 assert "open_tool" in tools408 assert "secret" not in tools409 410 async def test_ruled_tool_call_denied_for_anonymous(411 self, mcp_post, response_body412 ) -> None:413 sent = await mcp_post(414 mcp_server(self._app()),415 "/mcp",416 {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "secret", "arguments": {}}},417 )418 assert result_of(response_body, sent)["error"]["code"] == -32000