calute.mcp.manager

calute.mcp.manager#

MCP Manager for managing multiple MCP server connections.

This module provides the MCPManager class for orchestrating connections to multiple MCP servers, including: - Connection lifecycle management (add, remove, disconnect) - Unified access to tools, resources, and prompts across all servers - Cross-server tool invocation and resource reading - Capability discovery and summarization

The manager abstracts away the complexity of dealing with multiple MCP servers, providing a single interface for tool and resource access.

class calute.mcp.manager.MCPManager[source]#

Bases: object

Manager for multiple MCP server connections.

Manages connections to multiple MCP servers, provides unified access to tools and resources, and converts MCP tools to Calute functions.

servers#: Dictionary of server name to MCPClient

logger#: Logger instance

async add_server(config: MCPServerConfig) → bool[source]#

Add and connect to an MCP server.

Creates a new MCPClient instance, attempts to connect, and registers it with the manager if successful. Skips servers that are disabled in their configuration.

Parameters: config – Server configuration specifying connection details.
Returns: True if server added successfully, False otherwise.

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

Call an MCP tool by name.

Finds the tool across all connected servers and executes it on the server that provides it. The first matching tool found is used.

Parameters

tool_name – Name of the tool to call.
arguments – Dictionary of arguments to pass to the tool.

Returns

Tool execution result from the MCP server.

Raises

ValueError – If the tool is not found in any connected server.

async disconnect_all() → None[source]#

Disconnect from all MCP servers.

Gracefully closes all active server connections and clears the internal server registry. This method is safe to call even if no servers are connected.

get_all_prompts() → list[calute.mcp.types.MCPPrompt][source]#

Get all prompts from all connected servers.

Aggregates prompts from all connected MCP servers into a single list.

Returns: List of all available MCP prompts across all servers.

get_all_resources() → list[calute.mcp.types.MCPResource][source]#

Get all resources from all connected servers.

Aggregates resources from all connected MCP servers into a single list.

Returns: List of all available MCP resources across all servers.

get_all_tools() → list[calute.mcp.types.MCPTool][source]#

Get all tools from all connected servers.

Aggregates tools from all connected MCP servers into a single list.

Returns: List of all available MCP tools across all servers.

get_capabilities_summary() → dict[str, Any][source]#

Get summary of all capabilities across servers.

Provides an overview of the tools, resources, and prompts available from each connected server, useful for debugging and monitoring.

Returns: Dictionary mapping server names to capability counts, with keys ‘tools’, ‘resources’, and ‘prompts’ for each server.

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

Get a prompt by name.

Searches all connected servers for a prompt matching the given name and retrieves it from the server that provides it.

Parameters

name – Name of the prompt to retrieve.
arguments – Optional dictionary of arguments for prompt rendering.

Returns

Rendered prompt text as a string.

Raises

ValueError – If the prompt is not found in any connected server.

get_server(name: str) → calute.mcp.client.MCPClient | None[source]#

Get MCP server client by name.

Retrieves the MCPClient instance for a specific server, allowing direct access to the server’s capabilities and methods.

Parameters: name – Name of the server to retrieve.
Returns: MCPClient instance or None if not found.

list_servers() → list[str][source]#

Get list of connected server names.

Returns: List of server names currently connected to the manager.

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

Read a resource by URI.

Searches all connected servers for a resource matching the given URI and reads its content from the server that provides it.

Parameters: uri – Resource URI to read.
Returns: Resource content as returned by the MCP server.
Raises: ValueError – If the resource is not found in any connected server.

async remove_server(name: str) → None[source]#

Remove and disconnect from an MCP server.

Disconnects from the specified server and removes it from the manager. If the server is not found, this method does nothing.

Parameters: name – Name of the server to remove.

calute.mcp.manager

Contents

calute.mcp.manager#