diff --git a/src/reactome_mcp/README.md b/src/reactome_mcp/README.md index 3658a91..e50f5c8 100644 --- a/src/reactome_mcp/README.md +++ b/src/reactome_mcp/README.md @@ -60,6 +60,34 @@ not because the tool exists. ## Transport +Two, and which one you can use depends on where this runs. + +| variable | transport | where | +|---|---|---| +| `REACTOME_MCP_URL` | Streamable HTTP | **anywhere**, including the container | +| `REACTOME_MCP_SERVER` | stdio, spawning `node` | a developer's machine only | + +`REACTOME_MCP_URL` wins when both are set. + +**stdio cannot work in the deployed container.** The image is Python: it has no +`node`, and it does not mount reactome-mcp. `REACTOME_MCP_SERVER` can never be +satisfied there, so the live destination worked on every machine it was tested +on and none that it ships to. That is why the HTTP transport exists. + +To run one alongside the chatbot: + +```bash +cd ~/git/reactome-mcp && npm ci && npm run build +MCP_HTTP_PORT=4320 node dist/http-server.js +# then, for the chatbot: +REACTOME_MCP_URL=http://127.0.0.1:4320 +``` + +reactome-mcp binds loopback by default; see `specs/002-transport-and-hosting` +in that repository for why, and for the shape of a hosted deployment. + +## Transport internals + stdio, by spawning the server. reactome-mcp also serves Streamable HTTP, so a hosted instance can be used instead once there is one; that is `specs/006-mcp-hosting`. Keeping the transport inside `MCPProcessManager` is diff --git a/src/reactome_mcp/http_client.py b/src/reactome_mcp/http_client.py new file mode 100644 index 0000000..7d15c1e --- /dev/null +++ b/src/reactome_mcp/http_client.py @@ -0,0 +1,178 @@ +"""Talk to a reactome-mcp server over Streamable HTTP. + +The stdio client spawns `node` and speaks over a pipe. That works on a +developer's machine and cannot work where the chatbot actually runs: the +deployed image is Python, has no node, and does not mount reactome-mcp. So +`REACTOME_MCP_SERVER` can never be satisfied in the container -- the live +destination worked everywhere I tested it and nowhere it ships. + +reactome-mcp also serves Streamable HTTP, which is what a sibling container or +a hosted instance offers. This speaks that. + +Two things about the protocol worth stating, because both are easy to get wrong +and neither fails loudly: + + - The session id arrives in the `mcp-session-id` **response header** of the + initialize call, and must be sent on every request afterwards. Omit it and + the server answers 400, not a hint. + - Responses come back **SSE-framed** (`content-type: text/event-stream`, + `event: message` / `data: {...}`) rather than as bare JSON, even for a + single reply to a POST. Parsing the body as JSON gets a decode error on + text that is perfectly valid. +""" + +import json +import logging +from typing import Any + +import httpx + +from reactome_mcp.client import PROTOCOL_VERSION, MCPToolError + +logger = logging.getLogger(__name__) + +ACCEPT = "application/json, text/event-stream" + + +class MCPHttpClient: + """The same surface as the stdio `MCPClient`, over HTTP.""" + + def __init__(self, base_url: str, timeout: float = 30.0) -> None: + self.base_url = base_url.rstrip("/") + self.endpoint = f"{self.base_url}/mcp" + self.timeout = timeout + self._session_id: str | None = None + self._client = httpx.AsyncClient(timeout=timeout) + + def _headers(self) -> dict[str, str]: + headers = {"Content-Type": "application/json", "Accept": ACCEPT} + if self._session_id: + headers["mcp-session-id"] = self._session_id + return headers + + @staticmethod + def _parse(response: httpx.Response) -> dict[str, Any]: + """Read one JSON-RPC message out of a JSON or SSE body.""" + body = response.text + if "text/event-stream" in response.headers.get("content-type", ""): + # event: message \n data: {...}. Take the last data line: a stream + # may carry progress notifications before the reply. + payloads = [ + line[len("data:") :].strip() + for line in body.splitlines() + if line.startswith("data:") + ] + if not payloads: + raise MCPToolError(f"no data in SSE response: {body[:200]}") + body = payloads[-1] + + try: + message = json.loads(body) + except json.JSONDecodeError as exc: + raise MCPToolError(f"MCP returned invalid JSON: {exc}") from exc + + if not isinstance(message, dict): + raise MCPToolError( + f"MCP returned a {type(message).__name__}, expected an object" + ) + + if "error" in message: + error = message["error"] + raise MCPToolError(f"MCP error {error.get('code')}: {error.get('message')}") + + result = message.get("result", {}) + if not isinstance(result, dict): + raise MCPToolError( + f"MCP returned a {type(result).__name__} result, expected an object" + ) + return result + + async def initialize(self, client_name: str = "reactome-chatbot") -> dict[str, Any]: + response = await self._client.post( + self.endpoint, + headers=self._headers(), + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": client_name, "version": "1"}, + }, + }, + ) + response.raise_for_status() + + self._session_id = response.headers.get("mcp-session-id") + if not self._session_id: + raise MCPToolError( + "the server issued no mcp-session-id on initialize; every later " + "request would be refused" + ) + result = self._parse(response) + + # A notification: no id, no reply expected. + await self._client.post( + self.endpoint, + headers=self._headers(), + json={"jsonrpc": "2.0", "method": "notifications/initialized"}, + ) + + server = result.get("serverInfo", {}) + logger.info( + "MCP server ready over HTTP at %s: %s %s", + self.base_url, + server.get("name", "unknown"), + server.get("version", ""), + ) + return result + + async def call( + self, method: str, params: dict[str, Any] | None = None + ) -> dict[str, Any]: + response = await self._client.post( + self.endpoint, + headers=self._headers(), + json={"jsonrpc": "2.0", "id": 2, "method": method, "params": params or {}}, + ) + response.raise_for_status() + return self._parse(response) + + async def call_tool( + self, name: str, arguments: dict[str, Any] | None = None + ) -> str: + if self._session_id is None: + await self.initialize() + + result = await self.call( + "tools/call", {"name": name, "arguments": arguments or {}} + ) + blocks = result.get("content", []) + if not isinstance(blocks, list): + raise MCPToolError(f"{name} returned no content blocks") + text = "\n".join( + str(block.get("text", "")) + for block in blocks + if isinstance(block, dict) and block.get("type") == "text" + ) + if result.get("isError"): + raise MCPToolError(text or f"{name} failed with no message") + return text + + async def list_tools(self) -> list[dict[str, Any]]: + if self._session_id is None: + await self.initialize() + tools = (await self.call("tools/list")).get("tools", []) + if not isinstance(tools, list): + raise MCPToolError(f"MCP returned a {type(tools).__name__} tool list") + return [tool for tool in tools if isinstance(tool, dict)] + + async def aclose(self) -> None: + if self._session_id: + try: + await self._client.delete(self.endpoint, headers=self._headers()) + except Exception as exc: + logger.debug("could not end the MCP session cleanly: %s", exc) + await self._client.aclose() + self._session_id = None diff --git a/src/reactome_mcp/session.py b/src/reactome_mcp/session.py index dea8375..47f0e00 100644 --- a/src/reactome_mcp/session.py +++ b/src/reactome_mcp/session.py @@ -25,25 +25,32 @@ from langchain_core.tools import BaseTool from reactome_mcp.client import MCPClient +from reactome_mcp.http_client import MCPHttpClient from reactome_mcp.process import MCPConnectionError, MCPProcessManager -from reactome_mcp.tools import create_mcp_tools +from reactome_mcp.tools import ToolCaller, create_mcp_tools logger = logging.getLogger(__name__) _lock = asyncio.Lock() _manager: MCPProcessManager | None = None +_http: MCPHttpClient | None = None _tools: list[BaseTool] | None = None _failed = False def mcp_server_path() -> Path | None: - """Where the MCP server is, if this deployment has one.""" + """A local server to spawn, if one is configured.""" configured = os.getenv("REACTOME_MCP_SERVER") return Path(configured) if configured else None +def mcp_server_url() -> str | None: + """A running server to connect to, if one is configured.""" + return os.getenv("REACTOME_MCP_URL") or None + + def is_configured() -> bool: - return mcp_server_path() is not None + return mcp_server_url() is not None or mcp_server_path() is not None async def get_mcp_tools() -> list[BaseTool] | None: @@ -68,19 +75,30 @@ async def get_mcp_tools() -> list[BaseTool] | None: if _failed: return None + url = mcp_server_url() server_path = mcp_server_path() - if server_path is None: - return None + # HTTP first. The deployed image is Python with no node and does not + # mount reactome-mcp, so spawning one cannot work there -- stdio is for + # a developer's machine, where the repo and node both exist. + where = url or str(server_path) try: - manager = MCPProcessManager(server_path) - await manager.start() - if manager.process is None: - raise MCPConnectionError("server did not start") - client = MCPClient(manager.process) - # The handshake is the readiness check: if this returns, the server - # is up and answering. - await client.initialize() - _manager = manager + client: ToolCaller + if url is not None: + http = MCPHttpClient(url) + # The handshake is the readiness check: if this returns, the + # server is up and answering. + await http.initialize() + _http, client = http, http + elif server_path is not None: + manager = MCPProcessManager(server_path) + await manager.start() + if manager.process is None: + raise MCPConnectionError("server did not start") + stdio = MCPClient(manager.process) + await stdio.initialize() + _manager, client = manager, stdio + else: + return None _tools = create_mcp_tools(client) except Exception as exc: _failed = True @@ -94,7 +112,7 @@ async def get_mcp_tools() -> list[BaseTool] | None: "MCP server unavailable (%s); live Reactome tools are off for this " "process. Checked %s.%s", exc, - server_path, + where, f" Server said: {stderr}" if stderr else "", ) return None @@ -104,10 +122,12 @@ async def get_mcp_tools() -> list[BaseTool] | None: async def shutdown() -> None: - global _manager, _tools + global _manager, _http, _tools if _manager is not None: await _manager.stop() - _manager, _tools = None, None + if _http is not None: + await _http.aclose() + _manager, _http, _tools = None, None, None def _terminate_at_exit() -> None: diff --git a/tests/reactome_mcp/test_http_client.py b/tests/reactome_mcp/test_http_client.py new file mode 100644 index 0000000..f95209a --- /dev/null +++ b/tests/reactome_mcp/test_http_client.py @@ -0,0 +1,203 @@ +"""The HTTP transport, against a fake server. + +This exists because the stdio client cannot work where the chatbot runs. The +deployed image is Python, has no `node`, and does not mount reactome-mcp, so +`REACTOME_MCP_SERVER` can never be satisfied in the container -- the live +destination worked on every machine it was tested on and none that it ships to. + +Two protocol details are pinned here because both are easy to get wrong and +neither fails in a way that points at the cause: the session id arrives in a +response header, and replies come back SSE-framed rather than as bare JSON. +""" + +import asyncio +import json +from typing import Any + +import httpx +import pytest + +from reactome_mcp.client import MCPToolError +from reactome_mcp.http_client import MCPHttpClient + + +def sse(payload: dict[str, Any]) -> str: + """How the server actually frames a reply to a POST.""" + return f"event: message\ndata: {json.dumps(payload)}\n\n" + + +def _client(handler: Any, url: str = "http://mcp.test") -> MCPHttpClient: + client = MCPHttpClient(url) + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + return client + + +def test_the_session_id_is_read_from_the_header_and_resent() -> None: + """Omit it on later requests and the server answers 400, not a hint.""" + seen: list[str | None] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.headers.get("mcp-session-id")) + body = json.loads(request.content) + if body.get("method") == "initialize": + return httpx.Response( + 200, + headers={ + "content-type": "text/event-stream", + "mcp-session-id": "abc-123", + }, + text=sse( + { + "jsonrpc": "2.0", + "id": 1, + "result": {"serverInfo": {"name": "reactome"}}, + } + ), + ) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + text=sse( + { + "jsonrpc": "2.0", + "id": 2, + "result": {"content": [{"type": "text", "text": "96"}]}, + } + ), + ) + + client = _client(handler) + assert asyncio.run(client.call_tool("reactome_species")) == "96" + + # initialize carries none; everything after carries the issued id. + assert seen[0] is None + assert all(s == "abc-123" for s in seen[1:]), seen + + +def test_an_sse_framed_reply_is_parsed() -> None: + """Parsing the body as JSON gets a decode error on text that is valid.""" + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + result: dict[str, Any] = ( + {"serverInfo": {"name": "reactome"}} + if body.get("method") == "initialize" + else {"content": [{"type": "text", "text": "hello"}]} + ) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream", "mcp-session-id": "s"}, + text=sse({"jsonrpc": "2.0", "id": 1, "result": result}), + ) + + assert asyncio.run(_client(handler).call_tool("x")) == "hello" + + +def test_a_plain_json_reply_is_also_accepted() -> None: + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + result: dict[str, Any] = ( + {"serverInfo": {}} + if body.get("method") == "initialize" + else {"content": [{"type": "text", "text": "json"}]} + ) + return httpx.Response( + 200, + headers={"content-type": "application/json", "mcp-session-id": "s"}, + json={"jsonrpc": "2.0", "id": 1, "result": result}, + ) + + assert asyncio.run(_client(handler).call_tool("x")) == "json" + + +def test_a_missing_session_id_is_refused_up_front() -> None: + """Rather than sending every later request to be rejected one at a time.""" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + text=sse({"jsonrpc": "2.0", "id": 1, "result": {}}), + ) + + with pytest.raises(MCPToolError, match="no mcp-session-id"): + asyncio.run(_client(handler).initialize()) + + +def test_a_jsonrpc_error_is_raised() -> None: + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + if body.get("method") == "initialize": + return httpx.Response( + 200, + headers={"content-type": "text/event-stream", "mcp-session-id": "s"}, + text=sse({"jsonrpc": "2.0", "id": 1, "result": {}}), + ) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + text=sse( + { + "jsonrpc": "2.0", + "id": 2, + "error": {"code": -32601, "message": "no such tool"}, + } + ), + ) + + with pytest.raises(MCPToolError, match="no such tool"): + asyncio.run(_client(handler).call_tool("nope")) + + +def test_a_tool_error_result_raises() -> None: + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + result = ( + {"serverInfo": {}} + if body.get("method") == "initialize" + else { + "isError": True, + "content": [{"type": "text", "text": "pathway not found"}], + } + ) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream", "mcp-session-id": "s"}, + text=sse({"jsonrpc": "2.0", "id": 1, "result": result}), + ) + + with pytest.raises(MCPToolError, match="pathway not found"): + asyncio.run(_client(handler).call_tool("reactome_get_pathway", {"id": "nope"})) + + +def test_an_http_failure_surfaces() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503, text="upstream is down") + + with pytest.raises(httpx.HTTPStatusError): + asyncio.run(_client(handler).initialize()) + + +def test_the_last_data_line_wins() -> None: + """A stream may carry progress notifications before the reply.""" + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + if body.get("method") == "initialize": + return httpx.Response( + 200, + headers={"content-type": "text/event-stream", "mcp-session-id": "s"}, + text=sse({"jsonrpc": "2.0", "id": 1, "result": {}}), + ) + stream = sse({"jsonrpc": "2.0", "method": "notifications/progress"}) + sse( + { + "jsonrpc": "2.0", + "id": 2, + "result": {"content": [{"type": "text", "text": "the answer"}]}, + } + ) + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, text=stream + ) + + assert asyncio.run(_client(handler).call_tool("x")) == "the answer"