Documentation Index
Fetch the complete documentation index at: https://docs.camel-ai.org/llms.txt
Use this file to discover all available pages before exploring further.
MCPConnectionError
class MCPConnectionError(Exception):
class MCPToolError(Exception):
ensure_strict_json_schema
def ensure_strict_json_schema(schema: dict[str, Any]):
strict standard that the OpenAI API expects.
_ensure_strict_json_schema
def _ensure_strict_json_schema(json_schema: object):
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):
class MCPToolkit(BaseToolkit):
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:
- 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
- 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()
- 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
Returns:
bool: True if the toolkit is connected to all MCP servers,
False otherwise.
connect_sync
Synchronously connect to all MCP servers.
disconnect_sync
def disconnect_sync(self):
enter
Synchronously enter the async context manager.
exit
def __exit__(
self,
exc_type,
exc_val,
exc_tb
):
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
):
_load_clients_from_config
def _load_clients_from_config(self, config_path: str):
_load_clients_from_dict
def _load_clients_from_dict(self, config: Dict[str, Any]):
_create_client_from_config
def _create_client_from_config(self, name: str, cfg: Dict[str, Any]):
def _ensure_strict_tool_schema(self, tool: FunctionTool):
- 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.
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):
def call_tool_sync(self, tool_name: str, tool_args: Dict[str, Any]):
def list_available_tools(self):
