Camel.toolkits.mcp toolkit

​

MCPClient

class MCPClient(BaseToolkit):

Internal class that provides an abstraction layer to interact with external tools using the Model Context Protocol (MCP). It supports three modes of connection:

1. stdio mode: Connects via standard input/output streams for local command-line interactions.

2. SSE mode (HTTP Server-Sent Events): Connects via HTTP for persistent, event-based interactions.

3. streamable-http mode: Connects via HTTP for persistent, streamable interactions.

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

1. Using the async context manager:

async with MCPClient(command_or_url="...") as client:
# Client is connected here
result = await client.some_tool()
# Client is automatically disconnected here

2. Using the factory method:

client = await MCPClient.create(command_or_url="...")
# Client is connected here
result = await client.some_tool()
# Don't forget to disconnect when done!
await client.disconnect()

3. Using explicit connect/disconnect:

client = MCPClient(command_or_url="...")
await client.connect()
# Client is connected here
result = await client.some_tool()
# Don't forget to disconnect when done!
await client.disconnect()

Attributes: command_or_url (str): URL for SSE mode or command executable for stdio mode. (default: :obj:None) args (List[str]): List of command-line arguments if stdio mode is used. (default: :obj:None) env (Dict[str, str]): Environment variables for the stdio mode command. (default: :obj:None) timeout (Optional[float]): Connection timeout. (default: :obj:None) headers (Dict[str, str]): Headers for the HTTP request. (default: :obj:None) mode (Optional[str]): Connection mode. Can be “sse” for Server-Sent Events, “streamable-http” for streaming HTTP, or None for stdio mode. (default: :obj:None) strict (Optional[bool]): Whether to enforce strict mode for the function call. (default: :obj:False)

​

init

def __init__(
    self,
    command_or_url: str,
    args: Optional[List[str]] = None,
    env: Optional[Dict[str, str]] = None,
    timeout: Optional[float] = None,
    headers: Optional[Dict[str, str]] = None,
    mode: Optional[str] = None,
    strict: Optional[bool] = False
):

​

connect_sync

def connect_sync(self):

Synchronously connect to the MCP server.

​

disconnect_sync

def disconnect_sync(self):

Synchronously disconnect from the MCP server.

​

connection_sync

def connection_sync(self):

Synchronously connect to the MCP server.

​

list_mcp_tools_sync

def list_mcp_tools_sync(self):

Synchronously list the available tools from the connected MCP server.

​

generate_function_from_mcp_tool

def generate_function_from_mcp_tool(self, mcp_tool: 'Tool'):

Dynamically generates a Python callable function corresponding to a given MCP tool.

Parameters:

• mcp_tool (Tool): The MCP tool definition received from the MCP server.

Returns:

Callable: A dynamically created async Python function that wraps the MCP tool.

​

generate_function_from_mcp_tool_sync

def generate_function_from_mcp_tool_sync(self, mcp_tool: 'Tool'):

Synchronously generate a function from an MCP tool.

​

_build_tool_schema

def _build_tool_schema(self, mcp_tool: 'Tool'):

​

get_tools

def get_tools(self):

Returns:

List[FunctionTool]: A list of FunctionTool objects representing the functions in the toolkit.

​

get_text_tools

def get_text_tools(self):

Returns:

str: A string containing the descriptions of the tools in the toolkit.

​

call_tool_sync

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

Synchronously call a tool.

​

session

def session(self):

​

create_sync

def create_sync(
    self,
    command_or_url: str,
    args: Optional[List[str]] = None,
    env: Optional[Dict[str, str]] = None,
    timeout: Optional[float] = None,
    headers: Optional[Dict[str, str]] = None,
    mode: Optional[str] = None
):

Synchronously create and connect to the MCP server.

​

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.

Parameters:

• exc_type (Optional[Type[Exception]]): The type of exception that occurred during the execution of the with statement.
• exc_val (Optional[Exception]): The exception that occurred during the execution of the with statement.
• exc_tb (Optional[TracebackType]): The traceback of the exception that occurred during the execution of the with statement.

Returns:

None

​

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.

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

1. Using the async context manager:

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:

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:

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:

• servers (Optional[List[MCPClient]]): List of MCPClient instances to manage. (default: :obj:None)
• config_path (Optional[str]): Path to a JSON configuration file defining MCP servers. (default: :obj:None)
• config_dict (Optional[Dict[str, Any]]): Dictionary containing MCP server configurations in the same format as the config file. (default: :obj:None)
• strict (Optional[bool]): Whether to enforce strict mode for the function call. (default: :obj:False)
• Note: Either servers, config_path, or config_dict must be provided. If multiple are provided, servers from all sources will be combined. For web servers in the config, you can specify authorization headers using the “headers” field to connect to protected MCP server endpoints. Example configuration: .. code-block:: json { "mcpServers": { "protected-server": { "url": "https://example.com/mcp", "timeout": 30, "headers": { "Authorization": "Bearer YOUR_TOKEN", "X-API-Key": "YOUR_API_KEY" } } } }
• Attributes:
• servers (List[MCPClient]): List of MCPClient instances being managed.

​

init

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

​

_load_servers_from_config

def _load_servers_from_config(self, config_path: str, strict: Optional[bool] = False):

Loads MCP server configurations from a JSON file.

Parameters:

• config_path (str): Path to the JSON configuration file.
• strict (bool): Whether to enforce strict mode for the function call. (default: :obj:False)

Returns:

List[MCPClient]: List of configured MCPClient instances.

​

_load_servers_from_dict

def _load_servers_from_dict(self, config: Dict[str, Any], strict: Optional[bool] = False):

Loads MCP server configurations from a dictionary.

Parameters:

• config (Dict[str, Any]): Dictionary containing server configurations.
• strict (bool): Whether to enforce strict mode for the function call. (default: :obj:False)

Returns:

List[MCPClient]: List of configured MCPClient instances.

​

connect_sync

def connect_sync(self):

Synchronously connect to all MCP servers.

​

disconnect_sync

def disconnect_sync(self):

Synchronously disconnect from all MCP servers.

​

connection_sync

def connection_sync(self):

Synchronously connect to all MCP servers.

​

is_connected

def is_connected(self):

Returns:

bool: True if connected, False otherwise.

​

get_tools

def get_tools(self):

Returns:

List[FunctionTool]: Combined list of all available function tools.

​

get_text_tools

def get_text_tools(self):

Returns:

str: A string containing the descriptions of the tools in the toolkit.