《MCP + LangChain：构建可治理的 AI Agent 工作流——从工具调用到任务编排的工程实践》

沈宥

发布于 2026-01-08 11:09:40

5500

文章被收录于专栏：从头开始学习测试开发从头开始学习测试开发

一、引言：当 MCP 遇上 LangChain，能力才真正“活”起来

在上一篇中，我们构建了 单个安全、可审计的 MCP Tool。但现实场景中，AI 往往需要 多个工具协同完成复杂任务，例如：

“帮我查一下用户 A 最近一笔未支付订单，并生成催付话术。”

这涉及：

查询用户 ID（可能需模糊匹配）
获取该用户最近订单
判断是否未支付
调用话术模板引擎生成文案

单个 Tool 无法完成，必须引入 Agent 编排框架。而 LangChain 是当前最成熟的方案之一。

但问题来了：LangChain 默认的 Tool 调用是“裸奔”的——无权限、无审计、无限流。如何让它与我们精心设计的 MCP 工具链无缝集成？

本文将手把手教你：用 LangChain 调度 MCP Tool，同时保留完整的治理能力。

二、核心思想：MCP 是“能力提供方”，LangChain 是“调度指挥官”

角色	职责	技术栈
MCP Server	提供标准化、安全、带鉴权的 Tool 接口	FastAPI + Pydantic
LangChain Agent	理解用户意图，规划 Tool 调用顺序	LangChain + LLM
测开工程师	确保整个链路可监控、可回溯、可拦截	日志 + 告警 + Mock

✅ 关键原则：LangChain 永远不直接访问数据库或内部服务，所有操作必须通过 MCP Tool。

三、实战：将 MCP Tool 注册为 LangChain Tool

3.1 定义 MCP Client（封装 HTTP 调用）

1# mcp_client.py
2import httpx
3from pydantic import BaseModel
4
5class MCPClient:
6    def __init__(self, base_url: str, token: str):
7        self.base_url = base_url
8        self.headers = {"Authorization": f"Bearer {token}"}
9    
10    async def call_tool(self, tool_name: str, input_data: dict) -> dict:
11        async with httpx.AsyncClient() as client:
12            resp = await client.post(
13                f"{self.base_url}/tools/{tool_name}",
14                json=input_data,
15                headers=self.headers,
16                timeout=10.0
17            )
18            resp.raise_for_status()
19            return resp.json()

3.2 将 MCP Tool 包装为 LangChain Tool

1# langchain_tools.py
2from langchain.tools import BaseTool
3from pydantic import BaseModel
4from mcp_client import MCPClient
5
6class QueryOrderStatusInput(BaseModel):
7    order_id: str
8
9class QueryOrderStatusTool(BaseTool):
10    name = "query_order_status"
11    description = "根据订单ID查询订单状态，仅返回状态和支付信息"
12    args_schema = QueryOrderStatusInput
13    
14    def __init__(self, mcp_client: MCPClient):
15        super().__init__()
16        self.mcp_client = mcp_client
17
18    def _run(self, order_id: str) -> str:
19        # 同步调用（生产建议用 async）
20        result = self.mcp_client.call_tool("query_order_status", {"order_id": order_id})
21        return f"订单 {order_id} 状态: {result['status']}, 支付状态: {result['pay_status']}"
22
23    async def _arun(self, order_id: str) -> str:
24        result = await self.mcp_client.call_tool("query_order_status", {"order_id": order_id})
25        return f"订单 {order_id} 状态: {result['status']}, 支付状态: {result['pay_status']}"

💡 测开注意：这里 args_schema 强制使用 Pydantic，确保输入合法。

四、构建可治理的 Agent：注入用户上下文与审计

LangChain Agent 默认是“无状态”的，但我们必须让它知道 当前用户是谁。

4.1 自定义 Agent Executor（注入 user_id）

1from langchain.agents import initialize_agent, AgentType
2from langchain.chat_models import ChatOpenAI
3
4def create_governed_agent(user_id: str, access_token: str):
5    mcp_client = MCPClient("https://mcp.yourcompany.com", token=access_token)
6    
7    tools = [
8        QueryOrderStatusTool(mcp_client),
9        GenerateReminderMessageTool(mcp_client),  # 另一个 MCP Tool
10    ]
11    
12    llm = ChatOpenAI(model="gpt-4o")
13    
14    agent = initialize_agent(
15        tools,
16        llm,
17        agent=AgentType.OPENAI_FUNCTIONS,
18        verbose=True,
19        handle_parsing_errors=True
20    )
21    
22    # 关键：在调用时，MCP Client 会携带 token，后端可解析出 user_id
23    return agent

🔒 安全闭环：前端传入 access_token → MCP Client 携带 → MCP Server 验证 → 执行权限校验。

五、审计增强：记录完整 Agent 思考链

LangChain 的 verbose=True 会打印 Thought/Action/Observation，但这不够结构化。

5.1 自定义 Callback 记录全流程

1from langchain.callbacks.base import BaseCallbackHandler
2
3class AuditCallbackHandler(BaseCallbackHandler):
4    def on_agent_action(self, action, **kwargs):
5        logger.info("agent_step",
6                    tool=action.tool,
7                    tool_input=action.tool_input,
8                    log=action.log)
9
10    def on_agent_finish(self, finish, **kwargs):
11        logger.info("agent_complete", output=finish.return_values["output"])
12
13# 使用
14agent = create_governed_agent(...)
15response = agent.run("查订单 ORD123456", callbacks=[AuditCallbackHandler()])

📊 日志示例：

1agent_step | tool=query_order_status | tool_input={"order_id":"ORD123456"}
2agent_step | tool=generate_reminder | tool_input={"status":"unpaid"}
3agent_complete | output="您好，您的订单尚未支付..."

六、异常兜底：防止 Agent 无限循环或越权

6.1 设置最大步数（max_iterations）

1agent = initialize_agent(
2    ...,
3    max_iterations=5,  # 防止死循环
4    early_stopping_method="generate"  # 超步数后让模型总结
5)

6.2 敏感操作二次确认（人工介入）

对于高风险操作（如退款），可在 MCP Tool 中返回“需审批”：

1# MCP Tool 返回
2{"status": "pending_approval", "ticket_id": "TICKET_789"}
3
4# LangChain 自定义 Output Parser 检测此状态
5if "pending_approval" in output:
6    raise HumanInterventionRequired("需要人工审批")

🧪 测开价值：你的自动化测试可以模拟“审批通过”或“拒绝”，验证整个流程。

七、多环境与 Mock：测开的主场

7.1 测试环境自动 Mock MCP 响应

1# conftest.py (pytest)
2@pytest.fixture
3def mock_mcp_client():
4    with patch("mcp_client.MCPClient.call_tool") as mock_call:
5        mock_call.side_effect = lambda tool, inp: {
6            "query_order_status": {"status": "unpaid", "pay_status": "failed"},
7            "generate_reminder": {"message": "请尽快支付..."}
8        }[tool]
9        yield

7.2 E2E 测试 Agent 行为

1def test_agent_generates_reminder_for_unpaid_order(mock_mcp_client):
2    agent = create_governed_agent("test_user", "fake_token")
3    response = agent.run("帮我处理未支付订单 ORD123")
4    assert "请尽快支付" in response
5    assert "TICKET_" not in response  # 确保未触发审批

✅ 优势：无需启动真实 MCP Server，快速验证 Agent 逻辑。

八、性能与限流：别让一个用户拖垮系统

在 MCP Server 层做 Token 级限流（如 100 QPS / user）
在 LangChain 层做 Agent 并发控制（如 asyncio.Semaphore）
对慢 Tool（如生成报告）设置 独立超时

九、总结：测开如何主导 AI Agent 治理？

能力	实现方式
权限控制	MCP Server 校验 Token + user_id
数据安全	Pydantic 输出白名单
行为审计	Callback + Structlog
异常兜底	Max steps + 人工审批
自动化测试	Mock MCP Client + E2E 验证