第5章：工具集成 —— Function Calling、MCP 协议与工具链设计

本章导引

如果说 Context Manager 负责让 Agent "看到"信息，那么 Tool Executor 负责让 Agent "动手"做事。工具集成是 Agent 从"聊天机器人"到"自主系统"的关键跨越——没有工具，Agent 只能做语言输出；有了工具，Agent 可以操作文件、执行代码、查询数据库、调用 API。

本章从三个层次深入工具集成：

  1. Function Calling 基础（5.1）：从底层 API 原理到通用工具执行引擎
  2. MCP 协议（5.2）：Model Context Protocol——Agent 与工具的标准化连接
  3. 工具链设计模式（5.3）：组合、安全、容错——生产级工具系统

5.1 Function Calling 深度实战

5.1.1 Function Calling 原理：从 API 到 Agent 执行

Function Calling 的工作机制可以分解为五个步骤：

用户输入："帮我读取 config.json 并检查配置是否正确"
    │
    ▼
┌─────────────────────────────────────────────────┐
│ Step 1: Harness 组装 Context，附带工具定义        │
│                                                    │
│ Context 中包含了可用工具的 JSON Schema：             │
│ {                                                  │
│   "name": "read_file",                             │
│   "description": "读取文件内容",                    │
│   "parameters": {                                  │
│     "path": { "type": "string", "required": true } │
│   }                                                │
│ }                                                  │
└────────────────────┬────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────┐
│ Step 2: Model 判断需要调用工具                     │
│                                                    │
│ Model 输出（不是纯文本，而是结构化 tool_use）：       │
│ {                                                  │
│   "type": "tool_use",                              │
│   "id": "tool_001",                                │
│   "name": "read_file",                             │
│   "input": { "path": "config.json" }               │
│ }                                                  │
└────────────────────┬────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────┐
│ Step 3: Harness 解析 tool_use 指令                 │
│                                                    │
│ Tool Executor 根据 name 找到对应的 handler          │
│ 校验 input 参数是否符合 JSON Schema                 │
└────────────────────┬────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────┐
│ Step 4: 执行工具，获取结果                          │
│                                                    │
│ const content = await fs.readFile("config.json")   │
│ → "{ \"port\": 3000, \"db\": \"postgresql\" }"     │
└────────────────────┬────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────┐
│ Step 5: 工具结果注入 Context，进入下一轮推理        │
│                                                    │
│ Context 中新增 tool_result 消息                     │
│ Model 基于结果继续推理："配置看起来是合理的..."      │
└─────────────────────────────────────────────────┘

5.1.2 工具定义规范：JSON Schema、参数校验、类型安全

JSON Schema 完整规范

一个好的工具定义是工具可靠执行的前提。以下是生产级的工具定义规范：

// TypeScript: 工具定义的完整 TypeScript 类型
interface ToolDefinition {
  /** 工具唯一名称（使用 snake_case） */
  name: string;
  /** 工具功能描述（Model 根据此描述决定何时调用） */
  description: string;
  /** 参数 JSON Schema */
  inputSchema: {
    type: 'object';
    properties: Record<string, ParameterSchema>;
    required?: string[];
  };
  /** 执行配置 */
  execution?: {
    timeoutMs?: number;           // 超时（默认 30000ms）
    maxRetries?: number;          // 最大重试次数（默认 0）
    retryableErrors?: string[];   // 可重试的错误类型
    concurrency?: 'sequential' | 'parallel'; // 并发策略
  };
  /** 安全配置 */
  security?: {
    permission: 'read' | 'write' | 'execute' | 'admin';
    requiresConfirmation?: boolean;
    allowedPaths?: string[];     // 文件路径白名单
    blockedPatterns?: string[];  // 参数黑名单模式
  };
  /** 输出配置 */
  output?: {
    maxLength?: number;          // 最大输出长度（默认 10000）
    truncationMessage?: string;  // 截断提示
    format?: 'text' | 'json' | 'markdown';
  };
}

interface ParameterSchema {
  type: 'string' | 'number' | 'boolean' | 'object' | 'array';
  description: string;
  enum?: string[];              // 枚举值
  default?: unknown;
  minimum?: number;
  maximum?: number;
  pattern?: string;             // 正则校验
  maxLength?: number;
  items?: ParameterSchema;      // 数组元素类型
  properties?: Record<string, ParameterSchema>; // 嵌套对象
}

# Python: 工具定义
from pydantic import BaseModel, Field
from typing import Optional, Literal
from enum import Enum

class ToolPermission(str, Enum):
    READ = "read"
    WRITE = "write"
    EXECUTE = "execute"
    ADMIN = "admin"

class ToolDefinition(BaseModel):
    name: str
    description: str
    input_schema: dict  # JSON Schema
    execution: Optional[dict] = None
    security: Optional[dict] = None
    output: Optional[dict] = None

# 使用 Pydantic 定义工具参数（自动生成 JSON Schema）
from pydantic import BaseModel

class ReadFileArgs(BaseModel):
    """读取文件的参数"""
    path: str = Field(description="文件路径（相对于工作目录）")
    encoding: str = Field(default="utf-8", description="文件编码")
    max_lines: Optional[int] = Field(default=None, description="最大读取行数")

class WriteFileArgs(BaseModel):
    """写入文件的参数"""
    path: str = Field(description="文件路径")
    content: str = Field(description="文件内容")
    mode: Literal["overwrite", "append"] = Field(default="overwrite")

class RunCommandArgs(BaseModel):
    """执行命令的参数"""
    command: str = Field(description="要执行的 Shell 命令")
    working_dir: Optional[str] = Field(default=None, description="工作目录")
    timeout: int = Field(default=30, ge=1, le=300, description="超时秒数")

# 生成 JSON Schema（直接传给 LLM API 的 tools 参数）
read_file_schema = ReadFileArgs.model_json_schema()
write_file_schema = WriteFileArgs.model_json_schema()
run_command_schema = RunCommandArgs.model_json_schema()
类型安全的参数校验
// TypeScript: Zod Schema 驱动的工具参数校验
import { z } from 'zod';

// 用 Zod 定义参数 Schema（比手写 JSON Schema 更类型安全）
const ReadFileArgs = z.object({
  path: z.string().describe('文件路径'),
  encoding: z.enum(['utf-8', 'ascii', 'base64']).default('utf-8'),
  maxLines: z.number().int().positive().optional(),
});

const WriteFileArgs = z.object({
  path: z.string().describe('文件路径'),
  content: z.string().describe('文件内容'),
  mode: z.enum(['overwrite', 'append']).default('overwrite'),
});

const SearchCodeArgs = z.object({
  pattern: z.string().describe('搜索模式（支持正则）'),
  path: z.string().default('.').describe('搜索路径'),
  fileTypes: z.array(z.string()).optional().describe('文件类型过滤'),
  maxResults: z.number().int().min(1).max(100).default(20),
});

// Zod Schema → JSON Schema（自动转换）
function zodToJsonSchema(schema: z.ZodObject<any>): object {
  // 使用 zod-to-json-schema 库
  return zodToJsonSchemaLib(schema);
}

// 类型安全的工具注册
class TypedToolRegistry {
  private tools = new Map<string, {
    schema: z.ZodObject<any>;
    handler: (args: any) => Promise<string>;
    definition: ToolDefinition;
  }>();

  register<T extends z.ZodObject<any>>(
    name: string,
    description: string,
    schema: T,
    handler: (args: z.infer<T>) => Promise<string>,
    options?: Partial<ToolDefinition['execution'] & ToolDefinition['security']>
  ): void {
    this.tools.set(name, {
      schema,
      handler,
      definition: {
        name,
        description,
        inputSchema: zodToJsonSchema(schema) as any,
        execution: options,
      },
    });
  }

  getDefinitions(): ToolDefinition[] {
    return Array.from(this.tools.values()).map(t => t.definition);
  }

  async execute(name: string, rawArgs: Record<string, unknown>): Promise<ToolResult> {
    const tool = this.tools.get(name);
    if (!tool) throw new Error(`未知工具: ${name}`);
    
    // Zod 自动校验 + 类型推断
    const args = tool.schema.parse(rawArgs);
    const result = await tool.handler(args);
    return { status: 'success', content: result };
  }
}

// 使用
const registry = new TypedToolRegistry();

registry.register(
  'read_file', '读取文件内容',
  ReadFileArgs,
  async (args) => {
    // args 的类型被 Zod 自动推断为 { path: string; encoding: 'utf-8'|'ascii'|'base64'; maxLines?: number }
    const content = await fs.promises.readFile(args.path, args.encoding);
    if (args.maxLines) {
      return content.split('\n').slice(0, args.maxLines).join('\n');
    }
    return content;
  },
  { permission: 'read', timeoutMs: 10000 }
);

5.1.3 工具执行流程：解析→校验→执行→格式化→注入上下文

完整的工具执行流程包含五个阶段：

// TypeScript: 完整的工具执行流程
class ToolExecutionPipeline {
  constructor(
    private validator: ParameterValidator,
    private sandbox: SandboxExecutor,
    private formatter: ResultFormatter,
    private security: SafetyGuard,
  ) {}

  async execute(
    toolCall: ToolCall,
    context: ExecutionContext,
  ): Promise<ToolResult> {
    const startTime = Date.now();
    const trace = context.observability.startSpan('tool_execution', {
      tool: toolCall.name,
      args: JSON.stringify(toolCall.arguments).slice(0, 200),
    });

    try {
      // ===== Phase 1: 解析 =====
      const tool = this.registry.get(toolCall.name);
      if (!tool) {
        return this.errorResult(toolCall, 'unknown_tool', 
          `未找到工具: ${toolCall.name}。可用: ${this.registry.list()}`);
      }

      // ===== Phase 2: 校验 =====
      const validationResult = await this.validator.validate(
        toolCall.arguments,
        tool.definition.inputSchema,
      );
      if (!validationResult.valid) {
        return this.errorResult(toolCall, 'validation_error',
          `参数校验失败: ${validationResult.errors.join('; ')}`);
      }

      // ===== Phase 3: 安全检查 =====
      const safetyCheck = await this.security.checkToolCall(
        toolCall.name,
        toolCall.arguments,
        context.userPermissions,
      );
      if (!safetyCheck.passed) {
        trace.setTag('blocked', true);
        trace.setTag('block_reason', safetyCheck.reason);
        return {
          toolCallId: toolCall.id,
          toolName: toolCall.name,
          status: 'permission_denied',
          content: `操作被安全护栏阻止: ${safetyCheck.reason}`,
          duration: Date.now() - startTime,
        };
      }

      // ===== Phase 4: 执行 =====
      const result = await this.sandbox.execute(
        tool.handler,
        toolCall.arguments,
        {
          timeout: tool.definition.execution?.timeoutMs ?? 30000,
          maxRetries: tool.definition.execution?.maxRetries ?? 0,
          workingDirectory: context.workingDirectory,
          env: context.env,
        },
      );

      // ===== Phase 5: 格式化 =====
      const formatted = this.formatter.format(result, {
        maxLength: tool.definition.output?.maxLength ?? 10000,
        format: tool.definition.output?.format ?? 'text',
        truncationMessage: tool.definition.output?.truncationMessage,
      });

      trace.setTag('success', true);
      trace.setTag('duration_ms', Date.now() - startTime);

      return {
        toolCallId: toolCall.id,
        toolName: toolCall.name,
        status: 'success',
        content: formatted,
        duration: Date.now() - startTime,
      };

    } catch (error: any) {
      trace.setTag('error', true);
      trace.setTag('error_type', error.constructor.name);
      
      return {
        toolCallId: toolCall.id,
        toolName: toolCall.name,
        status: error.message?.includes('timeout') ? 'timeout' : 'error',
        content: `工具执行失败: ${error.message}`,
        duration: Date.now() - startTime,
        error: error.message,
      };
    } finally {
      trace.end();
    }
  }
}

5.1.4 实践案例：构建通用工具注册与执行引擎

// TypeScript: 通用工具引擎完整实现
class UniversalToolEngine {
  private registry = new Map<string, RegisteredTool>();
  private pipeline: ToolExecutionPipeline;
  private concurrencyManager: ConcurrencyManager;

  constructor(config: ToolEngineConfig) {
    this.pipeline = new ToolExecutionPipeline(
      new ZodParameterValidator(),
      new SandboxExecutor(config.sandbox),
      new ResultFormatter(),
      config.safetyGuard,
    );
    this.concurrencyManager = new ConcurrencyManager({
      maxConcurrent: config.maxConcurrent ?? 5,
    });
  }

  /** 注册工具 */
  register<T extends z.ZodObject<any>>(opts: {
    name: string;
    description: string;
    schema: T;
    handler: (args: z.infer<T>, ctx: ExecutionContext) => Promise<string>;
    permission?: ToolPermission;
    timeoutMs?: number;
    requiresConfirmation?: boolean;
  }): void {
    const { name, description, schema, handler, ...options } = opts;

    this.registry.set(name, {
      definition: {
        name,
        description,
        inputSchema: zodToJsonSchema(schema) as any,
        execution: { timeoutMs: options.timeoutMs ?? 30000 },
        security: {
          permission: options.permission ?? 'read',
          requiresConfirmation: options.requiresConfirmation ?? false,
        },
      },
      schema,
      handler,
    });
  }

  /** 获取所有工具定义（SUB Agent 注册发现） */
  getDefinitions(): ToolDefinition[] {
    return Array.from(this.registry.values()).map(t => t.definition);
  }

  /** 批量执行工具调用 */
  async executeBatch(
    calls: ToolCall[],
    context: ExecutionContext,
  ): Promise<ToolResult[]> {
    // 按并发策略分组
    const groups = this.groupByConcurrency(calls);

    const results: ToolResult[] = [];

    for (const [strategy, group] of groups) {
      if (strategy === 'parallel') {
        // 并行执行
        const parallelResults = await Promise.all(
          group.map(call =>
            this.concurrencyManager.execute(() =>
              this.executeOne(call, context)
            )
          )
        );
        results.push(...parallelResults);
      } else {
        // 顺序执行（某些工具有依赖关系）
        for (const call of group) {
          const result = await this.concurrencyManager.execute(() =>
            this.executeOne(call, context)
          );
          results.push(result);
        }
      }
    }

    return results;
  }

  private async executeOne(
    call: ToolCall,
    context: ExecutionContext,
  ): Promise<ToolResult> {
    const tool = this.registry.get(call.name);
    if (!tool) {
      return {
        toolCallId: call.id,
        toolName: call.name,
        status: 'error',
        content: `未知工具: ${call.name}`,
        duration: 0,
        error: 'unknown_tool',
      };
    }

    return this.pipeline.execute(call, {
      ...context,
      toolHandler: tool.handler,
      toolSchema: tool.schema,
    });
  }
}

5.2 MCP 协议（Model Context Protocol）

5.2.1 MCP 协议架构：Client-Server 模型

MCP（Model Context Protocol）是 Anthropic 于 2024 年底发布的一种开放协议，定义了 Agent 与外部工具/资源服务的标准化交互方式。它的核心思想是将工具提供方（Server）和工具使用方（Client）解耦——Agent 是 Client，各种工具作为独立的 Server 运行。

┌─────────────────────┐         MCP Protocol        ┌─────────────────────┐
│                     │◄────────────────────────────►│                     │
│   MCP Client        │                              │   MCP Server        │
│   (Agent 框架)      │   tools/list                 │   (工具服务)         │
│                     │   tools/call                 │                     │
│   - 发现可用工具     │   resources/read              │   - 文件系统操作    │
│   - 调用工具        │   resources/list             │   - 数据库查询      │
│   - 订阅资源变更    │   prompts/list               │   - API 调用        │
│                     │   notifications/*             │   - 代码执行        │
└─────────────────────┘                              └─────────────────────┘

MCP 的三个核心原语：

原语 用途 示例
Tools Agent 可调用的操作 read_file, run_sql, deploy
Resources Agent 可访问的数据 file://config.json, postgres://db/schema
Prompts 预定义的提示词模板 code_review_prompt, commit_message_prompt

传输方式：

  • stdio：通过标准输入/输出通信（最简单，适合本地工具）
  • HTTP + SSE：通过 HTTP 请求和 SSE 流通信（适合远程服务）
  • WebSocket：双向实时通信（适合长时间运行的工具）

5.2.2 构建 MCP Server：从零实现

// TypeScript: 简单 MCP Server 实现
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from '@modelcontextprotocol/sdk/types.js';

// 创建 MCP Server
const server = new Server(
  {
    name: 'file-system-server',
    version: '1.0.0',
  },
  {
    capabilities: {
      tools: {},
      resources: { subscribe: true },
    },
  }
);

// 注册工具列表
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: 'read_file',
      description: '读取文件内容',
      inputSchema: {
        type: 'object',
        properties: {
          path: { type: 'string', description: '文件路径' },
          encoding: { type: 'string', default: 'utf-8' },
        },
        required: ['path'],
      },
    },
    {
      name: 'write_file',
      description: '写入文件',
      inputSchema: {
        type: 'object',
        properties: {
          path: { type: 'string', description: '文件路径' },
          content: { type: 'string', description: '文件内容' },
        },
        required: ['path', 'content'],
      },
    },
    {
      name: 'list_directory',
      description: '列出目录内容',
      inputSchema: {
        type: 'object',
        properties: {
          path: { type: 'string', description: '目录路径' },
        },
        required: ['path'],
      },
    },
  ],
}));

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case 'read_file': {
      const content = await fs.promises.readFile(
        args.path as string,
        (args.encoding as BufferEncoding) || 'utf-8'
      );
      return {
        content: [{ type: 'text', text: content }],
      };
    }

    case 'write_file': {
      await fs.promises.writeFile(
        args.path as string,
        args.content as string
      );
      return {
        content: [{ type: 'text', text: `文件已写入: ${args.path}` }],
      };
    }

    case 'list_directory': {
      const files = await fs.promises.readdir(args.path as string);
      return {
        content: [{ type: 'text', text: files.join('\n') }],
      };
    }

    default:
      throw new Error(`未知工具: ${name}`);
  }
});

// 启动 Server（通过 stdio 通信）
const transport = new StdioServerTransport();
await server.connect(transport);

console.error('MCP File System Server 已启动 (stdio)');

# Python: 简单 MCP Server 实现
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.server.stdio import stdio_server
import mcp.types as types

# 创建 Server
server = Server("file-system-server")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="read_file",
            description="读取文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                },
                "required": ["path"],
            },
        ),
        types.Tool(
            name="write_file",
            description="写入文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["path", "content"],
            },
        ),
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent]:
    if name == "read_file":
        path = arguments["path"]
        with open(path, "r") as f:
            content = f.read()
        return [types.TextContent(type="text", text=content)]
    
    elif name == "write_file":
        path = arguments["path"]
        content = arguments["content"]
        with open(path, "w") as f:
            f.write(content)
        return [types.TextContent(type="text", text=f"文件已写入: {path}")]
    
    raise ValueError(f"未知工具: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationCapabilities(
                sampling={},
                experimental={},
            ),
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

5.2.3 MCP Client 集成：Agent 与 MCP 生态的对接

// TypeScript: MCP Client 集成到 Agent
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js';

class MCPToolProvider {
  private clients: Map<string, Client> = new Map();

  /** 连接 MCP Server */
  async connectServer(name: string, config: MCPServerConfig): Promise<void> {
    let transport;
    
    switch (config.transport) {
      case 'stdio':
        transport = new StdioClientTransport({
          command: config.command!,
          args: config.args || [],
          env: config.env,
        });
        break;
      case 'http':
        transport = new HTTPClientTransport({
          url: config.url!,
        });
        break;
    }

    const client = new Client(
      { name: 'agent-mcp-client', version: '1.0.0' },
      { capabilities: { tools: {} } }
    );

    await client.connect(transport);
    this.clients.set(name, client);
    
    console.log(`[MCP] 已连接 Server: ${name}`);
  }

  /** 获取所有 MCP Server 的工具 */
  async discoverAllTools(): Promise<ToolDefinition[]> {
    const allTools: ToolDefinition[] = [];

    for (const [serverName, client] of this.clients) {
      const response = await client.request(
        { method: 'tools/list' },
        { timeout: 5000 }
      );

      for (const tool of response.tools) {
        allTools.push({
          ...tool,
          name: `${serverName}__${tool.name}`,  // 添加前缀防止冲突
          metadata: { mcpServer: serverName },
        });
      }
    }

    return allTools;
  }

  /** 调用 MCP 工具 */
  async callTool(
    fullName: string,
    args: Record<string, unknown>
  ): Promise<string> {
    // 解析 serverName__toolName 格式
    const [serverName, ...toolNameParts] = fullName.split('__');
    const toolName = toolNameParts.join('__');

    const client = this.clients.get(serverName);
    if (!client) throw new Error(`MCP Server 未连接: ${serverName}`);

    const response = await client.request(
      {
        method: 'tools/call',
        params: { name: toolName, arguments: args },
      },
      { timeout: 30000 }
    );

    // 提取文本内容
    return (response.content as any[])
      .filter(c => c.type === 'text')
      .map(c => c.text)
      .join('\n');
  }

  /** 断开所有连接 */
  async disconnectAll(): Promise<void> {
    for (const [name, client] of this.clients) {
      await client.close();
      console.log(`[MCP] 已断开: ${name}`);
    }
    this.clients.clear();
  }
}

5.2.4 实践案例：为 Agent 添加文件系统、数据库、API 三种 MCP 工具

# Python: 三个 MCP Server 的完整实现

# ===== Server 1: 文件系统工具 =====
# mcp_servers/filesystem_server.py

@server.list_tools()
async def list_filesystem_tools():
    return [
        types.Tool(name="fs_read", description="读取文件",
            inputSchema={"type": "object", "properties": {
                "path": {"type": "string"}}, "required": ["path"]}),
        types.Tool(name="fs_write", description="写入文件",
            inputSchema={"type": "object", "properties": {
                "path": {"type": "string"}, "content": {"type": "string"}},
                "required": ["path", "content"]}),
        types.Tool(name="fs_search", description="搜索文件内容",
            inputSchema={"type": "object", "properties": {
                "pattern": {"type": "string"}, "path": {"type": "string"}},
                "required": ["pattern"]}),
    ]

@server.call_tool()
async def handle_fs_tool(name: str, args: dict):
    if name == "fs_search":
        pattern = args["pattern"]
        search_path = args.get("path", ".")
        results = []
        for root, _, files in os.walk(search_path):
            for file in files:
                filepath = os.path.join(root, file)
                try:
                    with open(filepath, 'r', errors='ignore') as f:
                        for i, line in enumerate(f, 1):
                            if pattern in line:
                                results.append(f"{filepath}:{i}: {line.strip()}")
                except:
                    pass
        return [types.TextContent(type="text",
            text="\n".join(results[:50]) or "未找到匹配结果")]

# ===== Server 2: PostgreSQL 数据库工具 =====
# mcp_servers/database_server.py

import asyncpg

class DatabaseMCPServer:
    def __init__(self, conn_string: str):
        self.conn_string = conn_string
        self.pool: Optional[asyncpg.Pool] = None
    
    async def connect(self):
        self.pool = await asyncpg.create_pool(self.conn_string, min_size=1, max_size=5)
    
    @server.list_tools()
    async def list_tools(self):
        return [
            types.Tool(name="db_query", description="执行 SELECT 查询",
                inputSchema={"type": "object", "properties": {
                    "query": {"type": "string"}}, "required": ["query"]}),
            types.Tool(name="db_schema", description="查看表结构",
                inputSchema={"type": "object", "properties": {
                    "table": {"type": "string"}}}),
            types.Tool(name="db_explain", description="分析查询计划",
                inputSchema={"type": "object", "properties": {
                    "query": {"type": "string"}}, "required": ["query"]}),
        ]
    
    @server.call_tool()
    async def handle_tool(self, name: str, args: dict):
        async with self.pool.acquire() as conn:
            if name == "db_query":
                # 安全检查：只允许 SELECT
                query = args["query"].strip()
                if not query.upper().startswith("SELECT"):
                    return [types.TextContent(type="text",
                        text="错误: db_query 只允许 SELECT 语句")]
                
                rows = await conn.fetch(query)
                # 格式化为表格
                if not rows:
                    return [types.TextContent(type="text", text="查询结果为空")]
                
                headers = list(rows[0].keys())
                lines = [" | ".join(headers), "-" * 50]
                for row in rows[:100]:
                    lines.append(" | ".join(str(v) for v in row.values()))
                return [types.TextContent(type="text", text="\n".join(lines))]
            
            elif name == "db_schema":
                table = args.get("table", "%")
                rows = await conn.fetch("""
                    SELECT table_name, column_name, data_type, is_nullable
                    FROM information_schema.columns
                    WHERE table_name LIKE $1
                    ORDER BY table_name, ordinal_position
                """, table)
                # ... 格式化输出

# ===== Server 3: HTTP API 工具 =====
# mcp_servers/api_server.py

class APIMCPServer:
    @server.list_tools()
    async def list_tools(self):
        return [
            types.Tool(name="api_get", description="发送 GET 请求",
                inputSchema={"type": "object", "properties": {
                    "url": {"type": "string"}, "headers": {"type": "object"}},
                    "required": ["url"]}),
            types.Tool(name="api_post", description="发送 POST 请求",
                inputSchema={"type": "object", "properties": {
                    "url": {"type": "string"}, "body": {"type": "object"},
                    "headers": {"type": "object"}}, "required": ["url"]}),
        ]
    
    @server.call_tool()
    async def handle_tool(self, name: str, args: dict):
        async with httpx.AsyncClient(timeout=30) as client:
            if name == "api_get":
                resp = await client.get(args["url"], headers=args.get("headers"))
                return self._format_response(resp)
            elif name == "api_post":
                resp = await client.post(
                    args["url"], json=args.get("body"), headers=args.get("headers"))
                return self._format_response(resp)
    
    def _format_response(self, resp) -> list:
        result = f"状态: {resp.status_code}\n"
        result += f"Content-Type: {resp.headers.get('content-type', 'unknown')}\n\n"
        # 截断过长内容
        body = resp.text[:5000]
        result += body
        return [types.TextContent(type="text", text=result)]

# ===== 启动所有 MCP Server =====
# 在 Agent 启动时连接所有 MCP Server
mcp_provider = MCPToolProvider()
await mcp_provider.connectServer('filesystem', {
    transport: 'stdio',
    command: 'python',
    args: ['mcp_servers/filesystem_server.py'],
})
await mcp_provider.connectServer('database', {
    transport: 'stdio',
    command: 'python',
    args: ['mcp_servers/database_server.py'],
    env: { 'DATABASE_URL': process.env.DATABASE_URL },
})
await mcp_provider.connectServer('api', {
    transport: 'http',
    url: 'http://localhost:3001/mcp',
})

// 将 MCP 工具注入 Agent
const mcpTools = await mcpProvider.discoverAllTools();
for (const tool of mcpTools) {
  agent.registerTool(tool, async (args) => {
    return mcpProvider.callTool(tool.name, args);
  });
}

5.3 工具链设计模式

5.3.1 工具组合模式：顺序链、并行执行、条件分支

顺序链（Sequential Chain）
// TypeScript: 工具顺序链
class ToolChain {
  private steps: ToolChainStep[] = [];

  pipe(tool: string, argsMapper?: (prevResult: any) => Record<string, unknown>): this {
    this.steps.push({ tool, argsMapper });
    return this;
  }

  async execute(initialArgs: Record<string, unknown>, context: ExecutionContext): Promise<ToolChainResult> {
    const results: ToolResult[] = [];
    let lastResult: any = null;

    for (const step of this.steps) {
      const args = step.argsMapper ? step.argsMapper(lastResult) : initialArgs;
      const result = await context.toolExecutor.execute(
        { id: `chain_${results.length}`, name: step.tool, arguments: args },
        context
      );
      results.push(result);

      if (result.status === 'error') {
        return { success: false, results, failedAt: step.tool };
      }

      lastResult = result.content;
    }

    return { success: true, results };
  }
}

// 使用：代码分析流水线
const analysisChain = new ToolChain()
  .pipe('git_diff', (prev) => ({ branch: 'main' }))
  .pipe('static_analysis', (prev) => ({ diff: prev }))
  .pipe('security_scan', (prev) => ({ code: prev }))
  .pipe('generate_report', (prev) => ({ findings: prev }));
条件分支（Conditional Branching）
// TypeScript: 条件工具选择
class ToolSwitch {
  async route(
    condition: (context: ExecutionContext) => Promise<string>,
    branches: Record<string, () => Promise<ToolResult>>,
    context: ExecutionContext
  ): Promise<ToolResult> {
    const branch = await condition(context);
    const handler = branches[branch] || branches['default'];
    if (!handler) throw new Error(`未匹配的分支: ${branch}`);
    return handler();
  }
}

// 使用：根据代码语言选择不同的格式化工具
const formatter = new ToolSwitch();
const result = await formatter.route(
  async (ctx) => {
    const ext = path.extname(ctx.currentFile);
    if (ext === '.ts' || ext === '.tsx') return 'prettier';
    if (ext === '.py') return 'black';
    return 'default';
  },
  {
    prettier: async () => toolExecutor.execute({ name: 'run_prettier', ... }),
    black: async () => toolExecutor.execute({ name: 'run_black', ... }),
    default: async () => ({ status: 'success', content: '无需格式化' }),
  },
  context
);

5.3.2 工具错误处理：超时、重试、降级、人工兜底

工具调用失败的处理层次：
                    ┌──────────────────┐
                    │   人工兜底         │  ← 所有自动手段失败后
                    │   (HITL)          │
                    └────────┬─────────┘
                             │
                    ┌────────┴─────────┐
                    │   降级 (Fallback) │  ← 换一种方式实现目标
                    │   用备选工具代替    │
                    └────────┬─────────┘
                             │
                    ┌────────┴─────────┐
                    │   重试 (Retry)    │  ← 指数退避重试
                    │   3次 + 退避      │
                    └────────┬─────────┘
                             │
                    ┌────────┴─────────┐
                    │   超时处理        │  ← 30s超时 + 降级
                    │   Timeout        │
                    └──────────────────┘

// TypeScript: 工具错误处理策略
class ToolErrorHandler {
  private strategies: Map<string, ErrorStrategy> = new Map();

  registerStrategy(errorType: string, strategy: ErrorStrategy): void {
    this.strategies.set(errorType, strategy);
  }

  async handle(
    error: ToolError,
    toolCall: ToolCall,
    context: ExecutionContext
  ): Promise<ToolResult> {
    const strategy = this.strategies.get(error.type) ?? this.strategies.get('default')!;

    // Level 1: 重试
    if (strategy.retry && error.retryCount < strategy.maxRetries) {
      console.log(`[${toolCall.name}] 重试 ${error.retryCount + 1}/${strategy.maxRetries}`);
      await this.delay(strategy.retryDelayMs * Math.pow(2, error.retryCount));
      // 重新执行...
    }

    // Level 2: 降级
    if (strategy.fallbackTool) {
      console.log(`[${toolCall.name}] 降级到: ${strategy.fallbackTool}`);
      return context.toolExecutor.execute({
        id: toolCall.id,
        name: strategy.fallbackTool,
        arguments: strategy.mapArgs?.(toolCall.arguments) ?? toolCall.arguments,
      }, context);
    }

    // Level 3: 人工兜底
    if (strategy.requireHumanIntervention) {
      return {
        toolCallId: toolCall.id,
        toolName: toolCall.name,
        status: 'pending_human',
        content: `工具 ${toolCall.name} 执行失败，需要人工介入:\n${error.message}`,
        duration: 0,
        error: 'requires_human',
      };
    }

    // 最终失败
    return {
      toolCallId: toolCall.id,
      toolName: toolCall.name,
      status: 'error',
      content: `工具执行失败（已尝试所有恢复策略）: ${error.message}`,
      duration: 0,
      error: error.message,
    };
  }
}

// 配置降级策略
const errorHandler = new ToolErrorHandler();
errorHandler.registerStrategy('network_error', {
  retry: true, maxRetries: 3, retryDelayMs: 1000,
});
errorHandler.registerStrategy('timeout', {
  retry: true, maxRetries: 1, retryDelayMs: 0,
  fallbackTool: 'quick_preview',  // 降级到更快但可能不完整的替代
});
errorHandler.registerStrategy('permission_denied', {
  retry: false,
  requireHumanIntervention: true,  // 权限不足必须人工处理
});

5.3.3 工具安全模型：权限分级、沙箱执行、操作审计

五级权限模型
// TypeScript: 五级工具权限模型
enum ToolPermission {
  READ = 0,          // 只读：read_file, search_code, list_directory
  WRITE = 1,         // 写入：write_file, create_directory
  EXECUTE = 2,       // 执行：run_command, run_test
  NETWORK = 3,       // 网络：api_get, api_post, deploy
  SYSTEM = 4,        // 系统：install_package, modify_config, sudo
}

class ToolPermissionManager {
  // 权限级联：高级权限自动包含低级权限
  private permissionGrants = new Map<ToolPermission, ToolPermission[]>([
    [ToolPermission.SYSTEM, [SYSTEM, NETWORK, EXECUTE, WRITE, READ]],
    [ToolPermission.NETWORK, [NETWORK, EXECUTE, WRITE, READ]],
    [ToolPermission.EXECUTE, [EXECUTE, WRITE, READ]],
    [ToolPermission.WRITE, [WRITE, READ]],
    [ToolPermission.READ, [READ]],
  ]);

  canExecute(userPermission: ToolPermission, toolRequirement: ToolPermission): boolean {
    const grants = this.permissionGrants.get(userPermission) || [];
    return grants.includes(toolRequirement);
  }

  // 高风险操作需要二次确认
  requiresConfirmation(toolPermission: ToolPermission): boolean {
    return toolPermission >= ToolPermission.EXECUTE;
  }
}
沙箱执行
// TypeScript: 沙箱执行器
class SandboxExecutor {
  async execute(
    fn: Function,
    context: SandboxContext
  ): Promise<any> {
    // 策略1: 进程隔离（最安全）
    if (context.isolation === 'process') {
      return this.executeInChildProcess(fn, context);
    }
    
    // 策略2: Docker 容器隔离（最彻底）
    if (context.isolation === 'container') {
      return this.executeInContainer(fn, context);
    }
    
    // 策略3: VM2 沙箱（Node.js 环境）
    if (context.isolation === 'vm') {
      return this.executeInVM(fn, context);
    }
    
    // 策略4: 直接执行（仅限已审查的安全工具）
    return fn();
  }

  private async executeInContainer(fn: Function, context: SandboxContext): Promise<any> {
    // 在临时 Docker 容器中执行
    const containerId = await docker.createContainer({
      Image: 'agent-sandbox:latest',
      Cmd: ['node', '-e', fn.toString()],
      HostConfig: {
        Memory: 512 * 1024 * 1024,    // 512MB 内存限制
        NanoCpus: 1 * 1e9,            // 1 CPU 限制
        NetworkMode: 'none',           // 无网络访问
        ReadonlyRootfs: true,         // 只读文件系统
        Binds: [`${context.workDir}:/workspace:ro`],
      },
    });

    await docker.startContainer(containerId);
    const output = await docker.waitContainer(containerId);
    const logs = await docker.getContainerLogs(containerId);
    await docker.removeContainer(containerId);

    return logs;
  }
}

5.3.4 实践案例：企业级工具链管理系统

// TypeScript: 企业级工具链管理系统
class EnterpriseToolManager {
  private engine: UniversalToolEngine;
  private mcpProvider: MCPToolProvider;
  private chainRegistry: Map<string, ToolChain> = new Map();
  private auditLog: ToolAuditLogger;
  private rateLimiter: RateLimiter;

  constructor(config: EnterpriseToolConfig) {
    this.engine = new UniversalToolEngine(config.engine);
    this.mcpProvider = new MCPToolProvider();
    this.auditLog = new ToolAuditLogger(config.audit);
    this.rateLimiter = new RateLimiter(config.rateLimit);
  }

  /** 启动工具系统 */
  async initialize(): Promise<void> {
    // 1. 连接所有 MCP Server
    for (const [name, serverConfig] of Object.entries(this.config.mcpServers)) {
      await this.mcpProvider.connectServer(name, serverConfig);
    }

    // 2. 发现并注册 MCP 工具
    const mcpTools = await this.mcpProvider.discoverAllTools();
    for (const tool of mcpTools) {
      this.engine.registerDelegated(tool, (args) => 
        this.mcpProvider.callTool(tool.name, args)
      );
    }

    // 3. 注册工具链
    this.chainRegistry.set('code_review', new ToolChain()
      .pipe('git_diff')
      .pipe('static_analysis')
      .pipe('security_scan')
      .pipe('generate_review_report')
    );

    this.chainRegistry.set('deploy', new ToolChain()
      .pipe('run_tests')
      .pipe('build_docker_image')
      .pipe('push_to_registry')
      .pipe('update_deployment')
      .pipe('health_check')
    );

    console.log(`[ToolManager] 初始化完成: ${this.getAllTools().length} 个工具`);
  }

  /** 带完整审计的工具执行 */
  async executeWithAudit(
    call: ToolCall, 
    context: ExecutionContext
  ): Promise<ToolResult> {
    // 速率限制
    const canProceed = await this.rateLimiter.checkLimit(
      context.userId, call.name
    );
    if (!canProceed) {
      return {
        toolCallId: call.id,
        toolName: call.name,
        status: 'error',
        content: '达到速率限制，请稍后再试',
        duration: 0,
        error: 'rate_limited',
      };
    }

    // 审计记录：开始
    const auditEntry = await this.auditLog.start(call, context);

    // 执行
    const result = await this.engine.executeBatch([call], context);

    // 审计记录：完成
    await this.auditLog.complete(auditEntry.id, result[0]);

    // 如果结果中包含敏感信息，脱敏
    const sanitized = await this.sanitizeResult(result[0]);

    return sanitized;
  }

  /** 执行工具链 */
  async executeChain(
    chainName: string,
    context: ExecutionContext
  ): Promise<ToolChainResult> {
    const chain = this.chainRegistry.get(chainName);
    if (!chain) throw new Error(`未找到工具链: ${chainName}`);
    
    console.log(`[ToolChain] 开始执行: ${chainName}`);
    return chain.execute({}, context);
  }

  private getAllTools(): string[] {
    return Array.from(this.engine.getDefinitions().map(d => d.name));
  }
}

5.4 本章小结

本章深入了 Harness 六大组件中最关键的行动组件——Tool Executor 和工具集成体系：

  1. Function Calling 原理：Agent 通过工具定义（JSON Schema）告知 Model 有哪些工具可用，Model 在推理时自主判断需要调用哪个工具，Harness 负责执行并将结果反馈回上下文。

  2. 工具注册与执行引擎：基于 Zod/Pydantic 的类型安全参数校验，五阶段执行管道（解析→校验→安全检查→执行→格式化），并发生命周期管理。

  3. MCP 协议：Model Context Protocol 将工具提供方和使用方解耦，通过 stdio/HTTP/WebSocket 三种传输方式，实现了 Agent 与外部工具生态的标准化连接。

  4. 工具链设计模式：顺序链、条件分支、并行执行三种组合模式；超时→重试→降级→人工兜底的四级错误恢复链；五级权限模型+沙箱执行的安全模型。

工具集成的终极目标是：让 Agent 拥有安全、可靠、高效的行动能力——什么都能做，但什么都不能乱做。

关键术语

术语 英文 定义
Function Calling Function Calling 让 LLM 调用外部函数/工具的能力
JSON Schema JSON Schema 描述 JSON 数据结构的规范，用于定义工具参数
MCP Model Context Protocol Agent 与工具/资源服务的标准化通信协议
工具链 Tool Chain 多个工具按顺序/条件/并行的组合执行
沙箱执行 Sandbox Execution 在隔离环境中执行代码以防止安全风险
人工兜底 Human-in-the-Loop 自动处理失败时的人工介入机制

思考与练习

  1. Function Calling 实战：用 Zod 定义 5 个工具的参数 Schema，然后注册到 TypedToolRegistry。尝试传入合法和不合法的参数，观察 Zod 校验的错误信息。

  2. MCP Server 构建：实现一个 MCP Server，提供 weather_query(查询天气) 和 news_search(搜索新闻) 两个工具。用 MCP Client 连接并验证工具能正常工作。

  3. 工具链设计：为"代码部署"场景设计一个工具链（测试→构建→推送→部署→健康检查），实现每个工具的 stub（返回模拟结果），测试整个链的串联执行。

  4. 安全模型设计：为你的 Agent 工具系统设计权限分级方案。哪些工具需要人工确认？哪些工具可以自动执行？考虑以下场景：删除文件、修改数据库、调用外部 API、执行 Shell 命令、安装依赖包。

  5. MCP 生态探索：浏览 MCP 官方 registry（https://registry.modelcontextprotocol.io），找出 3 个对你的 Agent 最有用的第三方 MCP Server，尝试集成它们。

Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

【Mimo Code使用攻略】

Mimo Code推出免费体验Mimo-pro 2.5模型服务，但用户在安装chrome-devtools浏览器自动化插件时遇到困难。本文提供了详细解决方案：首先通过npm全局安装chrome-devtools-mcp插件，然后配置OpenCode MCP服务器，再以调试模式启动Chrome浏览器。安装完成后，AI助手可实现自动操作浏览器、网页分析、表单填写等功能。文章包含Windows/macO

avatar MCP技术社区

MCP 官方规范：stdio 为本地服务端默认通信载体（权威原文依据）

依靠操作系统管道实现进程间双向通信，无网络端口、无 TCP 开销Model Cont...。客户端（Claude Desktop、AI IDE）将 MCP Server 作为。分割，JSON 内部不能自带换行；部分 SDK 会追加。每条 JSON 消息以换行。

avatar MCP技术社区
cover

Windsurf官宣正式改名！太夯了！

avatar MCP技术社区