12 框架与协议

第一部分 速记页

第二部分 易混对照

第三部分 面试口述稿

3.1 “你为什么没用框架，而是自己写的循环？”

一开始自己写，是因为这个任务是单 Agent、线性的：调模型、执行工具、把结果放回去、重复，加上步数上限和卡住检测，六十多行就写完了。每一步做了什么一眼能看清，出问题直接打断点，不用去读框架源码。

后来我用 LangGraph 把同一个循环重写了一遍，为了对比公平，两个版本共用工具、提示词、重复调用检测和验收，只换循环的写法。同样的配置各跑三轮，手写 30 次全过，LangGraph 29 次全过，另外有一次是 API 余额不足崩了，不计入。步数和 token 也看不出差别，这符合预期，因为模型和工具都一样。

差别在工程上：LangGraph 版代码多了一倍，一百二十多行，控制流分散在几个路由函数里；但它每一步自动存检查点，进程被杀之后可以用 thread_id 接着跑。我实测过在第 12 步杀掉进程，续跑从第 13 步开始，已经执行过的工具调用没有重复。

所以我的结论是：简单线性的 Agent 手写更直接；需要中断恢复、人工审批、复杂分支的时候，LangGraph 是有实际价值的。

3.2 “LangGraph 的核心概念是什么？”

它把 Agent 写成一张有状态的图。三个要素：状态、节点、边。

状态是一个类型化的字典，定义了图里流转的所有数据，比如消息列表、当前步数。每个字段可以指定 reducer，决定更新怎么合并，最常用的 add_messages 是把新消息追加进列表，而不是覆盖。

节点是普通函数，接收当前状态，返回要更新的字段。

边决定下一步去哪。普通边是固定的，条件边由一个函数根据状态返回下一个节点的名字。一个典型的 Agent 就是两个节点：调模型的节点和执行工具的节点，模型节点后面接条件边，有工具调用就去工具节点，没有就结束，工具节点再连回模型节点。

编译时传入检查点保存器，每个超级步结束自动保存状态，同一个 thread_id 可以恢复。输出方面，stream 支持几种模式，updates 看每个节点改了什么，messages 拿逐 token 的模型输出，custom 可以在节点里用 get_stream_writer 发自定义事件，我项目推给前端的进度就是用这个。

3.3 “MCP 是什么？架构是怎样的？”

MCP 是 AI 应用和工具服务之间的开放协议，基于 JSON-RPC。它解决的问题是工具复用：一个工具写成 MCP server，任何支持 MCP 的应用都能接，不用每个应用各写一遍。

架构上三个角色：host 是 AI 应用本身，负责管理连接、权限、调用模型；client 由 host 创建，每个 client 只连一个 server；server 提供具体能力。规范特意要求 server 看不到完整对话，也看不到其他 server，对话历史留在 host 里。

server 能提供三种东西：tools 是模型决定调用的函数；resources 是数据，由应用决定要不要放进上下文；prompts 是提示词模板，由用户选择使用，比如斜杠命令。

传输有两种，stdio 是客户端启动子进程、通过标准输入输出通信，适合本地；Streamable HTTP 是每条消息一个 POST 请求，适合远程服务。

我项目写了一个 MCP server，直接复用 Agent 的工具函数和边界检查，所以任何支持 MCP 的客户端接上它，拿到的都是和我自己 Agent 一样的工具和限制。测试里用 SDK 的客户端连上去，验证了越界访问会返回 isError 结果。

3.4 “MCP 和 A2A 有什么区别？”

MCP 解决的是一个 Agent 怎么接工具和数据；A2A 解决的是 Agent 和 Agent 之间怎么协作。A2A 官方文档自己的说法也是两者互补：用 MCP 给单个 Agent 配工具，用 A2A 让这个 Agent 和其他 Agent 协作。

区别在于对方是不是”不透明的”。MCP 的 server 暴露的是具体函数，调用方知道参数和返回；A2A 里对方是一个完整的 Agent，你只知道它的 Agent Card 上写的能力，发一个任务过去，它内部用什么模型、什么工具你不知道，它可能要执行很久，中间还可能回来问你要补充信息。所以 A2A 的核心对象是 Task，有提交、执行中、需要输入、完成、失败这些状态，支持流式更新和推送通知。

A2A 是 Google 发起的，后来捐给了 Linux 基金会。我项目是单 Agent，没用到。

3.5 “Dify、Coze 这种平台和 LangGraph 怎么选？”

看谁来做、要做多深。

Dify 和 Coze 是可视化平台，拖拽搭工作流、接知识库、配工具，自带对话界面和 API，业务人员也能上手，适合流程比较固定、要快速上线的场景，比如内部知识库问答、简单的客服流程。两者都开源、能自己部署。

局限是遇到平台没提供的能力就很难扩展，复杂的分支、自定义的状态管理、细粒度的评测和调试都受平台限制，版本管理和代码审查也不如代码方便。

LangGraph 这类代码框架灵活，能做任意复杂的控制流，能接进自己的测试、CI、监控，但需要工程师来写和维护。

实际团队里两者经常并存：用平台快速验证想法、做简单场景，验证有价值、复杂度上来之后再用代码重写核心流程。

3.6 “你怎么看框架选型？”

我会问几个问题。

第一，任务结构。单 Agent 线性循环，手写就够；有复杂分支、并行、多 Agent、需要人工审批和中断恢复，框架价值大。

第二，要不要持久化。长任务、要能从中间恢复，自己写检查点要处理的细节很多，这是 LangGraph 最实在的价值。

第三，团队和生态。团队已经统一用某个框架，接入观测平台、评测工具都现成，就跟着用。

第四，锁定成本。模型接口、工具定义尽量保持框架无关，比如工具按 MCP 暴露、观测按 OpenTelemetry 埋点，以后换框架代价小。我项目的工具层就是独立的，手写循环、LangGraph、MCP server 三个入口共用同一套工具函数。

还有一点是别被框架的默认值骗了。我用 LangGraph 时发现很多教程说递归上限默认 25，实际 1.2 版是 10007，框架版本变化快，关键参数要自己实测，并且显式设置。

第四部分 逐个详解

4.1 框架到底替你做了什么

回忆 05 篇手写的 Agent 循环，一个能用的 Agent 至少要处理这些事：

flowchart LR
    subgraph 必须有
        A[调模型<br/>各家接口不同] --> B[解析工具调用<br/>参数 JSON 可能不合法]
        B --> C[执行工具<br/>异常转成结果]
        C --> D[结果放回消息]
        D --> E{停止条件}
        E -->|继续| A
    end
    subgraph 迟早要有
        F[流式输出]
        G[状态持久化 / 续跑]
        H[人工审批 暂停恢复]
        I[链路追踪]
        J[多 Agent 调度]
    end

左边的核心循环，手写几十行就行；右边这些，每一样自己做都要花不少功夫，而且容易出错（07 篇讲过检查点和幂等有多少细节）。框架的价值主要在右边。

代价：

4.2 框架和平台地图

flowchart TD
    subgraph 可视化平台
        DIFY[Dify]
        COZE[Coze]
    end
    subgraph Agent 框架
        LC[LangChain<br/>create_agent + 中间件]
        OAI[OpenAI Agents SDK]
        CAS[Claude Agent SDK]
        LI[LlamaIndex<br/>偏数据和检索]
    end
    subgraph 编排引擎
        LG[LangGraph<br/>状态图 检查点]
    end
    subgraph 协议
        MCP[MCP<br/>Agent ↔ 工具]
        A2A[A2A<br/>Agent ↔ Agent]
    end
    LC -->|构建在其上| LG
    LC -.-> MCP
    OAI -.-> MCP
    CAS -.-> MCP
    LG -.-> MCP

下面每个只讲清”是什么、适合什么”，内容来自各自的官方文档（2026-09 查阅）。

LangChain 是怎么来的，在 07 篇 4.5 节讲过：LangChain 是 Harrison Chase 2022 年 10 月开源的，最早用来把”提示词模板 → 模型 → 解析输出”串成一条链；补充一点，他当时在机器学习公司 Robust Intelligence 工作，一个多月后 ChatGPT 发布，大量开发者涌进来，LangChain 成了当时用得最多的框架之一，2023 年 4 月成立了公司（LangChain 的历史）。

LangChain

• v1 的核心是 create_agent：把模型、工具、提示词、**中间件（middleware）**组合成一个 Agent。中间件用来逐步加功能，官方举的例子有护栏、重试、路由、自定义工具策略
• 官方文档明确说 LangChain 的 Agent 构建在 LangGraph 之上，因此自带持久化、人工介入这些能力
• 官方给的选择建议：想要开箱即用、带上下文压缩和子 Agent 的，用 Deep Agents；想要高度可定制的，用 LangChain；需要把确定性流程和 Agent 行为组合在一起的复杂场景，用 LangGraph
• 岗位里出现最多（00 篇统计 54 个岗位里 16 次），面试时至少要能说清它和 LangGraph 的关系

LangGraph：见 4.3 节，我项目用过。

LlamaIndex

• 官方定位是面向自有数据构建 Agent 和工作流的数据框架，核心是”上下文增强”：让模型能用上私有数据
• 组件：数据连接器（接 API、数据库、PDF）、索引、查询引擎和对话引擎、Agent、事件驱动的工作流
• 配套的云服务有文档解析 LlamaParse 等
• 适合 RAG 为主、数据源多、文档解析复杂的项目

OpenAI Agents SDK

• 轻量的多 Agent 框架，几个核心概念：Agent（带指令和工具的模型）、交接（handoff）（把任务委派给另一个 Agent）、护栏（guardrails）（校验输入输出）、会话（sessions）（跨轮保存上下文）、追踪（tracing）
• 默认用 OpenAI 模型，也可以配置其他厂商
• 适合需要几个角色分工、互相转交的场景，13 篇多 Agent 会用到”交接”这个概念

Claude Agent SDK

• 把 Claude Code 背后的 Agent 循环、内置工具（读写编辑文件、执行命令、搜索网页）、上下文管理做成 Python 和 TypeScript 库
• 还提供钩子（在循环的关键节点运行自定义代码）、子 Agent、MCP、权限控制（哪些工具自动执行、哪些要批准）、会话（可以恢复或分叉）
• 和前面几个的区别：它自带一整套”能干活”的工具和循环，适合要做编码、文件处理类 Agent，不想自己实现工具循环的场景

Dify

• 开源的大模型应用开发平台，许可证是在 Apache 2.0 基础上加了附加条件的 Dify 开源许可证
• 可视化工作流、支持大量商业和开源模型、RAG 流水线、带内置工具的 Agent、监控和 API
• 可以用 Docker 自己部署

Coze（扣子）

• 字节跳动的 Agent 开发平台，开源版叫 Coze Studio，Apache 2.0 许可证，后端用 Go
• 可视化工作流、插件、知识库、Agent 搭建和调试

平台和框架的取舍：

4.3 LangGraph 核心概念

LangGraph 为什么出现（LangChain 早期的 AgentExecutor 把循环包死了，改不动、存不了档，2024 年 1 月团队发布了允许有环的图），在 07 篇 4.5 节讲过，这里不重复。

三要素：

TypedDict：Python typing 模块里的一种类型写法，声明”这个字典有哪些键、每个键是什么类型”。它运行时就是普通字典，类型信息给编辑器和框架看。

reducer（合并函数）：节点只返回”要改哪些字段”，框架把它合并进状态。默认是覆盖；想要别的合并方式，用 Annotated[类型, 合并函数] 指定：

class State(TypedDict):
    messages: Annotated[list, add_messages]
    step: int

• step 没有指定，节点返回 {"step": 3} 就直接覆盖
• messages 指定了 add_messages，节点返回 {"messages": [新消息]} 会追加到已有列表后面，而不是把整个列表换掉。add_messages 还会按消息 ID 去重，ID 相同的新消息会替换旧的
• Annotated：typing 里的写法，给类型附加额外信息。Annotated[list, add_messages] 意思是”类型是 list，附带信息是 add_messages”，LangGraph 读到这个附带信息就知道怎么合并

我项目的图：

flowchart TD
    START([START]) --> agent
    agent -->|after_agent：模型没调工具| verify
    agent -->|after_agent：有工具调用| tools
    tools -->|after_tools：卡住或到步数上限| verify
    tools -->|after_tools：否则| agent
    verify --> END([END])

对应的代码（agent/graph_agent.py）：

def after_agent(state: State):
    return "verify" if state["status"] == "agent_done" else "tools"

def after_tools(state: State):
    return "verify" if state["status"] == "stuck" or state["step"] >= core.MAX_STEPS else "agent"

g = StateGraph(State)
g.add_node("agent", agent)
g.add_node("tools", tools)
g.add_node("verify", verify)
g.add_edge(START, "agent")
g.add_conditional_edges("agent", after_agent, ["tools", "verify"])
g.add_conditional_edges("tools", after_tools, ["agent", "verify"])
g.add_edge("verify", END)
return g.compile(checkpointer=checkpointer)

• add_conditional_edges 的第三个参数列出这个路由函数可能返回的节点名，框架用它画图、检查
• compile(checkpointer=...) 编译成可执行的图，传入检查点保存器（07 篇讲过）

和手写版的对照：

流式模式（官方文档）：

stream_mode 可以传一个列表同时要多种。文档里还介绍了 version="v2" 的统一输出格式，每块是一个带 type 字段的字典；下面的实验用的是默认格式，同时要多种模式时每块是 (模式, 数据) 元组。

4.4 动手实验 1：一个最小的 LangGraph Agent

用 LangGraph 自带的 ToolNode 和 tools_condition，配一个写死剧本的假模型，看图怎么跑、流式输出长什么样。保存为 lg_min.py，用项目虚拟环境运行。

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.config import get_stream_writer
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition


@tool
def run_command(command: str) -> str:
    """执行一条构建命令"""
    if "build" in command and "-B" not in command:
        return "[exit=0] 编译完成"
    return "[exit=1] 找不到 zlib" if "ZLIB" not in command else "[exit=0] 配置完成"


SCRIPT = [
    AIMessage("", tool_calls=[{"name": "run_command", "args": {"command": "cmake -B build"}, "id": "c1"}]),
    AIMessage("", tool_calls=[{"name": "run_command", "args": {"command": "cmake -B build -DZLIB_ROOT=deps"}, "id": "c2"}]),
    AIMessage("", tool_calls=[{"name": "run_command", "args": {"command": "cmake --build build"}, "id": "c3"}]),
    AIMessage("编译成功"),
]


class State(TypedDict):
    messages: Annotated[list, add_messages]
    step: int


def agent(state: State):
    step = state["step"] + 1
    msg = SCRIPT[step - 1]
    get_stream_writer()(f"第 {step} 步，模型回复 {len(msg.tool_calls)} 个工具调用")
    return {"messages": [msg], "step": step}


g = StateGraph(State)
g.add_node("agent", agent)
g.add_node("tools", ToolNode([run_command]))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition)
g.add_edge("tools", "agent")
graph = g.compile()

inputs = {"messages": [HumanMessage("编译这个项目")], "step": 0}
for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    if mode == "custom":
        print("custom ", chunk)
    else:
        for node, update in chunk.items():
            last = update["messages"][-1]
            print("updates", node, "->", last.type, (last.content or last.tool_calls[0]["args"]["command"])[:30])

final = graph.invoke(inputs)
print("消息数", len(final["messages"]), "步数", final["step"])
print(graph.get_graph().draw_mermaid())

实际输出（Python 3.13，langgraph 1.2.11）：

custom  第 1 步，模型回复 1 个工具调用
updates agent -> ai cmake -B build
updates tools -> tool [exit=1] 找不到 zlib
custom  第 2 步，模型回复 1 个工具调用
updates agent -> ai cmake -B build -DZLIB_ROOT=dep
updates tools -> tool [exit=0] 配置完成
custom  第 3 步，模型回复 1 个工具调用
updates agent -> ai cmake --build build
updates tools -> tool [exit=0] 编译完成
custom  第 4 步，模型回复 0 个工具调用
updates agent -> ai 编译成功
消息数 8 步数 4
---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	agent(agent)
	tools(tools)
	__end__([<p>__end__</p>]):::last
	__start__ --> agent;
	agent -.-> __end__;
	agent -.-> tools;
	tools --> agent;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

逐段讲。

定义工具。

• @tool：LangChain 的装饰器，把普通函数变成模型能调用的工具。它读取函数名当工具名、文档字符串当描述、类型注解生成参数 Schema
• 文档字符串（docstring）：函数定义下面第一行用三引号写的字符串，这里是 """执行一条构建命令"""。用 @tool 时必须写，否则工具没有描述会报错
• 函数体是假的：根据命令文字返回写死的结果，模拟”第一次配置找不到 zlib，加上 ZLIB_ROOT 后成功”

假模型的剧本。

• AIMessage(内容, tool_calls=[...])：LangChain 的”模型回复”消息类型。tool_calls 里每一项是一个字典，有 name、args、id，这是 LangChain 统一的格式，不管底层是哪家模型
• 前三条各要求调一次工具，第四条没有工具调用，表示结束

agent 节点。

• 参数 state 是当前状态，返回的字典是更新。返回 {"messages": [msg]}，由于 add_messages，消息被追加
• get_stream_writer()：拿到一个”写入函数”，调用它写的内容会出现在 custom 模式的流里。它只能在图运行时的节点里调用

建图。

• ToolNode([run_command])：LangGraph 预置的工具节点。它读最后一条消息的 tool_calls，执行对应工具，把每个结果包装成 ToolMessage（带着对应的 tool_call_id）返回
• tools_condition：预置的路由函数。最后一条消息有工具调用就返回 "tools"，没有就返回 END。所以这里 add_conditional_edges 不用写第三个参数
• add_edge("tools", "agent")：工具执行完回到模型，形成循环

运行。

• graph.stream(inputs, stream_mode=["updates", "custom"])：返回一个生成器，一边运行一边产出数据。同时要两种模式时，每次产出 (模式名, 数据) 元组，用 for mode, chunk in ... 拆开
• updates 模式的数据是 {节点名: 这个节点返回的更新}，所以要 chunk.items() 再拆一层
• last.type：消息类型，ai 是模型消息，tool 是工具结果
• last.content or last.tool_calls[0]["args"]["command"]：or 在左边为空字符串时取右边。模型消息有工具调用时内容是空的，就显示命令
• graph.invoke(inputs)：不要流式，直接跑完返回最终状态。这里重新跑了一遍，所以剧本又从头开始
• graph.get_graph().draw_mermaid()：导出图的 mermaid 代码，虚线表示条件边

看输出。

1. custom 事件总是出现在同一节点的 updates 之前，因为它是节点运行过程中写出的，而 updates 是节点返回之后才有的。我项目用 custom 推送工具开始执行这类进度，就是为了让前端在工具跑完之前就能看到
2. 消息数 8 = 1 条用户消息 + 4 条模型消息 + 3 条工具结果
3. 导出的 mermaid 图和 4.3 节画的结构一致：从 agent 出发有两条虚线（条件边）

自己改一改：

1. 把 stream_mode 改成 "values"（单个字符串，不是列表），看每次输出的内容变成什么
2. 把剧本第一条改成一次要两个工具调用（tool_calls 列表里放两个），看 ToolNode 怎么处理
3. 编译时加上 checkpointer=InMemorySaver()（从 langgraph.checkpoint.memory 导入），调用时传 config={"configurable": {"thread_id": "t1"}}，跑完后用 graph.get_state(config).values["step"] 查状态

4.5 MCP 架构

04 篇讲了 MCP 在工具调用里的位置、和 Function Calling 的分工。这里讲架构和协议本身，内容来自 MCP 规范 2026-07-28 版。

三个角色：

flowchart LR
    subgraph HOST["Host 进程（AI 应用）"]
        H[Host<br/>管权限 调模型<br/>持有完整对话]
        C1[Client 1]
        C2[Client 2]
        C3[Client 3]
        H --> C1
        H --> C2
        H --> C3
    end
    C1 --> S1[Server：文件和 Git<br/>本地]
    C2 --> S2[Server：数据库<br/>本地]
    C3 --> S3[Server：外部 API<br/>远程]

规范列出的设计原则里，有一条和安全直接相关：server 不能读到完整对话，也不能”看到”其他 server。server 只拿到必要的上下文，完整对话留在 host 里，跨 server 的交互由 host 控制。这意味着一个恶意或被攻破的 MCP server，接触不到别的 server 的数据，也读不到用户和模型的全部聊天。

三种原语，按”谁来控制”区分：

这张表是理解 MCP 的关键：不是所有东西都做成工具。只读的背景资料更适合做成 resource，由应用决定什么时候放进上下文，不用让模型每次想起来去调用；常用的任务模板做成 prompt，给用户点选。

2026-07-28 版的重要变化：协议无状态。 规范原文的说法是 MCP 是一个无状态协议，每个请求都是自包含的，自带协议版本和能力。和早期版本对比：

为什么这很重要？ 无状态意味着远程 MCP server 可以像普通 Web 服务一样放在负载均衡后面，任意一台机器都能处理任意请求，不用把同一个客户端粘在同一台机器上。网上大量资料讲的还是 initialize 握手那一套，面试时如果说到这一点，要说清楚是哪个版本。规范也写了和旧版本互相兼容的回退办法。

两种传输方式（规范传输一节）：

规范强调：协议语义在所有传输上完全一样，传输只管消息怎么打包和送达。所以同一个 server 代码，换个启动参数就能从 stdio 变成 HTTP 服务。

tools 的错误分两类（04 篇讲过，这里再强调一次）：工具不存在、请求格式错误走 JSON-RPC 协议错误；参数校验失败、业务失败走带 isError: true 的正常结果，交给模型自己修正。

4.6 动手实验 2：一个 MCP server 的三种原语

用项目虚拟环境里的 mcp SDK（2.2.0），在同一个进程里起一个 server、用客户端连上，把 tools、resources、prompts 都走一遍。保存为 mcp_min.py。

import asyncio

from mcp import Client
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

server = MCPServer("demo-build", instructions="演示用的构建工具服务")


@server.tool(description="读取项目里的文件")
def read_file(path: str) -> str:
    if path.startswith("/") or ".." in path:
        raise ToolError(f"路径越界: {path}")
    return f"{path} 的内容"


@server.resource("build://targets", description="支持的目标平台")
def targets() -> str:
    return "aarch64-linux-musl\nx86_64-linux-musl"


@server.prompt(description="交叉编译任务模板")
def cross_build(project: str, target: str = "aarch64-linux-musl") -> str:
    return f"把 {project} 交叉编译到 {target}，完成后说明产物位置。"


async def main():
    async with Client(server) as c:
        print("协议版本", c.protocol_version, "服务", c.server_info.name)
        print("能力", sorted(k for k, v in c.server_capabilities.model_dump().items() if v))
        for t in (await c.list_tools()).tools:
            print("工具", t.name, t.input_schema)
        ok = await c.call_tool("read_file", {"path": "CMakeLists.txt"})
        bad = await c.call_tool("read_file", {"path": "../secret"})
        print("调用", ok.is_error, ok.content[0].text, "|", bad.is_error, bad.content[0].text)
        print("资源", [str(r.uri) for r in (await c.list_resources()).resources])
        res = await c.read_resource("build://targets")
        print("读资源", res.contents[0].text.splitlines())
        p = await c.get_prompt("cross_build", {"project": "zlib"})
        print("提示词", p.messages[0].role, p.messages[0].content.text)


asyncio.run(main())

实际输出（Python 3.13，mcp 2.2.0）：

协议版本 2026-07-28 服务 demo-build
能力 ['prompts', 'resources', 'tools']
工具 read_file {'type': 'object', 'properties': {'path': {'title': 'Path', 'type': 'string'}}, 'required': ['path'], 'title': 'read_fileArguments'}
调用 False CMakeLists.txt 的内容 | True Error executing tool read_file: 路径越界: ../secret
资源 ['build://targets']
读资源 ['aarch64-linux-musl', 'x86_64-linux-musl']
提示词 user 把 zlib 交叉编译到 aarch64-linux-musl，完成后说明产物位置。

另外 stderr 上还有一行 server 打的 INFO 日志：Tool 'read_file' failed: ... 路径越界: ../secret。

这里的路径检查是演示用的简化写法，真实项目要按 11 篇的 resolve + is_relative_to 来写。

逐段讲。

server 端。

• MCPServer("demo-build", instructions=...)：创建 server，instructions 是给客户端（和模型）看的使用说明
• @server.tool(description=...)：注册工具。SDK 读函数的类型注解生成 inputSchema，所以输出里 path 是 string 且 required
• raise ToolError(...)：SDK 自己的异常类。我看了 SDK 源码里这个类的说明：抛它表示”预料之中的失败”，调用返回 is_error=True，消息原样给模型读；抛其他任何异常都会被当成崩溃，模型只能看到 Error executing tool <工具名>，看不到具体原因。所以项目 mcp_server.py 里的 guarded 要把 tools.ToolError 转成 SDK 的 ToolError，否则模型拿不到”路径越界”这种能指导下一步的信息
• @server.resource("build://targets", ...)：注册资源。第一个参数是资源的 URI（统一资源标识符，一个唯一的地址字符串），build:// 这种写法是自定义的，只要唯一就行
• @server.prompt(...)：注册提示词模板，函数参数就是模板参数，返回的字符串成为一条用户消息

client 端。

• async def main()：异步函数。里面可以用 await 等待耗时的操作（网络、子进程通信）而不阻塞整个程序。异步函数不能直接调用，要交给 asyncio.run(main()) 运行。14 篇会细讲异步
• async with Client(server) as c:：异步上下文管理器，和 with 一样保证连接用完关闭，只是进入和退出时要 await。这里把 server 对象直接传给 Client，SDK 在同一进程里用内存通道连接，不启动子进程，适合测试。项目 tests/test_mcp.py 里两种都测了：内存连接测工具逻辑，StdioServerParameters 启动真实子进程测命令行入口
• await c.list_tools()：列出工具；.tools 是工具列表，每个有 name、input_schema
• c.server_capabilities.model_dump()：server_capabilities 是一个 Pydantic 模型，model_dump() 转成字典。字典推导里 if v 只保留非空的能力
• bad.is_error 为 True，内容里带着原因，这就是 4.5 节说的”业务失败走正常结果”

看输出。

1. 协议版本是 2026-07-28，就是 4.5 节讲的无状态版本
2. 服务端能力自动包含了 prompts、resources、tools，因为三种都注册了
3. 越界调用没有让客户端抛异常，而是一个 is_error=True 的结果，模型可以读到”路径越界”然后换个路径
4. 提示词模板渲染成了一条 user 角色的消息，默认参数 target 自动填上了

自己改一改：

1. 在 read_file 里把 raise ToolError(...) 改成 raise ValueError(...)，看 bad.content[0].text 变成什么，体会为什么要用 SDK 的 ToolError
2. 给 read_file 加一个参数 offset: int = 1，看 input_schema 的变化
3. 再注册一个资源 build://toolchain，返回 toolchain 文件的内容，然后列出资源确认

4.7 我项目的 MCP server

agent/mcp_server.py（89 行）把 Agent 用的 5 个工具暴露成 MCP server。设计上有三点：

1. 复用同一套工具函数。

flowchart TD
    T["agent/tools.py<br/>list_files / read_file / write_file / run_command<br/>路径检查 白名单 凭据拦截 截断"]
    T --> A["build_agent.py<br/>手写循环"]
    T --> G["graph_agent.py<br/>LangGraph"]
    T --> M["mcp_server.py<br/>MCP server"]
    M --> C1[任意 MCP 客户端]

三个入口共用 tools.py，边界检查只写一份。11 篇讲过的”两个工具功能重叠，防护就只有弱的那个那么强”，放到入口层面也一样：如果 MCP 版本自己另写一套检查，迟早会和 Agent 版本不一致。工具描述也是从 tools.SCHEMAS 里读的，保证模型看到的说明一致。

2. 工作目录在启动时固定。

def create_server(workdir, kb_mode="bm25"):
    workdir = Path(workdir).resolve()
    ...
    @server.tool(description=DESCRIPTIONS["read_file"])
    def read_file(path: str, offset: int = 1, limit: int = tools.MAX_LINES) -> str:
        return guarded(tools.read_file, workdir, path, offset, limit)

workdir 不是工具参数，而是 create_server 的局部变量，被里面定义的函数”记住”了。这种写法叫闭包（closure）：内层函数可以使用外层函数的变量，外层函数返回后这些变量依然有效。好处是模型没法通过参数改工作目录。

3. 知识库按需加载。

kb = {}

@server.tool(description="交叉编译经验库。...")
def search_knowledge(query: str) -> str:
    if "kb" not in kb:
        kb["kb"] = knowledge.load(kb_mode)
    return kb["kb"].format(kb["kb"].search(query, topk=3))

向量检索要加载 bge-m3 模型，比较慢。第一次调用时才加载，存进字典 kb 里，之后直接用。用字典而不是普通变量，是因为内层函数里给外层变量重新赋值需要 nonlocal 声明，而修改字典的内容不需要。

测试（tests/test_mcp.py）验证了：5 个工具都在；正常读文件成功；read_file("../secret.txt") 和 run_command("cmake -E cat ../secret.txt") 都返回 is_error=True 且带原因；写文件成功；知识库能查到 k002；以 --kb none 通过 stdio 启动时没有 search_knowledge，且 .xbuild/zig.cmake 已经准备好。

局限：和 Agent 版本完全一样，11 篇讲的脚本绕过、环境变量泄露在 MCP 版本里同样存在；而且 MCP 客户端是别人的程序，调用前是否需要用户确认、调用记录怎么留，都由客户端决定，server 这边控制不了。

4.8 A2A：Agent 之间的协议

先说为什么会有 A2A。 MCP（2024 年 11 月，来历见 04 篇 4.11 节）解决了”Agent 怎么接工具”。到 2025 年，大公司内部开始有很多 Agent：报销的、招聘的、客服的，分别由不同团队、不同厂商、用不同框架做出来。让它们互相配合，要么每两个之间写一套专门的对接代码，要么把对方当成一个普通工具来调，可对方是个会自己规划、可能要跑几小时、中途还要问用户问题的 Agent，工具调用那种”一问一答”装不下。2025 年 4 月 9 日，Google 在 Cloud Next 大会上发布了 A2A，当时有 50 多家合作方（发布文章），文章里就写明它和 MCP 是互补关系；同年 6 月 23 日 Google 把它捐给了 Linux 基金会（Linux 基金会公告），由多家公司共同管理，不再归 Google 一家。

A2A（Agent2Agent Protocol）：让不同团队、不同框架做出来的 Agent 能互相发现、互相委派任务的开放协议。由 Google 发起，后来捐给 Linux 基金会，Apache 2.0 许可证，最新发布的规范版本是 1.0.0（官网，2026-09 查阅）。

和 MCP 的分工，A2A 官网自己的说法是：MCP 管 Agent 到工具的通信，A2A 管 Agent 到 Agent 的通信，两者互补，设计上就是配合使用的。

flowchart LR
    U[用户] --> TA[旅行规划 Agent]
    TA -->|A2A 发任务| FA[机票 Agent<br/>别的公司做的]
    TA -->|A2A 发任务| HA[酒店 Agent<br/>别的框架做的]
    FA -->|MCP| FT[航班查询 API]
    HA -->|MCP| HT[酒店库存数据库]
    TA -->|MCP| CAL[用户日历]

关键区别是”不透明”：通过 MCP 调工具，调用方知道函数名、参数、返回结构；通过 A2A 找另一个 Agent，只知道它声明的能力，它内部用什么模型、什么工具、怎么推理都不暴露。

核心概念（规范）：

Task 的状态（规范定义了 8 种）：

stateDiagram-v2
    [*] --> SUBMITTED
    SUBMITTED --> WORKING
    SUBMITTED --> REJECTED
    WORKING --> INPUT_REQUIRED
    INPUT_REQUIRED --> WORKING
    WORKING --> AUTH_REQUIRED
    AUTH_REQUIRED --> WORKING
    WORKING --> COMPLETED
    WORKING --> FAILED
    WORKING --> CANCELED
    COMPLETED --> [*]
    FAILED --> [*]
    CANCELED --> [*]
    REJECTED --> [*]

图里的状态转移是按状态名的含义画的示意，规范只列出了状态和哪些是终止状态，没有给这样一张完整的转移图。

为什么要有 INPUT_REQUIRED？ 被委派的 Agent 执行到一半可能发现信息不够（“您要经济舱还是商务舱？”），需要回头问调用方，这和 07 篇的人工审批暂停是同一类问题。

传输和获取结果：

长任务可能跑几小时，调用方不能一直挂着连接，所以推送通知是必要的。

现在要不要学？ 面试里被问到的概率比 MCP 低很多，岗位要求里也少见（00 篇用的 54 个岗位里，只有 2 个在加分项或技术清单里提到 A2A）。能说清楚”是什么、和 MCP 什么关系、核心是 Agent Card 和 Task”就够了。

4.9 什么时候用框架，什么时候自己写

flowchart TD
    Q0{业务人员要自己搭？<br/>流程固定、要快速上线？} -->|是| P[Dify / Coze 这类平台]
    Q0 -->|否| Q1{要中断恢复、人工审批<br/>或长时间运行？}
    Q1 -->|是| LG[LangGraph<br/>或基于它的 LangChain]
    Q1 -->|否| Q2{多个 Agent 分工、交接？}
    Q2 -->|是| MA[OpenAI Agents SDK / LangGraph<br/>先看 13 篇想清楚是否真需要]
    Q2 -->|否| Q3{需要现成的文件、命令、编辑工具<br/>做编码类 Agent？}
    Q3 -->|是| CAS[Claude Agent SDK]
    Q3 -->|否| Q4{主要是 RAG、数据源多？}
    Q4 -->|是| LI[LlamaIndex]
    Q4 -->|否| HW[手写循环<br/>几十行，完全可控]

这张图只是起点，实际还要看团队已有技术栈、观测和评测工具的集成。

无论选哪个，都建议保持两层独立：

1. 工具层独立于框架。工具就是普通函数，加上检查和截断，框架只负责调用。我项目的 tools.py 被三个入口共用，就是这个思路。以后换框架，工具不用重写
2. 评测独立于框架。评测脚本调的是统一的入口（我项目是 build_agent.build()，靠环境变量 AGENT_IMPL 切换手写或 LangGraph），换实现后同一套评测直接对比

我项目的最终结论（eval/REPORT.md）：对这种单 Agent、线性循环的任务，手写更直接；LangGraph 真正多给的是检查点和中断恢复，这在长任务（一次困难项目要跑两三分钟、几十万 token）里有实际价值。

还有一个诚实的补充：07 篇讲过，LangGraph 的检查点粒度是节点，工具节点执行到一半被杀，恢复后已执行的工具会再跑一次。框架给了续跑能力，但幂等仍然要自己做。

第五部分 对照项目

对照开源实现：pi

pi（05 篇第五部分介绍过）是 4.9 节决策图最下面”手写”那条路走到底的样子：不用 LangChain、LangGraph，从模型接口到主循环全部自己写，也不接 MCP。源码以 commit 7b4cfd6 为准。

分层。 它拆成三个包，别人可以只拿其中一层用：

主循环不知道有哪些具体工具，工具定义在最上面一层。这和本篇 4.9 节”工具层独立于框架”是同一个思路。

不接 MCP，改用”技能”。 README 明确写着不做 MCP，建议写带说明文档的命令行工具。做法是：每个技能是一个目录，里面有一个 SKILL.md（开头写名字和一句话描述），可能还有脚本。系统提示词里只列出每个技能的名字、描述和文件路径；模型判断任务用得上时，自己用 read 工具读 SKILL.md，再按里面的说明用 bash 执行脚本或命令（skills.ts L355-L383）。pi 实现的是公开的 Agent Skills 规范，也能直接加载 Claude Code、Codex 的技能目录。

两者不是谁取代谁。技能省上下文、写起来快，适合本来就有 shell 的编程 Agent；MCP 适合不给模型 shell 的应用（比如聊天客户端），或者需要按工具做权限控制的场景。我项目做 MCP server，是为了让 Claude Desktop、Cursor 这类客户端直接接入同一套带路径检查的工具，这层检查在技能方式下就没处放。

钩子和中间件，自己实现一层。 本篇速记页说 LangChain v1 用中间件给 Agent 加功能，pi 用”扩展”做同样的事。扩展是 TypeScript 模块，可以注册工具和命令，监听几十种事件（extensions.md）。比如 tool_call 事件在工具执行前触发，可以改参数或拦截；示例 permission-gate.ts 大约 30 行，命令里匹配到 rm -rf、sudo 就弹窗确认，没有界面时直接拦下。context 事件在每次调模型前触发，可以改要发出去的消息。

面试时怎么用这个例子。 被问”为什么不用框架”时，除了我自己手写 65 行和 LangGraph 126 行的对比，还可以举 pi：一个每天有人在用的编程 Agent，整条链路自己写，好处是每一层都能按需求设计，比如 07 篇的树形会话、06 篇的压缩规则；代价是这些都要自己写、自己维护，光 agent 这一个包的源码就有两万多行。框架省下的就是这部分工作量。

第六部分 追问清单

第七部分 闭卷自测

1. 框架主要替你做了哪些事？代价是什么？各举一个我项目里的例子。

2. LangChain 和 LangGraph 是什么关系？官方怎么建议在 Deep Agents、LangChain、LangGraph 之间选？

3. LangGraph 的状态、节点、边分别是什么？reducer 起什么作用？

4. 实验 1 里，为什么 custom 事件总是出现在同一节点的 updates 之前？消息数为什么是 8？

5. ToolNode 和 tools_condition 分别做什么？

6. MCP 的 host、client、server 各负责什么？client 和 server 是什么对应关系？规范对 server 能看到什么有什么要求？

7. MCP 三种原语分别由谁控制？什么东西适合做成 resource 而不是 tool？

8. MCP 2026-07-28 版在”状态”上有什么变化？为什么对远程部署有好处？

9. 实验 2 里如果工具抛的是 ValueError 而不是 SDK 的 ToolError，模型会看到什么？这对项目有什么启示？

10. 我项目的 MCP server 为什么要复用 tools.py？工作目录为什么不做成工具参数？

11. A2A 和 MCP 的核心区别是什么？A2A 的 Task 为什么需要 INPUT_REQUIRED 状态和推送通知？

12. 新项目做框架选型，你会考虑哪些问题？无论选哪个，哪两层建议保持独立？

13. pi 不接 MCP，而是用”技能”给 Agent 加能力。两种做法在上下文占用和使用前提上有什么不同？各适合什么场景？

延伸阅读

1. LangGraph 流式输出文档 — 七种流式模式和 v2 输出格式
2. LangChain 概览 — create_agent、中间件，以及和 LangGraph、Deep Agents 的关系
3. MCP 规范 2026-07-28：架构 — 角色、设计原则、无状态的能力协商
4. MCP 规范 2026-07-28：传输 — stdio 和 Streamable HTTP
5. A2A 协议规范 — Agent Card、Task 状态、三种协议绑定
6. OpenAI Agents SDK 文档 — Agent、交接、护栏、会话、追踪
7. Claude Agent SDK 概览 — 内置工具、钩子、子 Agent、权限
8. LlamaIndex 框架文档
9. Dify、Coze Studio — 两个开源的可视化平台
10. pi：Skills 和 Extensions（commit 7b4cfd6）— 不用框架、不接 MCP 的编程 Agent 怎么做能力扩展和事件钩子
11. Agent Skills 规范 和 Anthropic：Equipping agents for the real world with Agent Skills（2025-10）— SKILL.md 的格式和三层按需加载

下一篇：13 多 Agent——多个 Agent 分工听起来很美，什么时候真的值得，代价又是什么。