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.
⌘I