calute.mcp.client

calute.mcp.client#

MCP client implementation for connecting to MCP servers.

This module provides the core MCP client functionality for Calute, including: - Connection management for multiple transport types (STDIO, SSE, Streamable HTTP) - JSON-RPC 2.0 message handling for the MCP protocol - Tool discovery and invocation - Resource reading and prompt fetching - Session lifecycle management

The client supports both subprocess-based (STDIO) communication for local MCP servers and HTTP-based transports for remote servers. SSE and Streamable HTTP transports require the optional mcp SDK package.

class calute.mcp.client.MCPClient(config: MCPServerConfig)[source]#

Bases: object

Client for connecting to and interacting with MCP servers.

This client supports multiple transports for communicating with MCP servers: - STDIO: Local subprocess communication (for npx, uvx style servers) - SSE: Server-Sent Events over HTTP (legacy 2024-11-05 protocol) - STREAMABLE_HTTP: Streamable HTTP transport (recommended for 2025+)

Note: SSE and STREAMABLE_HTTP transports require the optional mcp package. Install with: pip install calute[mcp]

config#

MCP server configuration

process#

Subprocess for stdio transport (if applicable)

session_id#

Session identifier for this connection

tools#

Available tools from the server

resources#

Available resources from the server

prompts#

Available prompts from the server

async call_tool(tool_name: str, arguments: dict[str, Any]) → Any[source]#

Call a tool on the MCP server.

Invokes a tool by name with the provided arguments. Uses the SDK session for SSE/Streamable HTTP transports or JSON-RPC for STDIO transport.

Parameters

  • tool_name – Name of the tool to call.

  • arguments – Dictionary of arguments to pass to the tool.

Returns

Tool execution result content.

Raises

RuntimeError – If not connected to the server or tool call fails.

async connect() → bool[source]#

Connect to the MCP server.

Establishes a connection using the transport type specified in the configuration. Handles deprecated transport aliases (HTTP -> SSE, WEBSOCKET -> STREAMABLE_HTTP) for backward compatibility.

Returns

True if connection successful, False otherwise.

Note

After successful connection, use disconnect() to clean up resources.

async disconnect() → None[source]#

Disconnect from the MCP server.

Gracefully closes the connection and cleans up resources. For SDK-based transports (SSE, Streamable HTTP), closes the async exit stack. For STDIO transport, terminates the subprocess.

This method is safe to call multiple times.

async get_prompt(name: str, arguments: dict[str, Any] | None = None) → str[source]#

Get a prompt from the MCP server.

Retrieves and renders a prompt template with the provided arguments. Uses the SDK session for SSE/Streamable HTTP transports or JSON-RPC for STDIO transport.

Parameters

  • name – Name of the prompt to retrieve.

  • arguments – Optional dictionary of arguments for prompt rendering.

Returns

Rendered prompt text as a string.

Raises

RuntimeError – If not connected to the server or prompt fetch fails.

async read_resource(uri: str) → Any[source]#

Read a resource from the MCP server.

Fetches the content of a resource identified by its URI. Uses the SDK session for SSE/Streamable HTTP transports or JSON-RPC for STDIO transport.

Parameters

uri – Resource URI to read.

Returns

Resource content as returned by the server.

Raises

RuntimeError – If not connected to the server or resource read fails.

Contents