From c02219ad480e7e08855587315d08d7ca81b6bd8a Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Tue, 15 Sep 2026 15:27:05 +0000 Subject: [PATCH] Talk to the MCP over HTTP, because stdio cannot work where this runs The live destination spawns `node` and speaks over a pipe. That works on a developer's machine and cannot work in the deployed container: $ docker exec biochat_beta_guest sh -c 'command -v node || echo NO NODE' NO NODE $ docker inspect ... --format '{{range .HostConfig.Binds}}...' /home/awright/git/reactome_chatbot/config.yml:/app/config.yml /home/awright/git/reactome_chatbot/embeddings:/app/embeddings The image is Python, it has no node, and it does not mount reactome-mcp. `REACTOME_MCP_SERVER` can never be satisfied there. The feature worked on every machine it was tested on and none that it ships to -- the same shape as the streaming bug fixed yesterday, found the same way: by asking where it actually runs rather than where it was written. reactome-mcp gained Streamable HTTP, which is what a sibling container or a hosted instance offers. `REACTOME_MCP_URL` uses it, and wins over `REACTOME_MCP_SERVER` when both are set. stdio stays for local development, where the repo and node both exist. Two protocol details are pinned by tests, because both are easy to get wrong and neither fails in a way that points at the cause: - the session id arrives in the `mcp-session-id` **response header** of initialize and must be resent on every later request; omit it and the server answers 400. - replies come back **SSE-framed** (`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. The last data line wins, since a stream may carry progress notifications first. `create_mcp_tools` already took a `ToolCaller` Protocol, so the HTTP client drops in beside the stdio one with no change to the tools or the graph. Verified against a real server, end to end and through the graph: tools over HTTP : 5 species : **Total:** 96 answer : Reactome covers a total of 96 species... streamed : 46 tokens Eight tests against a mock transport, and the full suite passes. Co-Authored-By: Claude Opus 5 --- src/reactome_mcp/README.md | 28 ++++ src/reactome_mcp/http_client.py | 178 ++++++++++++++++++++++ src/reactome_mcp/session.py | 54 ++++--- tests/reactome_mcp/test_http_client.py | 203 +++++++++++++++++++++++++ 4 files changed, 446 insertions(+), 17 deletions(-) create mode 100644 src/reactome_mcp/http_client.py create mode 100644 tests/reactome_mcp/test_http_client.py 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"