---
id: cubepi-mcp
title: cubepi.mcp
sidebar_position: 5
hide_table_of_contents: false
---

# `cubepi.mcp`

### load_mcp_tools_http

_function_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/mcp/http_loader.py#L12)


### load_mcp_tools_stdio

_function_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/mcp/stdio_loader.py#L12)



<!-- GENERATED by build-api-reference.py — DO NOT EDIT -->