阶段 2：工具集成 ⭐ 必学

一句话总结：工具是 Agent 的手脚——定义好工具，Agent 就能做事。

📊 学习进度
状态：⬜ 未开始
预计时长：3-4 小时
已完成：0/3 个模块
在整体流程中的位置：AI Agent 开发·第 2 阶段

📍 本章定位
服务方案：方案 1（重要 60%）/ 方案 3（核心 80%）
学习方式：⭐ 必学
在流程中的作用：让 Agent 具备执行能力
核心知识点：工具定义、搜索工具、代码执行、API 调用
预计时长：3-4 小时
完成后能做什么：能为 Agent 注册多种工具

1. 传统模式：痛点与瓶颈

1.1 传统工具集成的局限性

传统 AI 应用的工具集成是"硬编码"模式——开发者在代码中直接写死 API 调用逻辑，每增加一个工具都需要修改代码、重新部署。这种模式在工具数量少时可行，但随着工具增多，维护成本呈指数增长。

传统工具集成的典型问题：

1.2 量化痛点数据

痛点维度	传统硬编码	MCP 标准化	改善幅度
新增工具开发时间	4-8 小时	30 分钟	90%
工具代码复用率	0%（每个项目重写）	80%（跨项目复用）	+80%
工具描述准确率	依赖开发者经验	标准化 Schema	+35%
跨框架兼容性	0%	90%	+90%
工具调试时间	2-4 小时	30 分钟	85%

数据来源：Anthropic MCP 2025 白皮书、社区开发者调查

1.3 OPC 场景下的核心矛盾

OPC 运营者需要快速集成各种工具（搜索、数据库、API、文件操作），但没有时间为每个项目从零编写工具集成代码。MCP 协议的出现解决了这个问题——一次编写，到处复用。

2. OPC 模式：重新定义

2.1 核心理念

MCP（Model Context Protocol）是 Anthropic 于 2024 年推出的标准化工具协议。它的核心理念是：将工具的定义、实现、调用标准化，让不同 Agent 框架可以共享同一套工具。

MCP 架构图：

2.2 人机分工矩阵

任务	人类角色	AI 角色	协作方式
选择 MCP 服务器	根据需求选择	推荐最佳选项	人决策，AI 建议
配置 MCP 服务器	设置认证信息	生成配置文件	人提供密钥，AI 生成配置
开发自定义 MCP 服务器	定义工具接口	生成代码	人审核，AI 实现
安全策略	定义权限边界	实现权限检查	人决策，AI 执行

2.3 效率对比

指标	传统硬编码	MCP 标准化	提效倍数
新增工具时间	4-8 小时	30 分钟	8-16x
代码复用率	0%	80%	∞
跨框架兼容	需重写	直接可用	∞
维护成本	高（每个工具独立）	低（统一协议）	3-5x

3. 实操案例

3.1 场景描述

场景：为 OPC 的竞品分析 Agent 集成以下工具：

Brave Search — 搜索竞品最新动态
Filesystem — 读写分析报告文件
PostgreSQL — 查询历史数据

技术栈：Claude API + Python + MCP SDK

3.2 执行过程

3.2.1 MCP 协议详解

MCP 使用 JSON-RPC 2.0 协议进行通信，支持两种传输方式：

传输方式	适用场景	说明
stdio	本地工具	通过标准输入输出通信，延迟低
SSE	远程工具	通过 HTTP 通信，支持远程部署

MCP 核心概念：

概念	说明	示例
Tool	可调用的函数	`search_web(query)`
Resource	可读取的数据	`file:///path/to/file`
Prompt	预定义的 Prompt 模板	`analyze_competitor(name)`

3.2.2 使用现有 MCP 服务器

Python 配置：

python

# 安装 MCP SDK
# pip install mcp

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 配置 MCP 服务器
server_params = StdioServerParameters(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-brave-search"],
    env={"BRAVE_API_KEY": "your-api-key"}
)

# 连接 MCP 服务器
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # 初始化连接
        await session.initialize()
        
        # 列出可用工具
        tools = await session.list_tools()
        print(f"可用工具: {[t.name for t in tools.tools]}")
        
        # 调用搜索工具
        result = await session.call_tool(
            "brave_search",
            arguments={"query": "OpenAI GPT-5 latest news"}
        )
        print(f"搜索结果: {result}")

Claude Desktop 配置（claude_desktop_config.json）：

json

{
  "mcpServers": {
    "brave-search": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-brave-search"],
      "env": {
        "BRAVE_API_KEY": "your-api-key"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/dir"]
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://user:pass@localhost/db"]
    }
  }
}

3.2.3 开发自定义 MCP 服务器

当社区没有现成的 MCP 服务器时，可以自己开发：

Python 自定义 MCP 服务器：

python

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json

# 创建 MCP 服务器
server = Server("my-custom-tools")

# 定义工具列表
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_competitor_news",
            description="获取竞品最新新闻动态。输入竞品名称，返回最近 7 天的新闻摘要。",
            inputSchema={
                "type": "object",
                "properties": {
                    "competitor_name": {
                        "type": "string",
                        "description": "竞品名称，如 'OpenAI', 'Anthropic'"
                    },
                    "days": {
                        "type": "integer",
                        "description": "查询天数，默认 7",
                        "default": 7
                    }
                },
                "required": ["competitor_name"]
            }
        ),
        Tool(
            name="analyze_sentiment",
            description="分析文本的情感倾向。输入文本内容，返回正面/负面/中性及置信度。",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {
                        "type": "string",
                        "description": "要分析的文本内容"
                    }
                },
                "required": ["text"]
            }
        )
    ]

# 实现工具执行逻辑
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_competitor_news":
        competitor = arguments["competitor_name"]
        days = arguments.get("days", 7)
        # 实际项目中调用新闻 API
        result = {
            "competitor": competitor,
            "period": f"最近 {days} 天",
            "news": [
                {"title": f"{competitor} 发布新产品", "date": "2025-06-10"},
                {"title": f"{competitor} 获得融资", "date": "2025-06-08"}
            ]
        }
        return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
    
    elif name == "analyze_sentiment":
        text = arguments["text"]
        # 实际项目中调用情感分析模型
        return [TextContent(type="text", text=json.dumps({
            "sentiment": "positive",
            "confidence": 0.85
        }))]
    
    raise ValueError(f"Unknown tool: {name}")

# 启动服务器
async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

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

TypeScript 自定义 MCP 服务器：

typescript

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  { name: "my-custom-tools", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// 定义工具列表
server.setRequestHandler("tools/list", async () => ({
  tools: [
    {
      name: "get_competitor_news",
      description: "获取竞品最新新闻动态",
      inputSchema: {
        type: "object",
        properties: {
          competitor_name: {
            type: "string",
            description: "竞品名称",
          },
          days: {
            type: "integer",
            description: "查询天数",
            default: 7,
          },
        },
        required: ["competitor_name"],
      },
    },
  ],
}));

// 实现工具执行
server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "get_competitor_news") {
    const competitor = args?.competitor_name as string;
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify({
            competitor,
            news: [
              { title: `${competitor} 发布新产品`, date: "2025-06-10" },
            ],
          }),
        },
      ],
    };
  }

  throw new Error(`Unknown tool: ${name}`);
});

// 启动服务器
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);

3.2.4 工具描述的最佳实践

工具描述的质量直接影响 Agent 选择工具的准确率。据 Anthropic 2025 年测试，优化工具描述可将调用准确率从 78% 提升至 94%。

工具描述优化清单：

要素	差的描述	好的描述
功能说明	"搜索"	"搜索网页最新信息，返回标题、摘要和链接"
使用时机	无	"当用户询问实时信息、新闻、最新动态时使用"
参数说明	"query: 搜索词"	"query: 搜索关键词，建议使用英文，2-5 个词"
返回格式	无	"返回 JSON 格式，包含 title、snippet、url 字段"
限制说明	无	"每次最多返回 10 条结果，不支持中文搜索"

优化示例：

python

# 优化前（调用准确率 78%）
{
    "name": "search",
    "description": "搜索网页",
    "input_schema": {
        "type": "object",
        "properties": {
            "q": {"type": "string"}
        }
    }
}

# 优化后（调用准确率 94%）
{
    "name": "brave_web_search",
    "description": """搜索网页最新信息。适用于以下场景：
    - 查询实时新闻和最新动态
    - 获取产品发布信息
    - 查找技术文档和教程
    不适用于：查询历史数据、数据库操作""",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词，建议 2-5 个英文词，如 'GPT-5 release date'"
            },
            "count": {
                "type": "integer",
                "description": "返回结果数量，默认 5，最大 10",
                "default": 5
            }
        },
        "required": ["query"]
    }
}

3.3 实用工具实现示例

3.3.1 代码执行工具（沙箱环境）

让 Agent 执行 Python 代码是数据分析场景的核心能力。关键是使用沙箱隔离执行，防止恶意代码。

python

import subprocess
import tempfile
import os
from typing import Dict

class CodeExecutionTool:
    """安全的代码执行工具 - 使用 Docker 沙箱"""

    def __init__(self, timeout: int = 30, memory_limit: str = "256m"):
        self.timeout = timeout
        self.memory_limit = memory_limit

    def execute(self, code: str, language: str = "python") -> Dict:
        """在沙箱中执行代码"""
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=f".{language}", delete=False
        ) as f:
            f.write(code)
            temp_file = f.name

        try:
            result = subprocess.run(
                [
                    "docker", "run", "--rm",
                    "--memory", self.memory_limit,
                    "--network", "none",  # 禁止网络访问
                    "-v", f"{temp_file}:/code/script.{language}",
                    "python:3.11-slim",
                    "python", f"/code/script.{language}"
                ],
                capture_output=True,
                text=True,
                timeout=self.timeout
            )

            return {
                "stdout": result.stdout,
                "stderr": result.stderr,
                "exit_code": result.returncode,
                "success": result.returncode == 0
            }
        except subprocess.TimeoutExpired:
            return {"error": "执行超时", "success": False}
        finally:
            os.unlink(temp_file)

# 注册为 MCP 工具
code_tool = {
    "name": "execute_python",
    "description": """在安全沙箱中执行 Python 代码。适用于：
    - 数据分析和计算
    - 图表生成
    - 文件处理
    不适用于：网络请求、系统操作、长时间运行的任务""",
    "input_schema": {
        "type": "object",
        "properties": {
            "code": {
                "type": "string",
                "description": "要执行的 Python 代码"
            }
        },
        "required": ["code"]
    }
}

3.3.2 MCP Resources 实现

MCP 不仅支持 Tool（工具调用），还支持 Resource（数据读取）。Resource 适合暴露只读数据源：

python

from mcp.server import Server
from mcp.types import Resource, TextContent
import json

server = Server("data-resources")

@server.list_resources()
async def list_resources():
    """列出可用的数据资源"""
    return [
        Resource(
            uri="config://app/settings",
            name="应用配置",
            description="当前 Agent 的配置信息",
            mimeType="application/json"
        ),
        Resource(
            uri="data://market/latest",
            name="最新市场数据",
            description="最近一次采集的市场数据",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    """读取资源内容"""
    if uri == "config://app/settings":
        return json.dumps({
            "model": "claude-sonnet-4-20250514",
            "max_iterations": 10,
            "token_budget": 4000
        })
    elif uri == "data://market/latest":
        return json.dumps({
            "BTC": {"price": 67500, "timestamp": "2025-06-15T10:00:00Z"},
            "ETH": {"price": 3500, "timestamp": "2025-06-15T10:00:00Z"}
        })
    raise ValueError(f"Unknown resource: {uri}")

3.3.3 数据库查询工具

为 Agent 提供安全的数据库查询能力，关键是限制查询范围和防止 SQL 注入：

python

import asyncpg
from typing import List, Dict

class DatabaseQueryTool:
    """安全的数据库查询工具"""

    # 允许查询的表和列（白名单）
    ALLOWED_TABLES = {
        "market_data": ["symbol", "price", "volume", "timestamp"],
        "trading_signals": ["symbol", "action", "confidence", "created_at"],
        "portfolio": ["symbol", "quantity", "avg_price"]
    }

    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.pool = None

    async def initialize(self):
        """初始化连接池"""
        self.pool = await asyncpg.create_pool(self.connection_string, min_size=2, max_size=10)

    async def query(self, table: str, conditions: Dict = None, limit: int = 100) -> List[Dict]:
        """安全查询 - 使用参数化查询防止注入"""
        if table not in self.ALLOWED_TABLES:
            return [{"error": f"表 {table} 不在允许列表中"}]

        allowed_columns = self.ALLOWED_TABLES[table]
        columns = ", ".join(allowed_columns)

        query = f"SELECT {columns} FROM {table}"
        params = []

        if conditions:
            where_clauses = []
            for i, (key, value) in enumerate(conditions.items()):
                if key not in allowed_columns:
                    continue
                where_clauses.append(f"{key} = ${i + 1}")
                params.append(value)
            if where_clauses:
                query += " WHERE " + " AND ".join(where_clauses)

        query += f" LIMIT {min(limit, 1000)}"  # 硬性限制最大返回行数

        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query, *params)
            return [dict(row) for row in rows]

3.4 前后对比

维度	传统硬编码	MCP 标准化	改善
新增工具时间	4-8 小时	30 分钟	90%
代码复用率	0%	80%	+80%
跨项目复用	不支持	直接复用	∞
调试效率	手动日志	MCP Inspector	85%
社区生态	无	100+ MCP 服务器	∞

4. 趋势预判（未来 1-3 年）

4.1 技术演进方向

技术方向	当前状态	1 年后	3 年后
MCP 服务器数量	100+	500+	2000+
主流框架支持	部分支持	全面支持	原生集成
远程 MCP 服务器	实验阶段	生产就绪	标准部署
安全标准	缺失	初步建立	完善

4.2 角色变化趋势

角色	当前	1 年后	3 年后
MCP 服务器开发者	小众	增长	生态核心
工具集成工程师	手动集成	MCP 配置	自动发现
Agent 安全审计	几乎不存在	新兴岗位	标准配置

4.3 OPC 需要提前准备的能力

MCP 协议理解：掌握 JSON-RPC 2.0、stdio/SSE 传输
工具接口设计：写出高质量的工具描述和 Schema
安全意识：理解最小权限原则，设计安全边界
调试能力：使用 MCP Inspector 调试工具调用

5. 核心洞察

🔑 关键洞察

MCP 协议的价值不在于技术复杂度，而在于标准化。就像 USB 统一了外设接口，MCP 统一了 AI 工具接口。掌握 MCP，意味着你的工具可以在任何支持 MCP 的 Agent 框架中复用。

⚠️ 安全警告

MCP 服务器可以访问文件系统、数据库、API 等敏感资源。务必遵循最小权限原则——只授予 Agent 完成任务所需的最小权限。一个权限过大的 MCP 服务器，可能导致数据泄露或资产损失。

6. 参考与延伸

[1] Model Context Protocol. "MCP Specification" — MCP 协议官方文档（2025）

[2] MCP Servers. "GitHub Repository" — 官方 MCP 服务器集合（2025）

[3] Anthropic. "MCP Announcement" — MCP 协议发布博客（2024）

[4] MCP Inspector. "Debug Tool" — MCP 工具调试工具（2025）

[5] Smithery. "MCP Registry" — MCP 服务器注册中心（2025）

[6] Anthropic. "MCP Resources" — MCP 资源规范（2025）

[7] OWASP. "AI Security Top 10" — AI 工具安全风险清单（2025）

[8] NIST. "AI Risk Management Framework" — AI 风险管理框架（2025）

[9] MCP Community. "MCP Servers GitHub" — 社区 MCP 服务器集合（2025）

[10] Anthropic. "Tool Use Best Practices" — 工具使用最佳实践（2025）

[11] MCP Specification. "Error Handling" — MCP 错误处理规范（2025）

[12] GitHub. "MCP Registry" — MCP 服务器注册中心（2025）

MCP 工具类型对比

MCP 协议定义了三种核心原语，各有不同的适用场景：

类型	用途	数据流向	示例	适用场景
Tool	执行操作	Agent → 外部	`search_web(query)`	需要副作用的操作
Resource	读取数据	外部 → Agent	`config://app/settings`	只读数据源
Prompt	预定义模板	预定义 → Agent	`analyze_competitor(name)`	标准化工作流

选择指南：

需要执行操作（搜索、写入、调用 API）→ 用 Tool
需要读取只读数据（配置、状态、数据源）→ 用 Resource
需要标准化的 Prompt 模板 → 用 Prompt

混合使用示例：

python

# 一个 MCP 服务器同时提供 Tool、Resource、Prompt
@server.list_tools()
async def list_tools():
    return [search_tool, write_file_tool]

@server.list_resources()
async def list_resources():
    return [config_resource, data_resource]

@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="daily_report",
            description="生成每日分析报告",
            arguments=[
                PromptArgument(name="date", description="报告日期", required=True)
            ]
        )
    ]

MCP 生态全景

主流 MCP 服务器列表

截至 2025 年 6 月，MCP 生态已有 100+ 个社区贡献的服务器。以下是最常用的 MCP 服务器分类：

类别	服务器	功能	安装方式
搜索	Brave Search	网页搜索	`npx @modelcontextprotocol/server-brave-search`
搜索	Tavily	AI 优化搜索	`npx @anthropic/tavily-mcp`
文件	Filesystem	文件读写	`npx @modelcontextprotocol/server-filesystem`
数据库	PostgreSQL	数据库查询	`npx @modelcontextprotocol/server-postgres`
数据库	SQLite	本地数据库	`npx @modelcontextprotocol/server-sqlite`
Git	Git	代码仓库操作	`npx @modelcontextprotocol/server-git`
Web	Fetch	HTTP 请求	`npx @modelcontextprotocol/server-fetch`
自动化	Puppeteer	浏览器自动化	`npx @anthropic/puppeteer-mcp`
通信	Slack	消息发送	`npx @anthropic/slack-mcp`
云服务	AWS S3	文件存储	`npx @anthropic/aws-s3-mcp`

MCP 服务器发现与评估

选择 MCP 服务器时，需要评估以下维度：

评估维度	说明	检查方法
活跃度	最近 3 个月有更新	GitHub commits
文档质量	有完整的 README 和示例	查看文档
安全性	有权限控制和审计日志	代码审查
性能	响应时间 < 2 秒	基准测试
兼容性	支持最新 MCP 协议版本	版本检查

MCP 服务器评估清单：

python

class MCPServerEvaluator:
    """MCP 服务器评估器"""

    def evaluate(self, server_info: dict) -> dict:
        """评估 MCP 服务器质量"""
        scores = {}

        # 1. 活跃度评估
        last_commit = server_info.get("last_commit_days_ago", 999)
        scores["activity"] = max(0, 1 - last_commit / 90)  # 90 天内线性衰减

        # 2. 文档质量评估
        has_readme = server_info.get("has_readme", False)
        has_examples = server_info.get("has_examples", False)
        scores["documentation"] = (0.5 if has_readme else 0) + (0.5 if has_examples else 0)

        # 3. 安全性评估
        has_auth = server_info.get("has_auth", False)
        has_audit = server_info.get("has_audit_log", False)
        scores["security"] = (0.6 if has_auth else 0) + (0.4 if has_audit else 0)

        # 4. 综合评分
        overall = sum(scores.values()) / len(scores)

        return {
            "scores": scores,
            "overall": overall,
            "recommendation": "使用" if overall > 0.7 else "谨慎使用" if overall > 0.4 else "不推荐"
        }

工具集成最佳实践

工具编排模式

在复杂场景中，多个工具需要按特定顺序或条件调用。以下是三种常见的工具编排模式：

模式	说明	适用场景	实现方式
顺序编排	工具按固定顺序执行	数据处理流水线	Chain 模式
条件编排	根据结果选择不同工具	决策树、路由	Router 模式
并行编排	多个工具同时执行	数据聚合、批量处理	Parallel 模式

顺序编排示例：

python

class ToolChain:
    """工具链 - 顺序执行多个工具"""

    def __init__(self, tools: list):
        self.tools = tools

    async def execute(self, input_data: dict) -> dict:
        """按顺序执行工具链"""
        result = input_data
        for tool in self.tools:
            result = await tool.execute(result)
        return result

# 使用示例
chain = ToolChain([
    SearchTool(query="latest crypto news"),
    SummarizeTool(max_length=200),
    TranslateTool(target_lang="zh")
])
result = await chain.execute({"query": "Bitcoin ETF"})

条件编排示例：

python

class ToolRouter:
    """工具路由器 - 根据条件选择工具"""

    def __init__(self, routes: dict):
        self.routes = routes  # {condition: tool}

    async def execute(self, input_data: dict) -> dict:
        """根据条件路由到对应工具"""
        for condition, tool in self.routes.items():
            if condition(input_data):
                return await tool.execute(input_data)
        raise ValueError("No matching route")

# 使用示例
router = ToolRouter({
    lambda x: x.get("type") == "price": PriceTool(),
    lambda x: x.get("type") == "news": NewsTool(),
    lambda x: x.get("type") == "analysis": AnalysisTool()
})
result = await router.execute({"type": "price", "symbol": "BTC"})

工具错误处理

工具调用失败是生产环境中的常见问题。健壮的错误处理机制是 Agent 可靠运行的关键。

工具错误分类与处理策略：

错误类型	原因	处理策略	是否可重试
超时	网络延迟、服务端慢	增加超时时间、重试	是
认证失败	API Key 无效/过期	更新凭证	否
速率限制	调用频率过高	退避重试、限流	是
参数错误	Schema 不匹配	修正参数、重新调用	否
服务不可用	外部服务宕机	降级、备用工具	是

工具错误处理实现：

python

import asyncio
from typing import Optional, Callable

class ToolErrorHandler:
    """工具错误处理器"""

    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    async def execute_with_retry(
        self,
        tool_func: Callable,
        args: dict,
        fallback: Optional[Callable] = None
    ) -> dict:
        """带重试和降级的工具执行"""
        last_error = None

        for attempt in range(self.max_retries):
            try:
                result = await tool_func(args)
                return {"success": True, "result": result, "attempts": attempt + 1}
            except TimeoutError:
                last_error = "超时"
                delay = self.base_delay * (2 ** attempt)
                await asyncio.sleep(delay)
            except RateLimitError:
                last_error = "速率限制"
                delay = self.base_delay * (2 ** attempt) * 2  # 更长的退避
                await asyncio.sleep(delay)
            except AuthenticationError:
                last_error = "认证失败"
                break  # 不可重试
            except Exception as e:
                last_error = str(e)
                break  # 未知错误不重试

        # 降级到备用工具
        if fallback:
            try:
                result = await fallback(args)
                return {"success": True, "result": result, "fallback": True}
            except Exception:
                pass

        return {"success": False, "error": last_error, "attempts": self.max_retries}

常见问题

问题	原因	解决方案
工具不被调用	描述不清晰	优化 description，添加使用场景
参数错误	Schema 不准确	完善 input_schema，添加 enum 约束
安全风险	工具权限过大	最小权限原则，限制访问范围
MCP 服务器启动失败	环境变量缺失	检查 API Key 等配置
工具调用超时	网络问题或服务端慢	设置 timeout，添加重试机制

下一步

完成工具集成后，进入阶段 3：状态管理 — 学习如何让 Agent 记住上下文，实现跨会话的持久化记忆。

阶段 2：工具集成 ⭐ 必学 ​

1. 传统模式：痛点与瓶颈 ​

1.1 传统工具集成的局限性 ​

1.2 量化痛点数据 ​

1.3 OPC 场景下的核心矛盾 ​

2. OPC 模式：重新定义 ​

2.1 核心理念 ​

2.2 人机分工矩阵 ​

2.3 效率对比 ​

3. 实操案例 ​

3.1 场景描述 ​

3.2 执行过程 ​

3.2.1 MCP 协议详解 ​

3.2.2 使用现有 MCP 服务器 ​

3.2.3 开发自定义 MCP 服务器 ​

3.2.4 工具描述的最佳实践 ​

3.3 实用工具实现示例 ​

3.3.1 代码执行工具（沙箱环境） ​

3.3.2 MCP Resources 实现 ​

3.3.3 数据库查询工具 ​

3.4 前后对比 ​

4. 趋势预判（未来 1-3 年） ​

4.1 技术演进方向 ​

4.2 角色变化趋势 ​

4.3 OPC 需要提前准备的能力 ​

5. 核心洞察 ​

6. 参考与延伸 ​

MCP 工具类型对比 ​

MCP 生态全景 ​

主流 MCP 服务器列表 ​

MCP 服务器发现与评估 ​

工具集成最佳实践 ​

工具编排模式 ​

工具错误处理 ​

常见问题 ​

下一步 ​

阶段 2：工具集成 ⭐ 必学

1. 传统模式：痛点与瓶颈

1.1 传统工具集成的局限性

1.2 量化痛点数据

1.3 OPC 场景下的核心矛盾

2. OPC 模式：重新定义

2.1 核心理念

2.2 人机分工矩阵

2.3 效率对比

3. 实操案例

3.1 场景描述

3.2 执行过程

3.2.1 MCP 协议详解

3.2.2 使用现有 MCP 服务器

3.2.3 开发自定义 MCP 服务器

3.2.4 工具描述的最佳实践

3.3 实用工具实现示例

3.3.1 代码执行工具（沙箱环境）

3.3.2 MCP Resources 实现

3.3.3 数据库查询工具

3.4 前后对比

4. 趋势预判（未来 1-3 年）

4.1 技术演进方向

4.2 角色变化趋势

4.3 OPC 需要提前准备的能力

5. 核心洞察

6. 参考与延伸

MCP 工具类型对比

MCP 生态全景

主流 MCP 服务器列表

MCP 服务器发现与评估

工具集成最佳实践

工具编排模式

工具错误处理

常见问题

下一步