McpToolManager

McpToolManager#

Overview#

McpToolManager 继承自 ToolManagerBase ， 用于统一管理**多个外部 MCP 服务器**及其暴露的工具、资源、提示词。

Unlike the native Python tool manager, the MCP tool manager communicates with MCP servers implemented in any language via the Model Context Protocol, natively supporting cross-language, cross-process, and even cross-host tool calls.

In the Agent scenario, it is responsible for providing a list of tools to AgentNode, executing the corresponding tool based on the tool calling instruction output by the Large Language Model (LLM), and feeding the execution result back to the LLM. It also supports advanced features such as viewing connection status, dynamically adding or deleting servers, and tool-level permission control.

Establish MCP connection#

Before using McpToolManager, you need to configure the corresponding connection parameters McpServerLink for each MCP service. Subsequently, by using McpToolManager in an async with statement, connections to all registered MCP servers will be automatically established and automatically disconnected after the code block ends. After the connection is established, you can call methods such as list_tools, call_tools, etc., to get tool information or execute tool calls.

McpToolManager provides two connection management modes, please be sure to choose based on the scenario:

  1. Asynchronous Context Manager (Recommended)

    from evofabric.core.tool import McpToolManager
from evofabric.core.typing import StreamableHttpLink

# create mcp tool manager
manager = McpToolManager(
    server_links={
        "math_server": StreamableHttpLink(url="http://127.0.0.1:8000/mcp"),
        "file_server": StreamableHttpLink(url="http://127.0.0.1:8001/mcp"),
    },
    timeout=300,
    persistent_link=False
)
async with manager:
    await manager.list_tools()
    ...
# After leaving the `with` block, **regardless of whether `persistent_link` is True**,
# all MCP connections will be forcibly disconnected to prevent resource leaks
# when the event loop shuts down.

  2. Manual connect / disconnect

    await manager.connect()     # establish connection

await manager.list_tools()
...

await manager.disconnect()   # must disconnect manually, otherwise the connection will remain active!

    If disconnect() is not called, it may cause resource leaks such as sockets and subprocesses.

Note

Even if persistent_link=True, using the with statement will also actively disconnect from all servers upon exiting; if you need to maintain a long connection, please use manual connect/disconnect and ensure that it is released at the appropriate time.

Add, delete MCP Server#

McpToolManager 支持在运行期间动态增删 MCP 服务器。 你可以随时添加新的服务器连接配置，也可以删除已有服务器。删除服务器时，工具管理器会自动断开与该服务器的现有连接。 同时，你也可以根据需要单独重连某个服务器或重连全部服务器，并通过 McpToolManagerget_mcp_status() 查看当前所有服务器的连接状态。

# dynamically add servers
await manager.add_mcp_servers({
    "new_server": StreamableHttpLink(url="http://127.0.0.1:8003/mcp")
})

# delete previously added servers (will auto-disconnect first)
await manager.delete_mcp_servers(["file_server"])

# reconnect on demand
await manager.connect("math_server")   # connect a single server
await manager.connect()                # reconnect all servers

# check connection status
status = await manager.get_mcp_status()
# example return: {"math_server": True, "file_server": False}

MCP Server Usage#

McpToolManager supports tool viewing, tool calling, prompt retrieval, resource retrieval, and other functions.

Example:

manager = McpToolManager(
    server_links={
        "math_server": StreamableHttpLink(url="http://127.0.0.1:8000/mcp"),
        "file_server": StreamableHttpLink(url="http://127.0.0.1:8001/mcp"),
    },
    timeout=300,
    persistent_link=False
)
async with manager:
    await manager.list_tools()
    res = await manager.call_tools([ToolCall(
        id="0",
        function=Function(
            name="math_server_add",
            arguments=json.dumps(
                {
                    "a": 2,
                    "b": 1
                }
            )
        )
    )])
    resources = await manager.list_resources()
    ...

Tool Controller (Optional)#

In McpToolManager, you can configure an optional ToolController for tool calls. The tool controller allows you to dynamically enable or disable specific tools through rules. When the controller is registered:

  1. list_tools() will automatically filter out disabled tools and no longer return the schema of these tools.

  2. call_tools() will return an exception message indicating that the tool is disabled, if an attempt is made to call a disabled tool.

In this way, you can flexibly manage the tool sets allowed to be called in different scenarios.

from evofabric.core.tool import ToolController

# create and register a tool controller
controller = ToolController(
    rules={
        "math_server_calculate": True,   # enabled
        "file_server_write": False       # disabled
    }
)
manager.set_tool_controller(controller)

# after registering the controller:
# - disabled tools will no longer appear in list_tools()
# - calling a disabled tool will raise an error or return a failure result
Contents