cubepi.mcp

load_mcp_tools_http

function

load_mcp_tools_http(server_url: str, *, headers: dict[str, str] | None = None, timeout: float = 30.0) -> list[AgentTool]

Connect to an HTTP/SSE MCP server, discover tools, return AgentTools.

Uses the mcp SDK's HTTP client. Each returned tool's execute method invokes tools/call against a fresh session — v1 simplicity, no pooling.

The transport's own timeout bounds the SSE connection; we additionally wrap initialize/list/call awaits in asyncio.wait_for so a server that accepts the connection but stalls on protocol messages still aborts.

source

load_mcp_tools_stdio

function

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 wall-clock timeout for initialize/list/call awaits. A hung server raises asyncio.TimeoutError instead of blocking forever.

source