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

@@ -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()

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

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

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",
 ]

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] = {}

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