feat(mcp): add stdio MCP tool loader + fake stdio server test (D2.3)

cubepi.mcp.load_mcp_tools_stdio(command, args, env, cwd) spawns an MCP
server subprocess via stdio, lists tools, returns cubepi.AgentTool list.
Each tool's execute opens a fresh subprocess per call (v1 simplicity).

Tested via tests/mcp/_fake_stdio_server.py — a minimal MCP server using
the mcp SDK's own server primitives, runnable as
'python -m tests.mcp._fake_stdio_server'.

Author: xfgong

cubepi/mcp/__init__.py

@@ -3,5 +3,6 @@
 cubepi[mcp] extra required.
 """
 from cubepi.mcp.http_loader import load_mcp_tools_http
+from cubepi.mcp.stdio_loader import load_mcp_tools_stdio
 
-__all__ = ["load_mcp_tools_http"]
+__all__ = ["load_mcp_tools_http", "load_mcp_tools_stdio"]

cubepi/mcp/stdio_loader.py

@@ -0,0 +1,69 @@
+"""stdio transport MCP tool loader."""
+from __future__ import annotations
+
+from typing import Any
+
+from cubepi.agent.types import AgentTool
+from cubepi.mcp._adapter import make_mcp_agent_tool
+
+
+async def load_mcp_tools_stdio(
+    command: str,
+    args: list[str],
+    *,
+    env: dict[str, str] | None = None,
+    cwd: str | None = None,
+    timeout: float = 30.0,
+) -> list[AgentTool]:
+    """Spawn a stdio MCP server subprocess, discover tools, return AgentTools.
+
+    Each returned tool's execute opens a fresh subprocess per call (v1
+    simplicity, no process pooling).
+
+    Args:
+        command: executable to run (e.g. "npx" or sys.executable)
+        args: argv for the server process
+        env: environment variables (passed to subprocess)
+        cwd: working directory for the subprocess
+        timeout: per-call timeout (not currently enforced strictly)
+    """
+    from mcp import ClientSession, StdioServerParameters
+    from mcp.client.stdio import stdio_client
+
+    server_params = StdioServerParameters(
+        command=command, args=args, env=env, cwd=cwd,
+    )
+
+    async def _call_remote(tool_name: str, args_dict: dict[str, Any]) -> dict[str, Any]:
+        async with stdio_client(server_params) as streams:
+            async with ClientSession(*streams) as session:
+                await session.initialize()
+                resp = await session.call_tool(tool_name, args_dict)
+                return _serialize_call_tool_response(resp)
+
+    async with stdio_client(server_params) as streams:
+        async with ClientSession(*streams) as session:
+            await session.initialize()
+            tools_resp = await session.list_tools()
+            tool_descs = tools_resp.tools
+
+    return [
+        make_mcp_agent_tool(
+            name=desc.name,
+            description=desc.description or "",
+            input_schema=desc.inputSchema or {"type": "object", "properties": {}},
+            call_remote=_call_remote,
+        )
+        for desc in tool_descs
+    ]
+
+
+def _serialize_call_tool_response(resp: Any) -> dict[str, Any]:
+    content = []
+    for c in (resp.content or []):
+        if getattr(c, "type", None) == "text":
+            content.append({"type": "text", "text": c.text})
+    return {
+        "content": content,
+        "isError": bool(getattr(resp, "isError", False)),
+    }

tests/mcp/_fake_stdio_server.py

@@ -0,0 +1,54 @@
+"""Minimal stdio MCP server for testing.
+
+Uses the mcp SDK's server primitives. Advertises one 'echo' tool.
+
+Run as: python -m tests.mcp._fake_stdio_server
+"""
+import asyncio
+
+
+async def main() -> None:
+    from mcp.server import NotificationOptions, Server
+    from mcp.server.models import InitializationOptions
+    from mcp.server.stdio import stdio_server
+    from mcp.types import TextContent, Tool
+
+    server = Server("fake-cubepi-test-server")
+
+    @server.list_tools()
+    async def _list_tools() -> list[Tool]:
+        return [
+            Tool(
+                name="echo",
+                description="Echo the input back",
+                inputSchema={
+                    "type": "object",
+                    "properties": {"text": {"type": "string"}},
+                    "required": ["text"],
+                },
+            ),
+        ]
+
+    @server.call_tool()
+    async def _call_tool(name: str, arguments: dict) -> list[TextContent]:
+        if name == "echo":
+            return [TextContent(type="text", text=arguments.get("text", ""))]
+        raise ValueError(f"unknown tool: {name}")
+
+    async with stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="fake-cubepi-test-server",
+                server_version="0.0.1",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/mcp/test_stdio_loader.py

@@ -0,0 +1,31 @@
+"""stdio MCP loader integration tests (D2.3)."""
+
+import sys
+
+import pytest
+
+
+def test_import_stdio_loader() -> None:
+    from cubepi.mcp import load_mcp_tools_stdio
+    assert callable(load_mcp_tools_stdio)
+
+
+@pytest.mark.asyncio
+async def test_stdio_loader_against_fake_server() -> None:
+    """Spawn the fake stdio server, list tools, invoke 'echo'."""
+    from cubepi.mcp import load_mcp_tools_stdio
+
+    tools = await load_mcp_tools_stdio(
+        command=sys.executable,
+        args=["-m", "tests.mcp._fake_stdio_server"],
+    )
+    assert len(tools) == 1
+    echo = tools[0]
+    assert echo.name == "echo"
+
+    args = echo.parameters(text="hello")
+    result = await echo.execute(args)
+    assert len(result.content) == 1
+    from cubepi.providers.base import TextContent
+    assert isinstance(result.content[0], TextContent)
+    assert result.content[0].text == "hello"