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

  1. 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()

  1. 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)
  • 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
):

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
):

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.

_ensure_strict_tool_schema

def _ensure_strict_tool_schema(self, tool: FunctionTool):

Ensure a tool has a strict schema compatible with OpenAI’s requirements.

Parameters:

  • tool (FunctionTool): The tool to check and update if necessary.

Returns:

FunctionTool: The tool with a strict schema.

get_tools

def get_tools(self):

Returns:

List[FunctionTool]: Combined list of all available function tools from all connected MCP servers with strict schemas. 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.