​

MCPConnectionError

class MCPConnectionError(Exception):

Raised when MCP connection fails.

​

MCPToolError

class MCPToolError(Exception):

Raised when MCP tool execution fails.

​

MCPToolkit

class MCPToolkit(BaseToolkit):

MCPToolkit provides a unified interface for managing multiple MCP server connections and their tools.

This class handles the lifecycle of multiple MCP server connections and offers a centralized configuration mechanism for both local and remote MCP services. The toolkit manages multiple :obj:MCPClient instances and aggregates their tools into a unified interface compatible with the CAMEL framework.

Connection Lifecycle: There are three ways to manage the connection lifecycle:

1. Using the async context manager (recommended):

.. code-block:: python

async with MCPToolkit(config_path=“config.json”) as toolkit:

​

Toolkit is connected here

tools = toolkit.get_tools()

​

Toolkit is automatically disconnected here

2. Using the factory method:

.. code-block:: python

toolkit = await MCPToolkit.create(config_path=“config.json”)

​

Toolkit is connected here

tools = toolkit.get_tools()

​

Don’t forget to disconnect when done!

await toolkit.disconnect()

3. Using explicit connect/disconnect:

.. code-block:: python

toolkit = MCPToolkit(config_path=“config.json”) await toolkit.connect()

​

Toolkit is connected here

tools = toolkit.get_tools()

​

Don’t forget to disconnect when done!

await toolkit.disconnect()

Parameters:

• clients (Optional[List[MCPClient]], optional): List of :obj:MCPClient instances to manage. (default: :obj:None)
• config_path (Optional[str], optional): Path to a JSON configuration file defining MCP servers. The file should contain server configurations in the standard MCP format. (default: :obj:None)
• config_dict (Optional[Dict[str, Any]], optional): Dictionary containing MCP server configurations in the same format as the config file. This allows for programmatic configuration without file I/O. (default: :obj:None)
• timeout (Optional[float], optional): Timeout for connection attempts in seconds. This timeout applies to individual client connections. (default: :obj:None)
• strict (Optional[bool], optional): Flag to indicate strict mode. (default: :obj:False)
• clients (List[MCPClient]): List of :obj:MCPClient instances being managed by this toolkit.

​

init

def __init__(
    self,
    clients: Optional[List[MCPClient]] = None,
    config_path: Optional[str] = None,
    config_dict: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None,
    strict: Optional[bool] = False
):

​

is_connected

def is_connected(self):

Returns:

bool: True if the toolkit is connected to all MCP servers, False otherwise.

​

connect_sync

def connect_sync(self):

Synchronously connect to all MCP servers.

​

disconnect_sync

def disconnect_sync(self):

Synchronously disconnect from all MCP servers.

​

enter

def __enter__(self):

Synchronously enter the async context manager.

​

exit

def __exit__(
    self,
    exc_type,
    exc_val,
    exc_tb
):

Synchronously exit the async context manager.

​

create_sync

def create_sync(
    cls,
    clients: Optional[List[MCPClient]] = None,
    config_path: Optional[str] = None,
    config_dict: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None,
    strict: Optional[bool] = False
):

Synchronously create and connect to all MCP servers.

​

_load_clients_from_config

def _load_clients_from_config(self, config_path: str):

Load clients from configuration file.

​

_load_clients_from_dict

def _load_clients_from_dict(self, config: Dict[str, Any]):

Load clients from configuration dictionary.

​

_create_client_from_config

def _create_client_from_config(self, name: str, cfg: Dict[str, Any]):

Create a single MCP client from configuration.

​

get_tools

def get_tools(self):

Returns:

List[FunctionTool]: Combined list of all available function tools from all connected MCP servers. Returns an empty list if no clients are connected or if no tools are available.

​

get_text_tools

def get_text_tools(self):

Returns:

str: A string containing the descriptions of all tools.

​

call_tool_sync

def call_tool_sync(self, tool_name: str, tool_args: Dict[str, Any]):

Synchronously call a tool.

​

list_available_tools

def list_available_tools(self):

Returns:

Dict[str, List[str]]: Dictionary mapping client indices to tool names.