Skip to main content

MCPConnectionError

class MCPConnectionError(Exception):
Raised when MCP connection fails.

MCPToolError

class MCPToolError(Exception):
Raised when MCP tool execution fails.

ensure_strict_json_schema

def ensure_strict_json_schema(schema: dict[str, Any]):
Mutates the given JSON schema to ensure it conforms to the strict standard that the OpenAI API expects.

_ensure_strict_json_schema

def _ensure_strict_json_schema(json_schema: object):

resolve_ref

def resolve_ref():

is_dict

def is_dict(obj: object):

is_list

def is_list(obj: object):

has_more_than_n_keys

def has_more_than_n_keys(obj: dict[str, object], n: int):

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. Strategy:
  • Ensure parameters exist with at least an empty properties object (OpenAI requirement).
  • Try converting parameters to strict using ensure_strict_json_schema.
  • If conversion fails, mark function.strict = False and keep best-effort parameters.

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.