jeevesagent.mcp

Model Context Protocol (MCP) integration.

• MCPServerSpec — declarative description of an MCP server (transport + connection details).

• MCPClient — wraps a single mcp.ClientSession. The mcp SDK is imported lazily inside MCPClient.connect() so the module loads without the mcp extra installed; the import fires only when actually connecting to a real server.

• MCPRegistry — implements ToolHost over N MCP servers. Connects all servers in parallel through an anyio.create_task_group(), builds a tool name index with auto-disambiguation, and routes call(tool, args) to the right session.

Submodules

• jeevesagent.mcp.client
• jeevesagent.mcp.registry
• jeevesagent.mcp.spec

Classes

MCPClient	One client per MCP server. Holds the live ClientSession.
MCPRegistry	Aggregates many MCPClient instances into a single ToolHost.
MCPServerSpec	How to find and talk to a single MCP server.

Package Contents

class jeevesagent.mcp.MCPClient(spec: jeevesagent.mcp.spec.MCPServerSpec, *, session: Any | None = None)

One client per MCP server. Holds the live ClientSession.

async aclose() → None

Tear down the session and underlying transport.

async call_tool(name: str, args: dict[str, Any]) → Any

Invoke name with args. Returns the SDK’s CallToolResult.

async connect() → None

Open the transport and initialise the session.

No-op if already connected (or a fake session was injected at construction time).

async list_tools() → list[Any]

Return whatever the SDK gave us — a list of tool descriptors.

Each descriptor has name, description, inputSchema. We don’t translate to ToolDef here — the registry does that, since it also assigns names with disambiguation.

property is_connected: bool

property name: str

property spec: jeevesagent.mcp.spec.MCPServerSpec

class jeevesagent.mcp.MCPRegistry(items: list[jeevesagent.mcp.spec.MCPServerSpec | jeevesagent.mcp.client.MCPClient] | None = None)

Aggregates many MCPClient instances into a single ToolHost.

async aclose() → None

async call(tool: str, args: collections.abc.Mapping[str, Any], *, call_id: str = '') → jeevesagent.core.types.ToolResult

async connect() → None

Connect every client in parallel and rebuild the index.

async list_tools(*, query: str | None = None) → list[jeevesagent.core.types.ToolDef]

async refresh() → None

Re-pull tool lists from every client and rebuild the index.

async watch() → collections.abc.AsyncIterator[jeevesagent.core.types.ToolEvent]

listChanged notifications. Not yet implemented; yields nothing.

property server_names: list[str]

class jeevesagent.mcp.MCPServerSpec

How to find and talk to a single MCP server.

Construct via the class methods stdio() or http() rather than the bare constructor — they enforce the right combination of fields per transport.

classmethod http(name: str, url: str, headers: dict[str, str] | None = None, *, description: str = '') → MCPServerSpec

Connect to url via Streamable HTTP transport.

classmethod stdio(name: str, command: str, args: list[str] | tuple[str, Ellipsis] | None = None, env: dict[str, str] | None = None, *, description: str = '') → MCPServerSpec

Spawn command as a subprocess and speak JSON-RPC over its stdio.

args: tuple[str, Ellipsis] = ()

command: str | None = None

description: str = ''

env: tuple[tuple[str, str], Ellipsis] = ()

headers: tuple[tuple[str, str], Ellipsis] = ()

name: str

transport: Literal['stdio', 'http']

url: str | None = None