工具系统与 MCP 集成

工具系统与 MCP 集成

学习目标

本章要解决什么问题：AutoGen 如何统一管理工具调用，包括本地函数、MCP 服务器和代码执行。你将学到：

• BaseTool 抽象与 FunctionTool 工厂
• Workbench 概念：持续状态的工具工作空间
• MCP 三种传输模式（stdio、SSE、Streamable HTTP）的集成
• 代码执行器的沙箱模式

前置知识

本章涉及工具调用和 MCP 协议的通用原理，建议先阅读：

• 工具调用协议与执行模型
• MCP 协议与生态集成

下文假设你已理解工具调用的基本流程，直接聚焦 AutoGen 的具体实现。

---

项目实践

BaseTool 与 FunctionTool

AutoGen Core 定义了工具的统一抽象：

# 方式一：从普通函数自动创建工具
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """Calculate Body Mass Index."""
    return weight_kg / (height_m ** 2)

tool = FunctionTool(calculate_bmi, description="Calculate BMI from weight and height.")

# 方式二：继承 BaseTool 自定义
class DatabaseQueryTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="db_query",
            description="Execute SQL query against the database.",
            parameters=SQLQuerySchema,
        )

    async def run(self, args: SQLQuerySchema, ctx: CancellationToken) -> ToolResult:
        result = await self._execute_query(args.sql)
        return ToolResult(content=result)

关键设计：

• FunctionTool 通过 Python 类型提示自动生成 JSON Schema
• BaseTool 允许自定义 schema 和运行逻辑
• ToolResult 统一表示工具执行结果（content、is_error）

Workbench：持续状态的工作空间

Workbench 是比 Tool 更高阶的抽象，代表一个有状态的工具集合：

class Workbench(ComponentBase):
    async def run_tool(self, name: str, arguments: dict, ctx: CancellationToken) -> ToolResult:
        ...
    def list_tools(self) -> List[Tool]:
        ...

Tool vs Workbench 的区别：

维度	Tool	Workbench
状态	无状态（每次调用独立）	有状态（维护会话/连接）
粒度	单一操作	工具集合
生命周期	调用即销毁	创建后可多次使用
示例	calculate_bmi	McpWorkbench（维护 MCP 连接）

MCP 集成：三种传输模式

autogen-ext 提供了 MCP 服务器的完整集成，支持三种传输模式：

from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams

# 方式一：Stdio（通过标准输入输出通信）
params = StdioServerParams(
    command="npx",
    args=["@playwright/mcp@latest", "--headless"],
)

async with McpWorkbench(params) as mcp:
    agent = AssistantAgent("browser", model_client=client, workbench=mcp)
    await Console(agent.run_stream(task="Search for AutoGen on the web"))

# 方式二：SSE（Server-Sent Events）
from autogen_ext.tools.mcp import SseServerParams

params = SseServerParams(url="http://localhost:8080/sse")

# 方式三：Streamable HTTP
from autogen_ext.tools.mcp import StreamableHttpParams

params = StreamableHttpParams(url="http://localhost:8080/mcp")

安全警告：AutoGen README 明确提示 “Only connect to trusted MCP servers as they may execute commands in your local environment”。

代码执行器

from autogen_ext.code_executors import LocalCommandLineCodeExecutor

executor = LocalCommandLineCodeExecutor(work_dir="coding")
# 或使用 Docker 沙箱
from autogen_ext.code_executors import DockerCommandLineCodeExecutor

executor = DockerCommandLineCodeExecutor(image="python:3.11")

支持的后端：

• LocalCommandLineCodeExecutor：本地命令行（开发环境）
• DockerCommandLineCodeExecutor：Docker 容器隔离（生产环境）
• AzureContainerCodeExecutor：Azure 容器实例

问题与规避

MCP 服务器的信任问题

陷阱：MCP 服务器可以执行任意命令，连接到不可信的服务器可能导致代码注入或信息泄露。

规避：

• 仅连接到来源可信的 MCP 服务器
• 在 Docker 容器中运行 MCP 服务器，限制权限
• 审查 MCP 服务器的源码，确认无恶意行为

工具输出大小限制

陷阱：某些工具（如文件读取、数据库查询）可能返回大量内容，超出模型上下文窗口。

规避：

• 在工具实现中设置最大输出长度
• 使用 Workbench 的分页查询能力
• 在 Agent 系统提示中指示模型”如果需要更多信息，分批查询”

设计取舍

Tool vs Workbench 的边界

为什么需要 Workbench？

对于像 MCP 服务器这样有连接状态的工具集合，如果每个工具独立管理连接：

• 每次调用都建立/销毁连接 → 性能浪费
• 多个工具之间无法共享上下文

Workbench 解决了这个问题：一次连接，多个工具共享状态。

替代方案

方案	状态管理	多工具	复杂度	示例
独立 Tool	无	不支持	低	简单函数调用
Workbench	有	支持	中	MCP 集成
直接 SDK 调用	有	支持	高	自定义客户端

参考来源

• AutoGen Tools Documentation — 工具使用指南
• 源码验证：autogen-core/tools/_base.py（Tool 抽象）
• 源码验证：autogen-ext/tools/mcp/（MCP 集成）