mcp_tool¶

MCP (Modular Command Protocol) Tools are function tools defined in a unified standard sharing with different models and services. They are served by an MCP server.

Features: - The MCPFunctionTool class, which wraps MCP tools as FunctionTool instances for use in agent workflows. - The MCPToolManager class, which manages multiple MCP server connections, loads server configurations from JSON, lists available resources/tools/prompts, and provides all tools as FunctionTool instances.

The module enables dynamic discovery and invocation of MCP tools for agent-based workflows.

Functions

`execute_mcp_op`(server_params, tool_name[, id])	Executes an operation using a specified MCP tool within a new client session.
`mcp_session_context`(server_params[, name])	Asynchronous context manager for establishing an MCP (Modular Communication Protocol) session.

Classes

`MCPFunctionTool`(server_params, mcp_tool)	A FunctionTool wrapper for MCP (Modular Command Protocol) tools.
`MCPServerSseParams`(url[, headers, timeout, ...])	Mirrors the params in mcp.client.sse.sse_client.
`MCPServerStdioParams`(command[, args, env, ...])	Mirrors mcp.client.stdio.StdioServerParameters, but lets you pass params without another import.
`MCPServerStreamableHttpParams`(url[, ...])	Mirrors the params in mcp.client.streamable_http.streamablehttp_client.
`MCPToolManager`([cache_tools_list, ...])	Manage MCP client connections and resources.

Bases: object

Mirrors mcp.client.stdio.StdioServerParameters, but lets you pass params without another import.

command: str¶

args: list[str] | None = None¶

env: dict[str, str] | None = None¶

cwd: str | Path | None = None¶

encoding: str | None = 'utf-8'¶

encoding_error_handler: Literal['strict', 'ignore', 'replace'] | None = 'strict'¶

class MCPServerSseParams(url: str, headers: dict[str, str] | None = None, timeout: float | None = 5, sse_read_timeout: float | None = 300)[source]¶

Bases: object

Mirrors the params in mcp.client.sse.sse_client.

url: str¶

headers: dict[str, str] | None = None¶

timeout: float | None = 5¶

sse_read_timeout: float | None = 300¶

class MCPServerStreamableHttpParams(url: str, headers: dict[str, str] | None = None, timeout: timedelta | None = datetime.timedelta(seconds=30), sse_read_timeout: timedelta | None = datetime.timedelta(seconds=300), terminate_on_close: bool | None = True)[source]¶

Bases: object

Mirrors the params in mcp.client.streamable_http.streamablehttp_client.

url: str¶

headers: dict[str, str] | None = None¶

timeout: timedelta | None = datetime.timedelta(seconds=30)¶

sse_read_timeout: timedelta | None = datetime.timedelta(seconds=300)¶

terminate_on_close: bool | None = True¶

mcp_session_context(server_params: MCPServerStdioParams | MCPServerSseParams | MCPServerStreamableHttpParams, name: str | None = None)[source]¶

Asynchronous context manager for establishing an MCP (Modular Communication Protocol) session.

Depending on the type of server_params, this function initializes a connection to an MCP server either via standard I/O or HTTP streaming, and yields an initialized ClientSession object.

Parameters:: server_params (MCPServerParameters) – Parameters for connecting to the MCP server. - If an instance of StdioServerParameters, connects via standard I/O. - If a string (interpreted as a URL), connects via HTTP streaming.
Yields:: ClientSession – An initialized client session for communicating with the MCP server.
Raises:: ValueError – If server_params is not a supported type.

async execute_mcp_op(server_params: MCPServerStdioParams | MCPServerSseParams | MCPServerStreamableHttpParams, tool_name: str, id=None, **params: dict) → str[source]¶

Executes an operation using a specified MCP tool within a new client session.

Parameters:

server_params (MCPServerParameters) – Parameters required to connect to the MCP server.
tool_name (str) – The name of the tool to execute.
id (optional) – An optional identifier for the operation.
**params (dict) – Additional parameters to pass to the tool.

Returns:

The textual result of the tool operation, or None if the operation failed.

Return type:

str

Raises:

Exception – If an error occurs during the tool execution, it is caught and logged, but not re-raised.

Side Effects:: Prints the result of the tool operation or an error message to the console, with colored output for emphasis.

Notes

The function uses an asynchronous context manager to handle the MCP session lifecycle.
The result is expected to be accessible via result.content[0].text.

class MCPFunctionTool(server_params: MCPServerStdioParams | MCPServerSseParams | MCPServerStreamableHttpParams, mcp_tool: Tool)[source]¶

Bases: FunctionTool

A FunctionTool wrapper for MCP (Modular Command Protocol) tools.

MCPFunctionTool enables seamless integration of MCP tools into agent workflows by exposing them as FunctionTool instances. It automatically translates the mcp.types.Tool into a FunctionTool. It allows dynamic discovery, description, and invocation of MCP tools, making them accessible to LLM-based agents or pipelines.

Note

Different from FunctionTool, MCPFunctionTool only supports acall since all tools are executed asynchronously in the MCP protocol.

Parameters:

server_params (MCPServerParameters) – The parameters required to connect to the MCP server. Could be a mcp.StdioServerParameters instance or a URL string.
mcp_tool (mcp.types.Tool) – The MCP tool instance to be used by this function tool.

Usage Example:

from adalflow.core.mcp_tool import MCPFunctionTool, mcp_session_context, StdioServerParameters

server_params = StdioServerParameters(
    command="python",
    args=["mcp_server.py"],
    env=None
)

async with mcp_session_context(server_params) as session:
    tools = await session.list_tools()
    tool = tools.tools[0]
    mcp_func_tool = MCPFunctionTool(server_params, tool)
    output = await mcp_func_tool.acall(param1="value1")

Features: - Wraps an MCP tool (from an MCP server) as a FunctionTool, providing a standardized interface for agent-based workflows. - Automatically generates a FunctionDefinition based on the MCP tool’s schema and description. - Supports asynchronous execution of the tool via the MCP protocol, using the provided server parameters. - Handles both stdio and HTTP-based MCP server connections.

async acall(*args: Any, **kwargs: Any) → FunctionOutput[source]¶

Execute the function asynchronously.

Need to be called in an async function or using asyncio.run.

Example:

import asyncio
server_params = StdioServerParameters(
    command="python",  # Command to run the server
    args=["mcp_server.py"],  # Arguments (path to your server script)
    env=None  # Optional environment variables
)
# Get tools from the server
async with mcp_session_context(server_params) as session:
    tools = await session.list_tools()
async def call_async_function():
    tool_1 = MCPFunctionTool(server_params, tools[0])
    output = await tool_1.acall()

asyncio.run(call_async_function())

bicall(*args, **kwargs)[source]¶: This function is not supported in MCPFunctionTool.

call(*args, **kwargs) → FunctionOutput[source]¶

Execute the function synchronously by running the async call method.

This is a convenience method that wraps the async acall method using asyncio.run(). It allows synchronous usage of MCP tools without requiring async/await syntax.

Parameters:

*args – Positional arguments (not supported, raises assertion error)
**kwargs – Keyword arguments to pass to the MCP tool

Returns:

The result of the MCP tool execution

Return type:

FunctionOutput

Example:

server_params = StdioServerParameters(
    command="python",
    args=["mcp_server.py"]
)
# Get tools from the server
async with mcp_session_context(server_params) as session:
    tools = await session.list_tools()

tool_1 = MCPFunctionTool(server_params, tools[0])
output = tool_1.call(param1="value1")  # Synchronous call

class MCPToolManager(cache_tools_list: bool = True, client_session_timeout_seconds: float | None = None)[source]¶

Bases: Component

Manage MCP client connections and resources.

Example:

from adalflow.core.mcp_tool import MCPToolManager, StdioServerParameters

manager = MCPToolManager()
# Add servers. Here we used a local stdio server defined in `mcp_server.py`.
# you can add more servers by calling `add_server` for multiple times.
manager.add_server("calculator_server", StdioServerParameters(
    command="python",  # Command to run the server
    args=["mcp_server.py"],  # Arguments (path to your server script)
))
manager.add_server("weather_server", StdioServerParameters(
    command="python",  # Command to run the server
    args=["mcp_weather_server.py"],  # Arguments (path to your weather server script)
))
await manager.list_all_tools()  # to see all available tools from all servers.
tools = await manager.get_all_tools()
# Add tools to agent
react = ReActAgent(
    max_steps=6,
    add_llm_as_fallback=True,
    tools=tools,
    model_client=model_client,
    model_kwargs=model_kwargs,
)

add_servers_from_json_file(json_path: str)[source]¶

Read MCP server configurations from a JSON file and add them to the manager.

Parameters:: json_path (str) – Path to the JSON file containing MCP server configurations.

Example of JSON structure:

{
    "mcpServers": {
        "mcp_weather_server": {
            "command": "python",
            "args": [
                "mcp_weather_server.py"
            ]
        }
    }
}

add_server(name: str, server_params: MCPServerStdioParams | MCPServerSseParams | MCPServerStreamableHttpParams)[source]¶

Adds a new MCP server to the internal server registry.

Parameters:

name (str) – The unique identifier for the server to be added.
server_params (MCPServerParameters) – An object containing the configuration parameters for the server. Could be a StdioServerParameters instance or a URL string.

Features:

If a server with the specified name does not already exist in the registry, it is added and a confirmation message is printed.
If a server with the specified name already exists, the addition is skipped and a ValueError is raised.

Raises:: ValueError – If a server with the specified name already exists in the registry.

async list_all_tools(server_names: List[str] = None)[source]¶

Print out all available resources, tools, and prompts from all added servers. If server_names is provided, only list tools for those specific servers.

Parameters:: server_names (List[str], optional) – A list of server names to filter the tools. If None, all servers are listed.

async get_all_tools(server_names: List[str] = None) → List[MCPFunctionTool][source]¶

Get all available functions from added servers as FunctionTool instances. If server_names is provided, only list tools for those specific servers.

Parameters:: server_names (List[str], optional) – A list of server names to filter the tools. If None, all servers are listed.