一、LangGraph + MCP 实战：把业务 API 封装成可控的 Agent 工具-ZaNgVVB ‘s Blog

前言

上一篇《AI Agent 开发实践：从提示词到可控执行》讨论了 Agent 的执行循环、工具边界和安全策略。这一篇继续往工程深处走：如何把订单、工单、知识库等业务能力包装成 Agent 可以稳定调用的工具。

直接把所有 HTTP 接口塞进 Prompt，短期看起来很快，长期一定会遇到三个问题：接口描述和实现不同步、每个 Agent 重复写适配代码、权限与审计散落在业务逻辑里。

MCP（Model Context Protocol）的价值不是让模型“更聪明”，而是为模型应用和外部能力之间建立一层稳定协议。它规定工具如何被发现、参数如何描述、结果如何返回，以及客户端如何连接一个或多个服务。

本文使用 Python 3.12、LangGraph 1.2.9、langchain-mcp-adapters 0.3.0 和 MCP 1.28.1，完成一个可控的订单 Agent。示例中的订单都是虚构数据，模型地址和密钥全部通过环境变量配置。

一、MCP 解决的不是“调用”，而是“边界”

普通 Tool Calling 通常把函数定义直接绑定给某个模型：

Agent 进程
  ├─ query_order()
  ├─ create_ticket()
  └─ refund_order()

当第二个 Agent 也需要这些能力时，往往会复制一遍工具代码。接口升级后，多份定义还容易出现漂移。

MCP 把工具提供方从 Agent 进程中拆出来：

Agent 只依赖协议和工具契约，不需要知道订单能力来自 Python、Java 还是远程服务。工具服务也不关心调用者背后使用什么模型。

维度	进程内 Tool Calling	MCP
工具发现	启动时绑定函数	客户端从 Server 加载
复用范围	单个应用	多个 Agent 或客户端
传输	进程内调用	stdio 或 Streamable HTTP
权限	容易散落在业务代码	可在网关、Client 和工具层分级
演进	调用方和工具强耦合	通过契约独立演进

MCP 不是微服务协议的替代品。业务服务仍然可以使用 REST、gRPC 或消息队列，MCP Server 更像一层面向 Agent 的防腐层：把复杂接口收敛成少量、语义明确、可审计的工具。

二、项目结构

本文示例按职责拆成四层：

agent-demo/
├─ order_mcp_server.py      # 工具定义和参数校验
├─ mcp_client.py            # 连接配置与工具加载
├─ graph.py                 # LangGraph 状态图
├─ guardrails.py            # 权限、确认和结果校验
├─ settings.py              # 环境变量配置
└─ tests/
   ├─ test_tools.py
   └─ test_guardrails.py

环境变量只保存连接信息，不把密钥写入代码：

LLM_BASE_URL=https://your-openai-compatible-endpoint/v1
LLM_API_KEY=replace-me
LLM_MODEL=your-tool-calling-model
MCP_ORDER_URL=http://127.0.0.1:8000/mcp
MCP_ACCESS_TOKEN=replace-me

这里特意使用 LLM_BASE_URL 和 LLM_MODEL，而不是把模型供应商写死。只要服务兼容 OpenAI Chat Completions 和工具调用格式，就可以通过配置替换。

三、用 FastMCP 暴露业务工具

先实现两个工具：查询订单是只读操作，创建工单会产生外部影响。

from fastmcp import FastMCP
from pydantic import BaseModel, Field

mcp = FastMCP("OrderCenter")


class OrderQuery(BaseModel):
    order_id: str = Field(pattern=r"^ORD-\d{4}$")


@mcp.tool()
def query_order(order_id: str) -> dict:
    """按业务订单号查询订单，只返回 Agent 决策所需字段。"""
    request = OrderQuery(order_id=order_id)
    order = order_repository.find(request.order_id)
    if order is None:
        return {"found": False, "order_id": request.order_id}
    return {
        "found": True,
        "order_id": order.order_id,
        "status": order.status,
        "amount": str(order.amount),
    }


@mcp.tool()
def create_ticket(order_id: str, reason: str) -> dict:
    """为订单创建售后工单。调用前必须经过用户确认。"""
    if len(reason.strip()) < 4:
        raise ValueError("reason 至少包含 4 个字符")
    return ticket_service.create(order_id=order_id, reason=reason)


if __name__ == "__main__":
    mcp.run(transport="http", host="127.0.0.1", port=8000)

工具描述必须说明边界，而不是只重复函数名。“查询订单”还不够，模型需要知道输入格式、返回范围以及是否产生副作用。

工具结果也不应该返回整个数据库实体。字段越多，上下文成本越高，敏感信息泄露的风险也越大。返回 found、业务主键、状态和必要金额，通常比直接序列化 ORM 对象更稳妥。

四、通过 MCP Client 加载工具

LangChain 官方的 langchain-mcp-adapters 提供了 MultiServerMCPClient。它可以同时加载多个 MCP Server，并把工具转换为 LangChain 可调用对象。

from langchain_mcp_adapters.client import MultiServerMCPClient


async def load_tools(settings):
    client = MultiServerMCPClient({
        "order": {
            "transport": "http",
            "url": settings.mcp_order_url,
            "headers": {
                "Authorization": f"Bearer {settings.mcp_access_token}"
            },
        }
    })
    return await client.get_tools()

当前 MultiServerMCPClient 默认是无状态的：一次工具调用创建一个 ClientSession，调用结束后清理。无状态适合查询类工具；如果工具依赖长会话、进度通知或服务端上下文，需要显式管理 session，不能假设 Client 会替你保留状态。

HTTP 连接必须设置超时、认证和 Trace Header。生产环境不要把 MCP Server 直接暴露在公网，也不要把内部 Token 写进 Prompt 或工具参数。

五、构建 LangGraph 执行循环

状态图只保留完成当前任务所需的信息：消息、工具权限和确认状态。

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain.messages import AnyMessage, SystemMessage, ToolMessage
from langgraph.graph import END, START, StateGraph


class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    allowed_tools: set[str]
    confirmed_call_ids: set[str]
    step_count: int


def call_model(state: AgentState):
    response = model_with_tools.invoke([
        SystemMessage(content=SYSTEM_PROMPT),
        *state["messages"],
    ])
    return {
        "messages": [response],
        "step_count": state.get("step_count", 0) + 1,
    }


def route_after_model(state: AgentState) -> Literal["tools", END]:
    last_message = state["messages"][-1]
    if state["step_count"] >= 8:
        return END
    return "tools" if last_message.tool_calls else END


builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", guarded_tool_node)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route_after_model, ["tools", END])
builder.add_edge("tools", "model")
graph = builder.compile(checkpointer=checkpointer)

最大步数不是装饰配置。如果模型反复选择错误工具，图必须有明确终止条件，否则成本和副作用都会失控。

六、权限控制必须在模型之外

不要在 System Prompt 里写一句“创建工单前请确认”就认为安全问题解决了。Prompt 只能影响模型决策，不能替代代码权限。

READ_TOOLS = {"query_order"}
WRITE_TOOLS = {"create_ticket"}


def authorize(tool_call, state: AgentState) -> None:
    name = tool_call["name"]
    call_id = tool_call["id"]

    if name not in state["allowed_tools"]:
        raise PermissionError(f"tool not allowed: {name}")

    if name in WRITE_TOOLS and call_id not in state["confirmed_call_ids"]:
        raise ConfirmationRequired({
            "tool": name,
            "arguments": tool_call["args"],
            "reason": "该操作会创建外部工单",
        })

推荐把工具分成四级：

级别	示例	策略
读取	查询订单、搜索文档	允许自动执行，但记录审计日志
低风险写入	创建草稿、生成临时文件	可按业务规则自动执行
高风险写入	发消息、创建工单、提交表单	执行前展示参数并要求确认
不可逆操作	删除数据、支付、修改权限	默认禁止或走专门审批流程

一次受控调用的完整路径如下：

确认对象必须绑定 tool_call_id 和参数摘要。不能让用户确认“创建工单”后，模型偷偷替换订单号或原因继续复用这次确认。

七、超时、重试和幂等性

工具调用失败不代表业务操作失败。创建工单请求超时后，服务端可能已经成功落库。如果 Client 直接重试，就会产生重复工单。

正确处理方式是按错误类型区分：

async def execute_with_policy(tool, arguments, request_id):
    try:
        return await asyncio.wait_for(
            tool.ainvoke(arguments | {"request_id": request_id}),
            timeout=5,
        )
    except asyncio.TimeoutError:
        if tool.name in READ_TOOLS:
            return await retry_once(tool, arguments)
        existing = await query_operation_result(request_id)
        if existing:
            return existing
        raise UnknownWriteResult("写操作结果未知，需要人工确认")

只读操作可对临时网络错误做有限重试。写操作必须携带幂等键，并在重试前查询执行结果。权限不足、参数错误和用户拒绝都不应该自动重试。

八、结果校验

工具返回成功并不等于任务完成。Agent 至少要校验：

返回结构是否符合契约。
业务主键是否与请求一致。
写操作是否能通过查询接口确认。
返回内容是否包含不应进入模型上下文的敏感字段。
最终回答是否准确引用工具结果，而不是重新编造。

下面是本文本地测试的实际输出。五个场景分别覆盖正常读取、错误参数、未确认写入、确认后写入和未知工具。

测试不是为了证明模型每次都选对工具，而是证明即使模型选错，系统也能把行为限制在可控边界内。

九、常见踩坑

9.1 工具粒度过大

execute_business_action(action, payload) 看起来复用性很强，实际会把所有权限、校验和审计集中到一个万能入口。更可靠的方式是让每个工具表达一个清晰业务动作。

9.2 把数据库查询直接开放给模型

万能 SQL 工具很难做行列级权限，也容易读取无关数据。优先提供领域工具，例如 query_order(order_id)，而不是 execute_sql(sql)。

9.3 工具异常只返回字符串

"调用失败" 无法指导下一步。建议统一返回错误代码、是否可重试、是否需要用户介入和安全摘要。

{
  "ok": false,
  "error_code": "CONFIRMATION_REQUIRED",
  "retryable": false,
  "message": "创建工单前需要用户确认"
}

9.4 日志记录完整请求体

调试阶段把 Header、Token 和整个工具结果写进日志，上线后很容易变成数据泄露。日志应使用字段白名单，并对 Token、身份证号、手机号等内容脱敏。

9.5 忽略协议版本和工具契约变更

MCP Server 和 Client 应分别发布版本，并为工具输入输出保留契约测试。删除字段、修改枚举含义和改变默认行为，都可能让旧 Agent 在无编译错误的情况下产生错误决策。

十、生产检查清单

MCP Server 只暴露必要工具，不暴露通用 Shell、SQL 或文件写入能力。
HTTP 连接启用 TLS、认证、请求大小限制和服务端超时。
高风险工具由代码触发确认，确认绑定调用 ID 和参数摘要。
写操作携带幂等键，超时后先查结果再决定是否重试。
工具输入使用结构化模型校验，输出使用稳定 Schema。
日志记录调用名称、耗时、结果摘要和错误码，不记录密钥。
Agent 设置最大步骤数、最大工具调用数和成本预算。
对错误参数、权限不足、超时、重复提交和用户拒绝建立自动化测试。

十一、总结

MCP 的核心价值不是多了一种调用方式，而是把 Agent 与业务能力之间的契约独立出来。LangGraph 负责状态和执行循环，MCP 负责工具发现与传输，业务服务负责真正的领域规则，权限和确认代码负责守住边界。

一个可上线的工具系统需要同时具备：明确的工具职责、结构化参数、最小结果集、分级权限、幂等写入、错误分类、结果校验和全链路审计。

下一篇继续解决 Agent 的“知识从哪里来”： LangGraph + PGVector 构建 RAG 知识库：从文档切分到答案评估。

前言