企业官网建设流程全解析

1. 这不是又一个“AI Agent 概念课”，而是一份能直接跑通的工程化开发手记

你点开这个标题，大概率是被“B站唯一”“从入门到代码实战”“2026最新版”这几个词钩住的。别急着划走——我就是那个在去年底把 Langchain 0.1.x 版本踩进坑里、今年初用 Langgraph Beta 调通第一个带工具调用的 A2A 流程、上个月刚把 MCP 协议跑通在 Playwright + Chrome DevTools 环境里的开发者。这不是录屏剪辑出来的“概念演示”，而是我把三个月里每天凌晨两点前的调试日志、失败截图、重装环境的命令记录、和三个不同团队同事反复对齐的协议字段定义，全部揉碎了重新组织成的一份可执行文档。

核心关键词就五个：MCP、Agent、A2A、Langchain、Langgraph。它们不是并列关系，而是层层嵌套的工程链条：MCP 是通信层协议，A2A（Agent-to-Agent）是交互范式，Agent 是运行实体，Langchain 是早期胶水框架，Langgraph 是当前最接近生产级的编排引擎。很多人卡在“知道名词但写不出第一行可运行代码”的阶段，根本原因不是学得不够多，而是没人告诉你：MCP 的wss://api.xiaozhi.me/mcp/?token=...这串 URL 里，token 解析失败会导致整个 Agent 启动时静默崩溃；Langgraph 的StateGraph必须显式声明add_edge的条件分支，否则工具调用后流程直接卡死；A2A 不是“两个 Agent 聊天”，而是必须定义清晰的message_schema和capability_negotiation机制，否则跨框架通信就是空中楼阁。

这份内容适合三类人：一是刚学完 Langchain 基础 API、想真正让 Agent “下地干活”的中级开发者；二是正在评估 MCP 协议落地可行性、需要真实链路验证的技术负责人；三是被“Agent 架构”“Agent 平台”这类术语绕晕、急需看到端到端数据流的同学。它不讲大模型原理，不对比 Llama vs Qwen，不分析 Transformer 层数——只聚焦一件事：如何让一个 Python 进程，通过标准协议，调用浏览器、调用数据库、调用另一个 Agent，并把结果可靠地返回给前端。下面所有内容，都来自我本地~/projects/mcp-a2a-demo目录下真实的git log记录和docker logs截图。

2. 为什么必须用 MCP + Langgraph 组合？而不是继续用 Langchain Chain？

2.1 Langchain 的“胶水困境”：当 Chain 遇到真实业务场景

Langchain 最初设计目标很明确：把 LLM 调用、Prompt 工程、向量检索这些离散能力，用Chain串成流水线。它成功了，但也埋下了硬伤。我拿一个真实需求举例：用户输入“帮我查一下昨天下午3点到5点，订单号以‘ORD-2024’开头的支付失败记录，并生成一张柱状图”。这个需求拆解后需要：

• 步骤1：用 LLM 理解时间范围、订单号前缀、失败状态等语义 → LangchainLLMChain
• 步骤2：连接 MySQL 执行 SQL 查询 → LangchainSQLDatabaseChain
• 步骤3：把查询结果喂给 Python 的matplotlib画图 → LangchainPythonAstREPLTool
• 步骤4：把图片 Base64 返回给前端 → LangchainTool返回格式处理

表面看，Chain 完全能覆盖。但实际跑起来，问题立刻暴露：

提示：Langchain 的SQLDatabaseChain默认会把原始 SQL 结果转成字符串再喂给 LLM，导致数值精度丢失；PythonAstREPLTool无法捕获matplotlib的异常（比如内存不足），错误直接吞掉；更致命的是，Chain 的执行是线性的，无法在步骤2失败时自动降级到“查最近7天失败记录”这种业务逻辑。

我试过用RouterChain做分支，但它的路由逻辑写在 Prompt 里，每次都要重新 inference，响应延迟从 800ms 涨到 3.2s。后来换成MultiRouteChain，又发现它要求每个子 Chain 的输入输出 schema 必须完全一致，而 SQL 查询返回的是 DataFrame，画图需要的是 numpy array，强行转换导致类型错误频发。

2.2 Langgraph 的“状态机革命”：用显式状态替代隐式流程

Langgraph 的核心突破，是把“流程控制权”从 LLM Prompt 里夺回来，交给开发者用代码定义。它基于StateGraph构建有向无环图（DAG），每个节点是一个纯函数（node），边（edge）是明确的条件判断。还是上面那个需求，用 Langgraph 实现：

from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any class AgentState(TypedDict): user_input: str sql_result: List[Dict] | None chart_data: bytes | None error: str | None def parse_query(state: AgentState) -> AgentState: # 这里调用 LLM 解析时间、订单号等，返回结构化参数 return {"parsed_params": {...}} def execute_sql(state: AgentState) -> AgentState: try: result = db.query( start_time=state["parsed_params"]["start"], end_time=state["parsed_params"]["end"], prefix=state["parsed_params"]["prefix"] ) return {"sql_result": result} except Exception as e: return {"error": f"SQL执行失败: {str(e)}"} def generate_chart(state: AgentState) -> AgentState: if not state["sql_result"]: return {"error": "无数据可绘图"} # 生成图表，返回 bytes return {"chart_data": chart_bytes} # 定义图 workflow = StateGraph(AgentState) workflow.add_node("parse", parse_query) workflow.add_node("sql", execute_sql) workflow.add_node("chart", generate_chart) # 显式定义边：parse -> sql，sql -> chart 或 sql -> error 处理 workflow.add_edge("parse", "sql") workflow.add_conditional_edges( "sql", lambda x: "error" in x and x["error"] is not None, {True: "handle_error", False: "chart"} ) workflow.add_edge("chart", END)

关键差异在哪？

• 错误可捕获：execute_sql的try/except直接控制流向handle_error节点，不依赖 LLM 判断；
• 状态可追踪：AgentState是 TypedDict，IDE 能自动补全state["sql_result"]，不会出现state.get("result")这种运行时才报错的写法；
• 扩展性明确：要加“发送邮件通知”节点？只需新增send_email函数，再加一条workflow.add_edge("chart", "send_email")，无需改任何 Prompt。

2.3 MCP 协议：让 Agent 不再是孤岛，而是可插拔的网络服务

Langgraph 解决了单个 Agent 内部的编排问题，但真实系统里，Agent 往往需要协作。比如“查订单”Agent 查完数据后，要把结果传给“生成报告”Agent，后者再调用“邮件发送”Agent。传统做法是写 REST API，但很快遇到问题：

• 每个 Agent 都要自己实现 HTTP Server、鉴权、重试、超时；
• 不同 Agent 用的框架不同（Python/Go/JS），JSON Schema 对不上；
• 一个 Agent 更新了接口，其他所有调用方都要同步改。

MCP（Model Communication Protocol）就是为解决这个而生的。它不是 RPC 框架，而是定义了一套标准化的消息交换契约。核心就三点：

1. 传输层统一用 WebSocket：wss://api.xiaozhi.me/mcp/?token=eyjhbgcioijfuzi1niisinr5cci6ikpxvcj9.eyj...这个 URL 里的 token 是 JWT，解析后包含scope字段，决定该连接能调用哪些 capability；
2. 消息体强制 JSON-RPC 2.0 格式：

   { "jsonrpc": "2.0", "id": "req-123", "method": "tool_call", "params": { "tool_name": "query_orders", "arguments": {"start": "2024-05-15T15:00:00Z", "prefix": "ORD-2024"} } }

3. Capability 注册机制：Agent 启动时，先发register_capability消息，声明自己支持query_orders、send_email等能力，其他 Agent 通过list_capabilities发现服务。

我实测下来，MCP 最大的价值不是“技术先进”，而是大幅降低协作成本。上周我们团队三个小组分别开发“数据查询”“图表生成”“邮件推送”三个 Agent，约定好都用 MCP 协议，最后联调只花了 2 小时——因为大家只关心：我的params字段是否符合对方注册的capability_schema，而不关心对方用的是 Flask 还是 FastAPI。

3. 从零搭建一个可运行的 MCP+Langgraph A2A 系统：环境、依赖与初始化

3.1 环境准备：为什么必须用 Python 3.11+ 和 Poetry？

很多教程还在用pip install langchain langgraph，这在新版本里会出问题。Langgraph 0.1.0+ 强依赖pydantic>=2.5，而旧版 Langchain 的langchain-core与之冲突。我踩过的坑：用pip install langchain[all]会强制降级pydantic到 1.x，导致 Langgraph 的StateGraph初始化时报ValidationError。

正确姿势是用Poetry管理依赖，它能精确锁定子依赖版本。初始化命令：

poetry init -n poetry add "langgraph>=0.1.12" "langchain>=0.1.20" "langchain-community>=0.0.35" poetry add "websocket-client>=1.7.0" "requests>=2.31.0" # MCP 客户端必需 poetry add "playwright>=1.42.0" # 后续浏览器自动化用 poetry shell

注意：playwright必须单独安装浏览器二进制，poetry add playwright只装 Python 包。执行playwright install chromium，否则后续BrowserTool会报BrowserType.launch: Executable doesn't exist。

Python 版本必须 ≥3.11。Langgraph 的asyncio事件循环优化（尤其是stream_events方法）在 3.10 下有竞态问题，我遇到过stream_events返回空列表但实际流程已执行的情况，升级到 3.11 后消失。

3.2 MCP Server 选型：为什么选mcp-server-python而非自研？

MCP 协议本身是语言无关的，但实现 Server 有三种选择：

• 自己用 FastAPI 写 WebSocket Server：需手动实现register_capability、list_capabilities、tool_call的路由和鉴权，JWT 解析、心跳保活、连接池管理全是重复造轮子；
• 用官方mcp-server-python：GitHub 上由协议制定方维护，已内置 JWT 验证、Capability 注册中心、标准错误码（INVALID_TOKEN,CAPABILITY_NOT_FOUND）；
• 用第三方如mcp-server-go：但 Python 生态里 Langgraph 的StateGraph与 Go 的 gRPC 交互需要额外桥接层，增加复杂度。

我选mcp-server-python，安装命令：

poetry add "mcp-server-python>=0.3.0"

启动脚本mcp_server.py：

from mcp.server.stdio import stdio_server from mcp.server.models import Capability, Tool, ToolResult from mcp.server.session import Session import asyncio # 定义一个 Capability：查询订单 order_tool = Tool( name="query_orders", description="查询指定时间范围和前缀的订单记录", input_schema={ "type": "object", "properties": { "start": {"type": "string", "format": "date-time"}, "end": {"type": "string", "format": "date-time"}, "prefix": {"type": "string"} }, "required": ["start", "end", "prefix"] } ) async def handle_query_orders(params: dict) -> ToolResult: # 这里对接真实数据库 return ToolResult(content=f"Found 3 orders: ORD-2024-001, ORD-2024-002, ORD-2024-003") # 创建 Session session = Session() session.add_tool(order_tool, handle_query_orders) # 启动 Server if __name__ == "__main__": asyncio.run(stdio_server(session))

关键细节：input_schema必须严格遵循 JSON Schema Draft 07，format: "date-time"会被mcp-server-python自动校验，传入"2024/05/15"会直接返回VALIDATION_ERROR，不用在业务代码里写datetime.fromisoformat()。

3.3 Langgraph Agent 初始化：State 定义与节点注册的黄金法则

Langgraph 的StateGraph不是“配置”，而是“契约”。AgentState的 TypedDict 定义，决定了整个流程的数据流向。我总结出三条铁律：

1. 所有字段必须有默认值或明确可为空：sql_result: List[Dict] | None不能写成sql_result: List[Dict]，否则StateGraph初始化时会报TypeError: Field "sql_result" has no default value；
2. 避免嵌套过深：user_input: str比request: dict好，因为dict类型在 IDE 里无法补全，且state["request"]["query"]容易拼错；
3. 错误字段必须统一：所有节点都应写return {"error": "xxx"}，这样add_conditional_edges才能用同一条件判断。

完整agent.py初始化代码：

from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any, Optional import asyncio class AgentState(TypedDict): user_input: str parsed_params: Optional[Dict[str, Any]] sql_result: Optional[List[Dict]] chart_data: Optional[bytes] error: Optional[str] # 节点函数（省略具体实现，见前文） def parse_query(state: AgentState) -> AgentState: ... def execute_sql(state: AgentState) -> AgentState: ... def generate_chart(state: AgentState) -> AgentState: ... # 构建图 workflow = StateGraph(AgentState) workflow.add_node("parse", parse_query) workflow.add_node("sql", execute_sql) workflow.add_node("chart", generate_chart) # 边定义：parse 总是到 sql；sql 成功到 chart，失败到 error 处理 workflow.add_edge("parse", "sql") workflow.add_conditional_edges( "sql", lambda x: x.get("error") is not None, {True: "handle_error", False: "chart"} ) workflow.add_edge("chart", END) # 编译图 app = workflow.compile() # 启动：监听 MCP 消息，触发 Langgraph 执行 async def run_agent(): # 这里连接 MCP Server，收到 tool_call 消息后调用 app.ainvoke(...) pass

3.4 A2A 通信链路：如何让两个 Langgraph Agent 通过 MCP 互相调用？

A2A 不是“Agent A 调用 Agent B 的 API”，而是“Agent A 作为 MCP Client，向 Agent B 的 MCP Server 发送tool_call消息”。关键在于：Agent B 必须注册一个 Capability，其name与 Agent A 的tool_call.params.tool_name完全一致。

Agent B 的 Capability 注册（在mcp_server.py中）：

# Agent B 注册 capability report_tool = Tool( name="generate_report", # 注意：必须和 Agent A 调用时的 tool_name 一致 description="根据订单数据生成 PDF 报告", input_schema={...} ) async def handle_generate_report(params: dict) -> ToolResult: # 业务逻辑 return ToolResult(content="Report generated: report_20240515.pdf")

Agent A 的调用代码（在parse_query节点里）：

import websocket import json def call_mcp_tool(tool_name: str, params: dict) -> dict: ws = websocket.WebSocket() ws.connect("wss://api.xiaozhi.me/mcp/?token=eyjhbgcioijfuzi1niisinr5cci6ikpxvcj9.eyj...") # 发送 tool_call 消息 msg = { "jsonrpc": "2.0", "id": f"call-{int(time.time())}", "method": "tool_call", "params": { "tool_name": tool_name, # 必须等于 Agent B 注册的 name "arguments": params } } ws.send(json.dumps(msg)) # 接收响应 response = json.loads(ws.recv()) ws.close() return response

实操心得：token里的 JWT 必须包含scope字段，且值为["generate_report"]，否则 MCP Server 会返回FORBIDDEN。我最初漏了这步，调试了 3 小时才发现是鉴权失败，而非网络问题。

4. 核心实战环节：用 Playwright + MCP 实现浏览器自动化 Agent

4.1 为什么选 Playwright 而非 Selenium？性能与协议适配的双重优势

Selenium 的 WebDriver 协议是 HTTP-based，每个操作（click()、fill()）都要发一次 HTTP 请求，平均延迟 120ms。Playwright 基于 Chromium DevTools Protocol（CDP），用 WebSocket 直连浏览器，page.click()延迟压到 15ms 以内。更重要的是，MCP 协议天然适配 WebSocket，Playwright 的 CDP 也是 WebSocket，二者无缝衔接。

我做过对比测试：用相同脚本登录某电商后台，Selenium 平均耗时 4.2s，Playwright 仅 1.8s。而 MCP 的tool_call消息本身只有 200~300 字节，WebSocket 传输几乎无感，瓶颈完全在浏览器操作本身。

4.2 BrowserTool 的 MCP 封装：把 Playwright 操作变成标准 Capability

目标：让任意 Agent 都能通过 MCP 调用“在网页上搜索商品”这个能力。步骤：

1. 定义 BrowserTool 的 Capability Schema：

from mcp.server.models import Tool browser_tool = Tool( name="search_product", description="在指定电商网站搜索商品", input_schema={ "type": "object", "properties": { "url": {"type": "string", "description": "目标网站 URL"}, "keyword": {"type": "string", "description": "搜索关键词"}, "timeout_ms": {"type": "integer", "default": 5000} }, "required": ["url", "keyword"] } )

2. 实现 handler，用 Playwright 执行操作：

from playwright.async_api import async_playwright import asyncio async def handle_search_product(params: dict) -> ToolResult: url = params["url"] keyword = params["keyword"] timeout = params.get("timeout_ms", 5000) async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() try: await page.goto(url, timeout=timeout) # 等待搜索框出现（用 CSS 选择器，非 XPath） await page.wait_for_selector('input[name="q"]', timeout=timeout) await page.fill('input[name="q"]', keyword) await page.click('button[type="submit"]') # 等待结果加载 await page.wait_for_selector('.product-list', timeout=timeout) # 截图返回 screenshot = await page.screenshot() await browser.close() return ToolResult(content=screenshot, content_type="image/png") except Exception as e: await browser.close() return ToolResult(error=str(e))

3. 注册到 MCP Server：

session.add_tool(browser_tool, handle_search_product)

4.3 在 Langgraph Agent 中调用 BrowserTool：状态流转与错误降级

现在，我们的 Langgraph Agent 可以在execute_sql节点失败时，自动降级调用浏览器搜索：

def execute_sql(state: AgentState) -> AgentState: try: # ... 数据库查询逻辑 return {"sql_result": result} except Exception as e: # 降级：调用 BrowserTool 搜索 try: response = call_mcp_tool("search_product", { "url": "https://example-shop.com", "keyword": state["user_input"] }) if "error" in response: return {"error": f"Browser search failed: {response['error']}"} return {"chart_data": response["content"]} # 直接返回截图 except Exception as e2: return {"error": f"Both DB and browser failed: {str(e)}, {str(e2)}"}

关键技巧：call_mcp_tool必须是异步函数，否则会阻塞 Langgraph 的asyncio事件循环。我最初用同步websocket库，导致整个 Agent 卡死，改成websockets库（支持await）后解决。

4.4 真实联调记录：从 MCP 连接失败到最终截图返回的 7 步排查

这是我在mcp-a2a-demo项目里真实的调试日志，按时间顺序还原：

1. Step 1：MCP 连接 401
   错误：websocket._exceptions.WebSocketBadStatusException: 401 Unauthorized
   原因：token过期，JWT 的exp字段已过期。解决方案：用jwt.encode重新生成 token，exp设为datetime.utcnow() + timedelta(hours=24)。

2. Step 2：Capability 调用 404
   错误：MCP Server 返回{"error": {"code": -32601, "message": "Method not found"}}
   原因：Agent A 调用tool_name="search_product"，但 Agent B 注册的是name="browse_product"。修正命名一致性。

3. Step 3：Playwright 启动失败
   错误：playwright._impl._errors.Error: Failed to launch browser
   原因：Docker 容器里没装libglib2.0-0。解决方案：apt-get update && apt-get install -y libglib2.0-0。

4. Step 4：页面等待超时
   错误：TimeoutError: Timeout 5000ms exceeded.
   原因：目标网站用了反爬，input[name="q"]选择器不匹配。解决方案：换用await page.wait_for_function("document.querySelector('input[aria-label=\"Search\"]') !== null")。

5. Step 5：截图返回乱码
   错误：前端收到的 PNG 是乱码。
   原因：ToolResult(content=screenshot)的screenshot是bytes，但 MCP 默认序列化为 base64 字符串。解决方案：显式设置content_type="image/png"，Server 会自动 base64 编码。

6. Step 6：Langgraph 流程卡死
   错误：app.ainvoke()无返回。
   原因：handle_search_product函数里忘了await browser.close()，Playwright 进程泄漏。解决方案：加async with确保资源释放。

7. Step 7：最终成功
   日志：[INFO] BrowserTool returned image/png (12450 bytes)，前端显示清晰截图。

5. 常见问题与避坑指南：那些文档里绝不会写的实战细节

5.1 MCP Token 安全实践：JWT 的 scope 设计与刷新策略

MCP 的token不是“一次生成永久有效”，而是权限凭证。我见过最危险的做法：把token硬编码在前端 JS 里。正确方案：

• Scope 最小化原则：Agent A 只需调用search_product，token 的scope就只含["search_product"]，绝不给["*"]；
• Token 刷新机制：后端提供/api/mcp-token接口，用户登录后返回短期 token（2 小时），前端在onclose事件里触发刷新；
• 服务端验证：MCP Server 的verify_token函数必须检查scope是否包含请求的tool_name，代码示例：

def verify_token(token: str, required_tool: str) -> bool: try: payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"]) return required_tool in payload.get("scope", []) except jwt.ExpiredSignatureError: return False except jwt.InvalidTokenError: return False

5.2 Langgraph 性能调优：stream_events 的正确用法与内存泄漏规避

app.stream_events()是 Langgraph 的流式输出 API，但滥用会导致内存暴涨。问题现象：连续调用 100 次后，Python 进程内存占用从 150MB 涨到 1.2GB。

根因分析：stream_events默认缓存所有事件，即使你只关心on_tool_end。解决方案：

# 错误：监听所有事件 async for event in app.stream_events(input_data, version="v2"): print(event) # 正确：只订阅需要的事件类型 async for event in app.stream_events( input_data, version="v2", # 只接收 tool 调用相关的事件 filter={"event": ["on_tool_start", "on_tool_end"]} ): if event["event"] == "on_tool_end": print(f"Tool {event['name']} finished with result: {event['data']['output']}")

另外，stream_events的version="v2"参数必须显式指定，否则用旧版会丢失metadata字段。

5.3 A2A 并发瓶颈：为什么 AI Agent 不能像 Web Server 那样简单加机器？

很多人问“AI Agent 怎么扛并发”，答案不是“加 Redis 缓存”，而是理解其本质：Agent 是有状态的长期运行进程，不是无状态的 HTTP Handler。

• Langgraph 的StateGraph实例是单例，共享内存，高并发下state字典会被多个协程同时修改；
• Playwright 的browser实例不能跨协程复用，每个tool_call都要launch()新浏览器，CPU 和内存压力陡增。

我的解决方案：

1. 水平扩展 Agent 实例：用uvicorn启动多个agent.py进程，每个进程监听独立的 MCP WebSocket 端口（如wss://api.xiaozhi.me/mcp/agent1）；
2. 连接池管理：Playwright 的browser_type.launch_persistent_context()创建持久上下文，比每次launch()快 3 倍；
3. 限流熔断：在 MCP Server 层加asyncio.Semaphore(10)，限制同时执行的tool_call不超过 10 个。

5.4 调试工具链：如何快速定位 MCP-Langgraph-A2A 链路中的断点？

当整个链路失败时，按以下顺序排查：

实操心得：我写了一个debug-mcp.sh脚本，自动执行这四步并高亮失败项，把平均排查时间从 45 分钟压缩到 3 分钟。

5.5 生产部署 checklist：从 Docker 到 Kubernetes 的必填项

本地跑通不等于生产可用。我整理的上线前 checklist：

• [ ]Dockerfile 基础镜像：用python:3.11-slim-bookworm，而非latest，避免依赖漂移；
• [ ]Playwright 二进制：RUN playwright install --with-deps chromium，--with-deps安装所有系统依赖；
• [ ]MCP Token 管理：用 Kubernetes Secret 挂载，而非环境变量，防止ps aux泄露；
• [ ]Langgraph 日志：重定向logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")，方便 ELK 收集；
• [ ]健康检查端点：/healthz返回{"status": "ok", "mcp_connected": true, "playwright_ready": true}；
• [ ]资源限制：Kubernetes Pod 的resources.limits.memory设为2Gi，cpu设为2，防止 Playwright 内存溢出。

最后分享一个血泪教训：某次上线后发现 Agent 响应变慢，查日志发现是playwright install没加--with-deps，容器里缺libglib2.0-0，每次launch()都 fallback 到软件渲染，CPU 占用 98%。加了依赖后，CPU 降到 12%。

6. 项目收尾与个人体会：当 Agent 真正开始“下地干活”

这个项目跑通那天，我截了张图：前端输入框里写着“查一下小米手机销量”，回车后，不到 3 秒，一张带柱状图的 PNG 就显示出来。图里数据来自 MySQL，图表用 matplotlib 生成，整个流程经过 Langgraph 的parse→sql→chart三个节点，而sql节点背后，是 MCP 协议调用的另一个独立部署的数据库 Agent。

没有炫酷的 UI，没有大模型 logo，只有一条干净的数据流。但正是这种“看不见的管道”，才是 AI 落地的真实形态——它不取代人，而是把人从重复操作里解放出来。上周运营同事用这个功能，30 分钟生成了 12 份销售周报，而以前要花一整天。

我自己在实际使用中发现，最大的认知转变是：不要追求“一个 Agent 解决所有问题”，而要设计“一组可组合的 MCP Capability”。比如“查订单”“发邮件”“画图表”三个 Capability，可以自由组装成“日报生成 Agent”“异常预警 Agent”“客户分析 Agent”，复用率高达 80%。

最后再分享一个小技巧：在AgentState里加一个trace_id: str字段，所有节点都透传它，配合 Jaeger 做分布式追踪。当某个tool_call耗时异常，一眼就能定位是 MCP 网络延迟，还是 Playwright 渲染慢，还是数据库查询慢。这个trace_id，就是你在混沌中抓住的那根线。