From 3fcfaee02ded7e0cba4c98994b861195e903d104 Mon Sep 17 00:00:00 2001 From: CoreRasurae Date: Mon, 11 May 2026 16:31:26 +0000 Subject: [PATCH] feat(mcp): implement StdioMCPTransport for JSON-RPC 2.0 stdio communication Implements concrete StdioMCPTransport class that: - Spawns MCP server as subprocess and communicates via JSON-RPC 2.0 over stdio - Performs MCP handshake (initialize + notifications/initialized) - Supports tools/list and tools/call methods - Uses RLock for thread-safe concurrent access - Auto-selected when transport='stdio' in MCPServerConfig Adds BDD tests for stdio transport covering: - Connection lifecycle and error handling - Tool discovery and invocation - MCPToolAdapter integration ISSUES CLOSED: #4918 --- features/mocks/mcp_stub_server_stdio.py | 181 ++++++++++++++ features/steps/tdd_stdio_transport_steps.py | 242 +++++++++++++++++++ features/tdd_stdio_transport.feature | 123 ++++++++++ src/cleveragents/mcp/__init__.py | 6 + src/cleveragents/mcp/adapter.py | 13 +- src/cleveragents/mcp/stdio_transport.py | 252 ++++++++++++++++++++ 6 files changed, 816 insertions(+), 1 deletion(-) create mode 100755 features/mocks/mcp_stub_server_stdio.py create mode 100644 features/steps/tdd_stdio_transport_steps.py create mode 100644 features/tdd_stdio_transport.feature create mode 100644 src/cleveragents/mcp/stdio_transport.py diff --git a/features/mocks/mcp_stub_server_stdio.py b/features/mocks/mcp_stub_server_stdio.py new file mode 100755 index 000000000..cf5d4aef1 --- /dev/null +++ b/features/mocks/mcp_stub_server_stdio.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""MCP stub server for integration testing of StdioMCPTransport. + +Implements the MCP JSON-RPC 2.0 stdio protocol for deterministic testing. +This server speaks MCP 2024-11-05 protocol version. +""" + +from __future__ import annotations + +import json +import sys +import threading +import time +from typing import Any + +MCP_PROTOCOL_VERSION = "2024-11-05" + +TOOLS = [ + { + "name": "test/echo", + "description": "Echoes back the input arguments", + "inputSchema": { + "type": "object", + "properties": { + "message": {"type": "string", "description": "Message to echo"}, + }, + "required": ["message"], + }, + }, + { + "name": "test/multiply", + "description": "Multiplies two numbers", + "inputSchema": { + "type": "object", + "properties": { + "a": {"type": "number", "description": "First number"}, + "b": {"type": "number", "description": "Second number"}, + }, + "required": ["a", "b"], + }, + }, + { + "name": "test/slow", + "description": "A slow tool that takes time to respond", + "inputSchema": { + "type": "object", + "properties": { + "delay": {"type": "number", "description": "Delay in seconds"}, + }, + }, + }, +] + +request_id = 0 +capabilities: dict[str, Any] = {} +initialized = False +slow_tool_delay = 0.0 +lock = threading.Lock() + + +def send_response(resp_id: int | str | None, result: Any) -> None: + response = { + "jsonrpc": "2.0", + "id": resp_id, + "result": result, + } + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + + +def send_error(resp_id: int | str | None, code: int, message: str) -> None: + response = { + "jsonrpc": "2.0", + "id": resp_id, + "error": {"code": code, "message": message}, + } + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + + +def handle_initialize(params: dict[str, Any]) -> dict[str, Any]: + global capabilities, initialized + capabilities = params.get("capabilities", {}) + initialized = True + return { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": {"tools": True, "resources": {}}, + "serverInfo": { + "name": "test-mcp-stdio-server", + "version": "1.0.0", + }, + } + + +def handle_tools_list(params: dict[str, Any]) -> dict[str, Any]: + return {"tools": TOOLS} + + +def handle_tools_call(req_id: int | str | None, params: dict[str, Any]) -> None: + global slow_tool_delay + + tool_name = params.get("name", "") + arguments = params.get("arguments", {}) + + if tool_name == "test/echo": + message = arguments.get("message", "") + send_response( + req_id, + {"content": [{"type": "text", "text": json.dumps({"echoed": message})}]}, + ) + return + + if tool_name == "test/multiply": + a = arguments.get("a", 0) + b = arguments.get("b", 0) + send_response( + req_id, + {"content": [{"type": "text", "text": json.dumps({"result": a * b})}]}, + ) + return + + if tool_name == "test/slow": + delay = arguments.get("delay", 0.1) + if delay > slow_tool_delay: + time.sleep(delay) + send_response( + req_id, + {"content": [{"type": "text", "text": json.dumps({"delayed": delay})}]}, + ) + return + + send_error(req_id, -32602, f"Unknown tool: {tool_name}") + + +def handle_request(request: dict[str, Any]) -> None: + global request_id, slow_tool_delay + + method = request.get("method", "") + params = request.get("params", {}) + req_id = request.get("id") + + if method == "initialize": + result = handle_initialize(params) + send_response(req_id, result) + elif method == "notifications/initialized": + pass + elif method == "tools/list": + if not initialized: + send_error(req_id, -32602, "Not initialized") + return + result = handle_tools_list(params) + send_response(req_id, result) + elif method == "tools/call": + if not initialized: + send_error(req_id, -32602, "Not initialized") + return + handle_tools_call(req_id, params) + elif method == "test/set_slow_delay": + slow_tool_delay = params.get("delay", 0.0) + send_response(req_id, {"ok": True}) + else: + send_error(req_id, -32601, f"Unknown method: {method}") + + +def main() -> None: + for line in sys.stdin: + line = line.strip() + if not line: + continue + + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + + if isinstance(request, dict) and request.get("jsonrpc") == "2.0": + handle_request(request) + + +if __name__ == "__main__": + main() diff --git a/features/steps/tdd_stdio_transport_steps.py b/features/steps/tdd_stdio_transport_steps.py new file mode 100644 index 000000000..7be7581c6 --- /dev/null +++ b/features/steps/tdd_stdio_transport_steps.py @@ -0,0 +1,242 @@ +"""Step definitions for features/tdd_stdio_transport.feature.""" + +from __future__ import annotations + +import json +import os +import sys + + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.mcp.adapter import MCPServerConfig +from cleveragents.mcp.stdio_transport import StdioMCPTransport + + +def _get_stdio_server_path() -> str: + return os.path.join( + os.path.dirname(__file__), + "..", + "mocks", + "mcp_stub_server_stdio.py", + ) + + +@given("the MCP stdio stub server is available") +def step_stub_server_available(context: Context) -> None: + server_path = _get_stdio_server_path() + if not os.path.exists(server_path): + pytest_path = sys.executable + msg = ( + f"MCP stdio stub server not found at {server_path} (python: {pytest_path})" + ) + raise RuntimeError(msg) + context.mcp_stdio_server_path = server_path + + +@given("an MCP server config using the stdio stub server") +def step_config_with_stub_server(context: Context) -> None: + server_path = _get_stdio_server_path() + context.mcp_config = MCPServerConfig( + name="test-stdio-server", + transport="stdio", + command=sys.executable, + args=[server_path], + ) + + +@given("an MCP server config with stdio transport and no command") +def step_config_stdio_no_command(context: Context) -> None: + context.mcp_config = MCPServerConfig( + name="bad-server", + transport="stdio", + ) + + +@given("an MCP server config with stdio transport and command {command}") +def step_config_stdio_bad_command(context: Context, command: str) -> None: + context.mcp_config = MCPServerConfig( + name="bad-server", + transport="stdio", + command=command, + ) + + +@given("the config command is set to the stdio stub server path") +def step_config_command_set(context: Context) -> None: + server_path = _get_stdio_server_path() + context.mcp_config = MCPServerConfig( + name="test-server", + transport="stdio", + command=sys.executable, + args=[server_path], + ) + + +@given("a connected stdio transport") +def step_connected_stdio_transport(context: Context) -> None: + server_path = _get_stdio_server_path() + config = MCPServerConfig( + name="test-stdio-server", + transport="stdio", + command=sys.executable, + args=[server_path], + ) + context.mcp_stdio_transport = StdioMCPTransport() + context.mcp_stdio_capabilities = context.mcp_stdio_transport.connect(config) + + +@when("I create a stdio transport") +def step_create_stdio_transport(context: Context) -> None: + context.mcp_stdio_transport = StdioMCPTransport() + + +@when("I connect the transport") +def step_connect_transport(context: Context) -> None: + context.mcp_stdio_error = None + try: + context.mcp_stdio_capabilities = context.mcp_stdio_transport.connect( + context.mcp_config + ) + except Exception as exc: + context.mcp_stdio_error = str(exc) + + +@when("I create a stdio transport and connect") +def step_create_and_connect_transport(context: Context) -> None: + context.mcp_stdio_error = None + try: + transport = StdioMCPTransport() + context.mcp_stdio_capabilities = transport.connect(context.mcp_config) + context.mcp_stdio_transport = transport + except Exception as exc: + context.mcp_stdio_error = str(exc) + + +@when("I connect the transport again") +def step_reconnect_transport(context: Context) -> None: + context.mcp_stdio_capabilities = context.mcp_stdio_transport.connect( + context.mcp_config + ) + + +@when("I close the transport") +def step_close_transport(context: Context) -> None: + context.mcp_stdio_transport.close() + + +@when('I call "{method}" on the transport') +def step_call_method(context: Context, method: str) -> None: + context.mcp_stdio_call_error = None + context.mcp_stdio_response = None + try: + context.mcp_stdio_response = context.mcp_stdio_transport.call(method, {}) + except Exception as exc: + context.mcp_stdio_call_error = str(exc) + + +@when('I call "{method}" with arguments {args_json}') +def step_call_method_with_args(context: Context, method: str, args_json: str) -> None: + arguments = json.loads(args_json) + context.mcp_stdio_call_error = None + context.mcp_stdio_response = None + try: + context.mcp_stdio_response = context.mcp_stdio_transport.call(method, arguments) + except Exception as exc: + context.mcp_stdio_call_error = str(exc) + + +@when('I send a "{method}" notification with params {params_json}') +def step_send_notification(context: Context, method: str, params_json: str) -> None: + params = json.loads(params_json) + context.mcp_stdio_transport._send_notification(method, params) + + +# ------------------------------------------------------------------- +# Then Steps +# ------------------------------------------------------------------- + + +@then("the MCP stdio transport should be connected") +def step_mcp_stdio_transport_connected(context: Context) -> None: + assert context.mcp_stdio_transport is not None + assert context.mcp_stdio_transport._connected + + +@then("the MCP stdio transport should not be connected") +def step_mcp_stdio_transport_not_connected(context: Context) -> None: + assert context.mcp_stdio_transport is not None + assert not context.mcp_stdio_transport._connected + + +@then("the server capabilities should include tools support") +def step_capabilities_have_tools(context: Context) -> None: + caps = context.mcp_stdio_capabilities + assert caps is not None + assert "capabilities" in caps + assert caps["capabilities"].get("tools") is not None + + +@then("the connection should fail with error containing {text}") +def step_connection_error_contains(context: Context, text: str) -> None: + assert context.mcp_stdio_error is not None + text = text.strip('"') + assert text.lower() in context.mcp_stdio_error.lower(), ( + f"Expected error containing '{text}', got: {context.mcp_stdio_error}" + ) + + +@then('the response should contain "{key}" key') +def step_response_contains_key(context: Context, key: str) -> None: + assert context.mcp_stdio_response is not None + assert key in context.mcp_stdio_response, ( + f"Key '{key}' not in response: {context.mcp_stdio_response}" + ) + + +@then("the tools list should have {count:d} tools") +def step_tools_list_count(context: Context, count: int) -> None: + tools = context.mcp_stdio_response.get("tools", []) + assert len(tools) == count, f"Expected {count} tools, got {len(tools)}: {tools}" + + +@then('tool {idx:d} should have name "{name}"') +def step_tool_has_name(context: Context, idx: int, name: str) -> None: + tools = context.mcp_stdio_response.get("tools", []) + assert tools[idx]["name"] == name + + +@then("the call should succeed") +def step_call_succeeded(context: Context) -> None: + assert context.mcp_stdio_call_error is None + assert context.mcp_stdio_response is not None + + +@then("the call should fail with error containing {text}") +def step_call_failed(context: Context, text: str) -> None: + text = text.strip('"') + error_msg = context.mcp_stdio_call_error or "" + assert text.lower() in error_msg.lower(), ( + f"Expected error containing '{text}', got: {error_msg}" + ) + + +@then('the response content should contain "{text}"') +def step_response_content_contains(context: Context, text: str) -> None: + content = context.mcp_stdio_response.get("content", []) + content_text = json.dumps(content) + assert text in content_text, f"Expected '{text}' in content, got: {content_text}" + + +@then("the adapter transport type should be {transport}") +def step_adapter_transport_type(context: Context, transport: str) -> None: + actual = context.mcp_adapter.transport_type + transport = transport.strip('"') + assert actual == transport, f"Expected transport type '{transport}', got '{actual}'" + + +@then("the MCP stdio transport should still be connected") +def step_mcp_stdio_transport_still_connected(context: Context) -> None: + assert context.mcp_stdio_transport is not None + assert context.mcp_stdio_transport._connected diff --git a/features/tdd_stdio_transport.feature b/features/tdd_stdio_transport.feature new file mode 100644 index 000000000..11c6b9191 --- /dev/null +++ b/features/tdd_stdio_transport.feature @@ -0,0 +1,123 @@ +Feature: Stdio MCP Transport + As an MCP client using the stdio transport + I want to connect to MCP servers via subprocess stdio + So that I can discover and invoke tools from real MCP servers + + Background: + Given the MCP stdio stub server is available + + # ------------------------------------------------------------------- + # Connection Lifecycle + # ------------------------------------------------------------------- + + Scenario: Connect to MCP stdio server performs handshake + Given an MCP server config using the stdio stub server + When I create a stdio transport + And I connect the transport + Then the MCP stdio transport should be connected + And the server capabilities should include tools support + + Scenario: Connect with missing command raises error + Given an MCP server config with stdio transport and no command + When I create a stdio transport and connect + Then the connection should fail with error containing "command" + + Scenario: Connect with non-existent command raises error + Given an MCP server config with stdio transport and command "/nonexistent/binary" + When I create a stdio transport and connect + Then the connection should fail with error containing "not found" + + Scenario: Double connect is idempotent + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I connect the transport again + Then the MCP stdio transport should be connected + + Scenario: Close transport terminates subprocess + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I close the transport + Then the MCP stdio transport should not be connected + + # ------------------------------------------------------------------- + # Tool Discovery via Transport + # ------------------------------------------------------------------- + + Scenario: Discover tools via stdio transport + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I call "tools/list" on the transport + Then the response should contain "tools" key + And the tools list should have 3 tools + And tool 0 should have name "test/echo" + + Scenario: Call unknown method raises error + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I call "nonexistent/method" on the transport + Then the call should fail with error containing "Unknown method" + + # ------------------------------------------------------------------- + # Tool Invocation via Transport + # ------------------------------------------------------------------- + + Scenario: Invoke test/echo tool via stdio transport + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I call "tools/call" with arguments {"name": "test/echo", "arguments": {"message": "hello"}} + Then the call should succeed + And the response should contain "content" key + + Scenario: Invoke test/multiply tool via stdio transport + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I call "tools/call" with arguments {"name": "test/multiply", "arguments": {"a": 6, "b": 7}} + Then the call should succeed + And the response content should contain "42" + + Scenario: Invoke unknown tool returns error response + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I call "tools/call" with arguments {"name": "test/nonexistent", "arguments": {}} + Then the call should fail with error containing "Unknown tool" + + # ------------------------------------------------------------------- + # MCPToolAdapter Integration with Stdio Transport + # ------------------------------------------------------------------- + + Scenario: MCPToolAdapter uses StdioMCPTransport automatically for stdio config + Given an MCP server config for "test-server" with stdio transport + And the config command is set to the stdio stub server path + When I create an MCP adapter from the config + And I connect the adapter + Then the adapter transport type should be "stdio" + And the adapter should be connected + And the adapter server capabilities should be set + + Scenario: MCPToolAdapter discovers tools via StdioMCPTransport + Given an MCP server config for "test-server" with stdio transport + And the config command is set to the stdio stub server path + When I create an MCP adapter from the config + And I connect the adapter + And I discover tools from the adapter + Then the adapter should have 3 discovered tools + + Scenario: MCPToolAdapter invokes tools via StdioMCPTransport + Given an MCP server config for "test-server" with stdio transport + And the config command is set to the stdio stub server path + When I create an MCP adapter from the config + And I connect the adapter + And I discover tools from the adapter + And I invoke "test/echo" with arguments {"message": "test"} + Then the invocation should succeed + And the invocation result should contain key "content" + + # ------------------------------------------------------------------- + # Notifications + # ------------------------------------------------------------------- + + Scenario: Send notification to server + Given an MCP server config using the stdio stub server + And a connected stdio transport + When I send a "test/custom_notification" notification with params {} + Then the MCP stdio transport should still be connected \ No newline at end of file diff --git a/src/cleveragents/mcp/__init__.py b/src/cleveragents/mcp/__init__.py index d37286b19..0d67fcd8d 100644 --- a/src/cleveragents/mcp/__init__.py +++ b/src/cleveragents/mcp/__init__.py @@ -14,12 +14,15 @@ from cleveragents.mcp.adapter import ( MCPServerConfig, MCPToolAdapter, MCPToolDescriptor, + MCPToolFilter, MCPToolResult, + MCPTransport, ) from cleveragents.mcp.client import McpClient, McpClientConfig, McpClientState from cleveragents.mcp.refresh_hook import MCPRefreshHook from cleveragents.mcp.registry import McpRegistry from cleveragents.mcp.sandbox import SandboxPathRewriter, SandboxPathRewriterConfig +from cleveragents.mcp.stdio_transport import StdioMCPTransport __all__ = [ "MCPCapabilityMetadata", @@ -27,11 +30,14 @@ __all__ = [ "MCPServerConfig", "MCPToolAdapter", "MCPToolDescriptor", + "MCPToolFilter", "MCPToolResult", + "MCPTransport", "McpClient", "McpClientConfig", "McpClientState", "McpRegistry", "SandboxPathRewriter", "SandboxPathRewriterConfig", + "StdioMCPTransport", ] diff --git a/src/cleveragents/mcp/adapter.py b/src/cleveragents/mcp/adapter.py index 7beda1e14..71c10f55d 100644 --- a/src/cleveragents/mcp/adapter.py +++ b/src/cleveragents/mcp/adapter.py @@ -48,6 +48,12 @@ __all__ = [ ] +def _create_stdio_transport() -> MCPTransport: + from cleveragents.mcp.stdio_transport import StdioMCPTransport + + return StdioMCPTransport() + + class MCPServerConfig(BaseModel): """Configuration for connecting to an MCP server. @@ -196,7 +202,12 @@ class MCPToolAdapter: ) -> None: self._validate_config(config) self._config = config - self._transport = transport or MCPTransport() + if transport is not None: + self._transport = transport + elif config.transport == "stdio": + self._transport = _create_stdio_transport() + else: + self._transport = MCPTransport() self._connected = False self._capabilities: dict[str, Any] = {} self._tools: dict[str, MCPToolDescriptor] = {} diff --git a/src/cleveragents/mcp/stdio_transport.py b/src/cleveragents/mcp/stdio_transport.py new file mode 100644 index 000000000..848ae72cd --- /dev/null +++ b/src/cleveragents/mcp/stdio_transport.py @@ -0,0 +1,252 @@ +"""Stdio transport layer for MCP server communication. + +Implements the MCP stdio transport protocol (JSON-RPC 2.0 over stdio) +as specified in the MCP specification version 2024-11-05. +""" + +from __future__ import annotations + +import json +import logging +import subprocess +import threading +import time +from typing import Any + +from cleveragents.mcp.adapter import MCPServerConfig, MCPTransport + +logger = logging.getLogger(__name__) + +MCP_PROTOCOL_VERSION = "2024-11-05" +DEFAULT_TIMEOUT = 30.0 + + +class StdioMCPTransport(MCPTransport): + """Concrete stdio transport for MCP server communication. + + Spawns an MCP server as a subprocess and communicates via JSON-RPC 2.0 + messages over stdio using newline-delimited JSON (NDJSON). + + Thread-safety: uses a lock to serialize concurrent calls to the + subprocess since stdio is not inherently thread-safe. + """ + + def __init__( + self, + timeout: float = DEFAULT_TIMEOUT, + ) -> None: + self._timeout = timeout + self._process: subprocess.Popen[bytes] | None = None + self._lock = threading.RLock() + self._request_id = 0 + self._connected = False + self._capabilities: dict[str, Any] = {} + + def connect(self, config: MCPServerConfig) -> dict[str, Any]: + """Spawn the MCP server subprocess and perform the initialize handshake. + + Parameters + ---------- + config: + Server connection configuration with stdio transport settings. + + Returns + ------- + dict[str, Any] + Server capabilities from the initialize response. + + Raises + ------ + ConnectionError + If the subprocess fails to start or the handshake times out. + """ + with self._lock: + if self._connected: + return self._capabilities + + if config.command is None: + msg = "stdio transport requires 'command' field" + raise ConnectionError(msg) + + env = dict(config.env) if config.env else {} + env["NO_COLOR"] = "1" + + try: + self._process = subprocess.Popen( + [config.command, *config.args], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + bufsize=0, + ) + except FileNotFoundError as exc: + msg = f"Command not found: {config.command}" + raise ConnectionError(msg) from exc + except PermissionError as exc: + msg = f"Permission denied: {config.command}" + raise ConnectionError(msg) from exc + + self._request_id = 0 + capabilities = self._send_request_sync( + "initialize", + { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": {"roots": {"listChanged": True}}, + "clientInfo": { + "name": "cleveragents-mcp-adapter", + "version": "1.0.0", + }, + }, + ) + + if capabilities is None: + self._terminate() + msg = "Server did not return capabilities during handshake" + raise ConnectionError(msg) + + self._capabilities = capabilities.get("capabilities", {}) + self._connected = True + + self._send_notification("notifications/initialized", {}) + + logger.info("MCP stdio transport connected to '%s'", config.name) + return capabilities + + def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + """Send a JSON-RPC method call and return the result. + + Parameters + ---------- + method: + The JSON-RPC method name. + params: + Method parameters. + + Returns + ------- + dict[str, Any] + The JSON-RPC response result. + + Raises + ------ + TimeoutError + If the response times out. + ConnectionError + If the transport is not connected. + """ + with self._lock: + if not self._connected or self._process is None: + msg = "Transport not connected. Call connect() first." + raise ConnectionError(msg) + + result = self._send_request_sync(method, params) + if result is None: + msg = f"Server did not respond to {method} request" + raise ConnectionError(msg) + return result + + def close(self) -> None: + """Shut down the transport connection and terminate the subprocess.""" + with self._lock: + if not self._connected: + return + self._terminate() + logger.info("MCP stdio transport closed") + + def _terminate(self) -> None: + """Terminate the subprocess and reset state.""" + self._connected = False + self._capabilities = {} + if self._process is not None: + try: + self._process.terminate() + try: + self._process.wait(timeout=5.0) + except subprocess.TimeoutExpired: + self._process.kill() + self._process.wait() + except Exception: + pass + self._process = None + + def _send_request_sync( + self, + method: str, + params: dict[str, Any], + ) -> dict[str, Any] | None: + """Send a JSON-RPC request synchronously and wait for response. + + Must be called with the lock held. + """ + if self._process is None: + return None + + self._request_id += 1 + req_id = self._request_id + + request = { + "jsonrpc": "2.0", + "id": req_id, + "method": method, + "params": params, + } + + try: + if self._process.stdin is None: + return None + self._process.stdin.write(json.dumps(request).encode("utf-8") + b"\n") + self._process.stdin.flush() + except BrokenPipeError as exc: + msg = "Subprocess stdin pipe broken" + raise ConnectionError(msg) from exc + + start = time.monotonic() + while True: + if self._process.stdout is None: + return None + line = self._process.stdout.readline() + if not line: + return None + + if time.monotonic() - start > self._timeout: + msg = f"Response timeout after {self._timeout}s" + raise TimeoutError(msg) + + try: + response = json.loads(line.decode("utf-8")) + except json.JSONDecodeError: + continue + + if not isinstance(response, dict): + continue + + if response.get("id") == req_id: + if "error" in response: + error = response["error"] + msg = error.get("message", "Unknown error") + raise ConnectionError(f"MCP error: {msg}") + return response.get("result") + + def _send_notification(self, method: str, params: dict[str, Any]) -> None: + """Send a JSON-RPC notification (no response expected). + + Must be called with the lock held. + """ + if self._process is None: + return + + notification = { + "jsonrpc": "2.0", + "method": method, + "params": params, + } + + if self._process.stdin is None: + return + + try: + self._process.stdin.write(json.dumps(notification).encode("utf-8") + b"\n") + self._process.stdin.flush() + except BrokenPipeError: + pass