架构定位

能力扩展

通过能力包、外部证据、持久状态和标准协议扩展 Agent 能做的事情。

如何为 Agent 增加 Skill、检索、记忆和外部协议能力？

概念校准

MCP

它是什么: 连接 Tool、Resource 和 Prompt 的标准协议，可用于本地或远程传输。
它不是什么: 不等于远程 API，不等于 Agent，也不是 Agent 获得工具的唯一方式。

工具

它是什么: Agent 可通过明确参数、权限和结果合同调用的受约束能力。
它不是什么: 工具不等于 MCP；本地函数、命令或 HTTP 客户端也可以是工具。

Harness

它是什么: 承载模型、工具、状态、权限、日志、预算、人工确认和恢复的运行控制环境。
它不是什么: 它能限制风险并保留证据，但不能直接阻止模型产生幻觉。

MCP Server、SaaS 与流式协议#

实现 Tool、Resource、状态、Streamable HTTP、鉴权和客户端联调。

本章只做一件事Forge 通过 MCP 为三个场景接入不同外部能力。

开始前，Forge 已经具备已完成 Module 12 的“多 Agent 协作与任务编排”，其通过验收的 solution 是本章起点。

完成后，Forge 将能够让 Forge 通过 MCP 访问外部服务，三个场景分别接入不同工具。

卡住时的最小恢复点先用本机无鉴权 MCP Mock 验证 Schema 和错误合同。

下载本章实验仓库Python 3.12 · pytest · Pydantic · SQLite · deterministic Mock

阶段能力链

01定义可验证的工具参数合同，并拒绝缺失字段和未知参数02注册带 MIME 类型的资源，并把读取限制在明确允许的 URI 前缀内03声明模板变量并在渲染前验证缺失值和多余值04按序组装流式响应，并拒绝重复、乱序和终止后的事件05验证令牌有效期和权限作用域，并只构造最小授权请求头06只对可安全重试的请求执行退避，并为写请求绑定幂等键07协商客户端与服务端能力，并只从已确认游标恢复兼容会话

每一步只增加一种可验证能力；后一步建立在前一步已经通过的代码和测试上。

概念校准\n\n在开始操作代码前，先把本章涉及的概念放回正确的工程边界。\n\n
\n\n### MCP mcp\n\n- 它是什么: 用于连接 Tool、Resource 和 Prompt 的标准协议，可运行在本地或远程传输之上，并用明确 Schema 描述能力。\n- 它不是什么: 它不等于远程 API，不等于 Agent，也不是接入工具的唯一方式；协议连接成功后仍需治理身份、权限和副作用。\n- 与相邻概念的关系: Tool 是可执行能力，Resource 是可读取信息，Prompt 是模板；MCP 规范发现与调用，Harness 控制实际运行。\n- 在 HeatStack Forge 中的位置: Forge 验证 MCP 服务的能力清单、Schema 与身份范围，再把允许的能力注册进现有工具系统，而不是绕过治理。\n- 典型误用与修正: 误用是连接服务器后把全部能力交给模型。修正方法是允许列表、最小权限、超时、审计、高风险确认和断线测试。\n\n### 工具 tool\n\n- 它是什么: Agent 可通过明确参数、权限、执行边界和结果合同调用的受约束能力，例如函数、命令、数据库查询或 HTTP 客户端。\n- 它不是什么: 工具不等于 MCP，也不等于任意脚本。协议可以暴露能力，但 Harness 仍决定是否允许调用及如何处理副作用。\n- 与相邻概念的关系: Agent Loop 选择工具，Schema 约束参数，Harness 执行权限检查和隔离，Verification 检查结果是否推进目标。\n- 在 HeatStack Forge 中的位置: Forge 的 tool registry 保存合同与风险等级；执行前生成变更计划，高风险写操作等待确认，执行后记录结果与恢复信息。\n- 典型误用与修正: 误用是让模型根据工具名称自由拼参数。修正方法是严格 Schema、确定性校验、超时、幂等键和拒绝分支测试。\n\n### Harness harness\n\n- 它是什么: 承载模型、工具、状态、权限、日志、预算、人工确认、隔离和恢复的运行控制环境。\n- 它不是什么: 它可以缩小风险、拒绝动作并保留证据，但不能保证模型输出永远正确，也不能代替领域验收。\n- 与相邻概念的关系: Agent Loop 在 Harness 中运行；Tool、Memory 和 MCP 连接受其策略约束；Observability 记录过程，Evaluation 判断行为。\n- 在 HeatStack Forge 中的位置: Forge Harness 管理工作目录、工具、确认队列、预算、Checkpoint、审计日志和回滚，把模型建议与真实副作用分开。\n- 典型误用与修正: 误用是把系统提示词当成安全边界。修正方法是把权限、路径、网络、密钥、预算和回滚落实在模型之外的控制中。\n\n
#

STAGE 01

定义可验证的工具参数合同，并拒绝缺失字段和未知参数

先看一个具体问题#

为什么只检查必填字段还不够？#

在 Module 12 的验收方案中，Forge 已经能够编排多个 Agent 完成任务，但还没有通过 MCP 连接外部工具。

本阶段要创建 forge/mcp/schema.py，定义 ToolSchema 数据类，并实现 validate 方法来检查参数是否符合合同。

如果只检查必填字段，攻击者或意外调用可能传入未声明的参数，例如 shell，从而注入额外能力。

测试 test_unknown_arguments_are_rejected 要求当传入 shell 参数时抛出包含 unknown 的 ValueError。

先做判断#

在实现 ToolSchema.validate 时，如果只检查 required 字段而忽略 allowed 字段，会发生什么？

所有合法参数都能通过验证，未知参数也会被接受
未知参数会被自动忽略
验证会抛出 TypeError
只有缺失必填字段时才会报错

判断依据: 正确答案是第一个选项：只检查必填字段会让未知参数也被接受，因为验证逻辑没有拒绝不在 allowed 元组中的键。

它是什么#

工具 Schema 是参数形状的合同#

ToolSchema 用 required 元组声明必须出现的参数名，用 allowed 元组声明所有可接受的参数名。

validate 方法先检查缺失的必填参数，再检查未知参数，最后返回参数的副本。

这个合同只负责参数形状验证，不涉及执行权限或副作用控制。

它不是什么#

工具 Schema 不是安全边界#

Schema 验证通过并不意味着工具调用是安全的，因为执行环境、权限和副作用仍由 Harness 控制。

它不能防止模型产生错误参数值，只能拒绝形状不合法的输入。

它也不是 MCP 协议本身，而是 MCP 工具定义的一部分。

它与相邻概念的关系#

Schema 与 Harness、Agent Loop 的关系#

Agent Loop 选择工具后，Harness 使用 ToolSchema.validate 检查参数，再决定是否允许调用。

Schema 的 allowed 元组限定了可接受的参数集合，防止意外能力注入。

验证通过后，Harness 仍可基于策略拒绝执行，例如检测到危险参数值。

本阶段的边界决定: 工具 Schema 只负责参数形状验证，不负责执行权限或副作用控制；因此，即使验证通过，Harness 仍必须独立检查权限和副作用。

实验 ID: mcp-saas-streaming-01-tool-schema-contract
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/01-tool-schema-contract/
验收: 待验收

教师示范#

目标: 演示 ToolSchema 如何拒绝未知参数：构造一个允许 query 和 limit 的 schema，传入包含 shell 的参数，观察 ValueError 并解释原因。

假设我们定义了一个工具 search_docs，它只接受 query 和 limit 两个参数。

如果调用者传入 shell 参数，验证器必须拒绝，因为 shell 不在 allowed 元组中。

在实现中，我们通过列表推导式找出所有不在 allowed 中的键，如果列表非空则抛出 ValueError。

实施步骤#

创建 ToolSchema 实例，required=("query",)，allowed=("query", "limit")。
调用 validate 并传入 {"query": "release policy", "shell": "rm -rf"}。
验证器发现 shell 不在 allowed 中，抛出 ValueError，错误消息包含 unknown。

真实 solution 代码: forge/mcp/__init__.py

"""MCP protocol building blocks."""

真实 solution 代码: forge/mcp/schema.py

from dataclasses import dataclass

@dataclass(frozen=True)
class ToolSchema:
    name: str
    required: tuple[str, ...]
    allowed: tuple[str, ...]

    def validate(self, arguments: dict[str, object]) -> dict[str, object]:
        missing = [key for key in self.required if key not in arguments]
        unknown = [key for key in arguments if key not in self.allowed]
        if missing:
            raise ValueError("missing required arguments: " + ", ".join(missing))
        if unknown:
            raise ValueError("unknown arguments: " + ", ".join(unknown))
        return dict(arguments)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

validate 方法首先检查缺失的必填参数，这里 query 存在，所以没有缺失。

然后检查未知参数：shell 不在 allowed 元组中，因此被加入 unknown 列表。

由于 unknown 列表非空，抛出 ValueError，消息中包含 unknown 和参数名。

带提示练习#

补全 ToolSchema.validate 中缺失的未知参数检查，使测试 test_unknown_arguments_are_rejected 通过。

打开 forge/mcp/schema.py，找到 validate 方法。
在检查缺失参数之后，添加代码找出所有不在 self.allowed 中的参数键。
如果未知参数列表非空，抛出 ValueError，消息格式为 unknown arguments: 后接逗号分隔的参数名。
运行 python -m pytest -q tests/test_stage.py 验证两个测试都通过。

逐级提示#

使用列表推导式：[key for key in arguments if key not in self.allowed]。
错误消息应包含 unknown 单词，以满足测试断言。
确保在返回参数副本之前检查未知参数。

完成标准

test_valid_arguments_are_copied 通过，验证合法参数返回副本。
test_unknown_arguments_are_rejected 通过，验证未知参数抛出 ValueError。

故障注入与排查#

当前 forge/mcp/schema.py 中的 validate 方法只检查了缺失的必填参数，没有检查未知参数。因此，当测试 test_unknown_arguments_are_rejected 传入 shell 参数时，验证器错误地接受了它，导致测试失败并抛出 AssertionError: unknown tool argument was accepted。

本阶段已经学习了 ToolSchema 的 allowed 元组的作用，并且知道如何通过列表推导式找出未知参数。因此，我们可以定位到 validate 方法中缺少的未知参数检查，并添加相应的代码来修复缺陷。

已验证的故障合同#

目标测试: test_unknown_arguments_are_rejected
故障文件: forge/mcp/schema.py

诊断顺序#

运行 python -m pytest -q ../tests/test_stage.py::test_unknown_arguments_are_rejected，观察测试失败。
查看 forge/mcp/schema.py 的 validate 方法，发现只有缺失参数检查，没有未知参数检查。
确认 allowed 元组包含 query 和 limit，但传入的 shell 不在其中。
得出结论：需要添加未知参数检查，拒绝不在 allowed 中的键。

修复原则: 在 validate 方法中添加未知参数检查：计算 unknown = [key for key in arguments if key not in self.allowed]，如果 unknown 非空则抛出 ValueError("unknown arguments: " + ", ".join(unknown))。

python -m pytest -q ../tests/test_stage.py::test_unknown_arguments_are_rejected

迁移挑战#

在办公研究场景中，你需要使用 ToolSchema 定义工具 search_docs，要求参数 query，允许参数 limit 和 sort，并验证传入额外参数 file_type 时抛出 ValueError。

改变的条件: 将允许参数集合从 (‘query’, ‘limit’) 扩展为 (‘query’, ‘limit’, ‘sort’)，并传入额外参数 ‘file_type’ 来验证拒绝行为。

编写一个 Python 脚本，创建 ToolSchema 实例，名称为 ‘search_docs’，required=(‘query’,)，allowed=(‘query’, ‘limit’, ‘sort’)。然后调用 validate 方法，传入参数 {‘query’: ‘quarterly report’, ‘sort’: ‘date’}，验证返回副本；再传入参数 {‘query’: ‘quarterly report’, ‘file_type’: ‘pdf’}，捕获 ValueError 异常并打印异常消息。

完成标准

脚本成功创建 ToolSchema 实例，allowed 包含 ‘sort’。
调用 validate 传入包含 ‘sort’ 的参数时，返回参数副本且不抛出异常。
调用 validate 传入包含 ‘file_type’ 的参数时，捕获到 ValueError 异常，且异常消息包含 ‘unknown’ 和 ‘file_type’。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么仅检查必填字段不足以防止未知参数？请结合 ToolSchema 的 allowed 元组解释。

一个有力回答应包含#

必填字段检查只能确保 required 中的参数存在，但无法阻止调用者传入未声明的参数。
allowed 元组定义了所有可接受的参数名，验证器必须拒绝不在其中的键，否则可能引入意外能力。
未知参数可能被下游代码误用，导致安全漏洞或行为异常。

掌握证据

ToolSchema.validate 返回合法参数的副本，测试 test_valid_arguments_are_copied 通过。
ToolSchema.validate 对未知参数抛出 ValueError，测试 test_unknown_arguments_are_rejected 通过。
在办公研究场景中，使用 ToolSchema 定义 search_docs 并验证拒绝额外参数 file_type。

STAGE 02

注册带 MIME 类型的资源，并把读取限制在明确允许的 URI 前缀内

先看一个具体问题#

如何防止资源注册越界，确保只能访问允许的 URI 前缀？#

当前 forge/mcp/schema.py 已经存在，但还没有资源注册表。测试文件 tests/test_stage.py 中导入了 forge.mcp.resources 模块，但该模块尚未创建，导致测试收集阶段直接报错 ModuleNotFoundError: No module named 'forge.mcp.resources'。

你需要创建 forge/mcp/resources.py，实现 Resource 数据类和 ResourceRegistry 类。ResourceRegistry 在构造时接收一个允许的 URI 前缀元组，并在注册资源时强制检查 URI 是否以这些前缀开头。

如果注册的 URI 不在允许前缀内，必须抛出 ValueError，且错误消息中包含 "allowed" 字样。同时，注册表还需要支持读取已注册的资源，并保留其 MIME 类型。

先做判断#

在 ResourceRegistry.register 中，如果传入的 resource.uri 不以 self.allowed_prefixes 中的任何一个前缀开头，应该如何处理？

直接存储资源，因为注册表只负责存储，不负责校验。
抛出 ValueError，并提示 URI 不在允许前缀内。
静默忽略该资源，不存储也不报错。
将 URI 自动转换为允许前缀下的路径再存储。

判断依据: 正确选项是抛出 ValueError。注册表必须执行前缀策略，否则越界资源会被接受，导致后续读取时可能泄露不应访问的内容。静默忽略或自动转换都会掩盖错误，使调用方无法感知配置问题。

它是什么#

ResourceRegistry 是什么#

ResourceRegistry 是一个内存中的资源存储容器，它维护一个从 URI 到 Resource 对象的映射，并在注册时强制实施 URI 前缀策略。

它通过 allowed_prefixes 元组定义可接受的 URI 命名空间，例如 ("forge://workspace/",) 表示只允许以 forge://workspace/ 开头的 URI。

注册方法 register 会检查资源 URI 是否以任一允许前缀开头，如果不符合则抛出 ValueError，从而防止越界资源进入注册表。

它不是什么#

ResourceRegistry 不是什么#

它不是内容解析器或传输层：它不关心资源内容的具体格式，也不负责将资源发送给客户端。

它不是安全边界本身：它只执行前缀检查，但调用方仍需确保 allowed_prefixes 的配置正确，且后续读取时也要再次校验 URI。

它不是全局资源管理器：每个注册表实例独立维护自己的映射，不同实例之间不共享数据。

它与相邻概念的关系#

与其他组件的关系#

Resource 数据类是不可变对象，包含 uri、mime_type 和 content 三个字段，注册表存储这些对象的引用。

ResourceRegistry 依赖 allowed_prefixes 来定义边界，而 register 和 read 方法都使用这个边界进行校验。

测试 test_registered_resource_preserves_media_type 验证注册后读取的资源 MIME 类型保持不变，而 test_resource_outside_allowed_prefix_is_rejected 验证越界资源被拒绝。

本阶段的边界决定: 注册表只负责存储和前缀策略，不负责内容解析或传输。因此，在 register 中必须检查 URI 前缀，否则越界资源会被接受；在 read 中也要检查 URI 前缀，防止调用方直接读取未注册的越界 URI。

实验 ID: mcp-saas-streaming-02-resource-registry
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/02-resource-registry/
验收: 待验收

教师示范#

目标: 演示 ResourceRegistry 如何拒绝越界资源：创建允许前缀 forge://workspace/ 的注册表，尝试注册 file:///etc/passwd，观察 ValueError。

假设你正在构建一个 MCP 服务器，需要暴露工作区内的资源。你创建了一个 ResourceRegistry，只允许 forge://workspace/ 前缀。

现在有人试图注册一个本地文件 file:///etc/passwd，这个 URI 不在允许前缀内。如果注册表不检查前缀，这个敏感文件就会被暴露。

正确的实现应该在 register 方法中检查 URI 是否以 self.allowed_prefixes 中的任一前缀开头，如果不满足则抛出 ValueError。

实施步骤#

创建 ResourceRegistry 实例，传入 ("forge://workspace/",) 作为允许前缀。
构造一个 Resource 对象，URI 为 file:///etc/passwd，MIME 类型为 text/plain，内容为 b"secret"。
调用 registry.register(resource)，预期抛出 ValueError，因为 URI 不在允许前缀内。

真实 solution 代码: forge/mcp/resources.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Resource:
    uri: str
    mime_type: str
    content: bytes

class ResourceRegistry:
    def __init__(self, allowed_prefixes: tuple[str, ...]):
        self.allowed_prefixes = allowed_prefixes
        self._items: dict[str, Resource] = {}

    def register(self, resource: Resource) -> None:
        if not resource.uri.startswith(self.allowed_prefixes):
            raise ValueError("resource URI is outside allowed prefixes")
        if resource.uri in self._items:
            raise ValueError("resource URI must be unique")
        self._items[resource.uri] = resource

    def read(self, uri: str) -> Resource:
        if not uri.startswith(self.allowed_prefixes):
            raise PermissionError("resource URI is outside allowed prefixes")
        return self._items[uri]

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

测试期望注册越界资源时抛出 ValueError，但当前实现没有检查前缀，直接存储了资源，导致测试失败。

修复方法是在 register 方法开头添加前缀检查：if not resource.uri.startswith(self.allowed_prefixes): raise ValueError("resource URI is outside allowed prefixes")。

这样，任何不在允许前缀内的 URI 都会被拒绝，从而保证注册表只包含授权命名空间内的资源。

带提示练习#

在 ResourceRegistry.register 中添加 URI 前缀检查，使测试 test_resource_outside_allowed_prefix_is_rejected 通过。

打开 forge/mcp/resources.py 文件（如果不存在则创建）。
定义 Resource 数据类，包含 uri、mime_type、content 三个字段，并使用 @dataclass(frozen=True) 装饰。
定义 ResourceRegistry 类，构造函数接收 allowed_prefixes: tuple[str, ...] 并存储，同时初始化空字典 self._items。
实现 register 方法：首先检查 resource.uri 是否以 self.allowed_prefixes 中的任一前缀开头，如果不是则抛出 ValueError，错误消息中包含 "allowed"。
实现 read 方法：检查 uri 是否以允许前缀开头，如果不是则抛出 PermissionError，然后从 self._items 中返回对应资源。

逐级提示#

使用 str.startswith(tuple) 可以一次检查多个前缀。
错误消息中必须包含 "allowed" 字样，否则测试断言会失败。
read 方法中如果 URI 不在允许前缀内，应抛出 PermissionError 而不是 ValueError，以区分注册和读取的不同错误类型。

完成标准

运行 python -m pytest -q tests/test_stage.py 后，两个测试均通过。
test_resource_outside_allowed_prefix_is_rejected 不再抛出 AssertionError，而是成功捕获 ValueError。

故障注入与排查#

故障夹具 fixtures/failure.json 描述了一个资源作用域绕过缺陷：注册表接受配置的 URI 命名空间之外的资源。在 forge/mcp/resources.py 中，register 方法缺少前缀强制检查，导致 file:// 资源被 forge:// 注册表接受。

本阶段的核心学习成果就是实现 URI 前缀约束。故障夹具中的缺陷正是缺少这一约束，而你已经通过引导练习掌握了如何在 register 中添加前缀检查。因此，你可以直接定位到 register 方法，补上缺失的校验逻辑，使测试 test_resource_outside_allowed_prefix_is_rejected 通过。

已验证的故障合同#

目标测试: test_resource_outside_allowed_prefix_is_rejected
故障文件: forge/mcp/resources.py

诊断顺序#

运行故障命令 python -m pytest -q ../tests/test_stage.py::test_resource_outside_allowed_prefix_is_rejected，观察到测试失败，错误为 AssertionError: out-of-scope resource was registered。
查看 forge/mcp/resources.py 中的 register 方法，发现它直接执行 self._items[resource.uri] = resource，没有检查 URI 前缀。
对比测试期望：测试尝试注册 file:///etc/passwd，期望抛出 ValueError，但实际没有抛出，说明前缀检查缺失。
修复：在 register 方法开头添加 if not resource.uri.startswith(self.allowed_prefixes): raise ValueError("resource URI is outside allowed prefixes")。

修复原则: 在 ResourceRegistry.register 方法中，存储资源之前先检查 resource.uri 是否以 self.allowed_prefixes 中的任一前缀开头。如果不满足，抛出 ValueError，错误消息包含 "allowed"。这样，越界资源在注册时就会被拒绝，测试 test_resource_outside_allowed_prefix_is_rejected 将通过。

python -m pytest -q ../tests/test_stage.py::test_resource_outside_allowed_prefix_is_rejected

迁移挑战#

在音乐创作场景中，你需要使用 ResourceRegistry 管理歌词文件。允许的前缀是 forge://workspace/，你注册了一个资源 forge://workspace/lyrics.txt，MIME 类型为 text/plain，内容为 b"La la la"。然后尝试注册一个外部 URL https://example.com/lyrics.txt，预期会抛出 ValueError。

改变的条件: 将资源 URI 从工作区路径改为外部 HTTPS URL，以测试前缀约束是否仍然有效。

编写一个 Python 脚本或交互式会话，创建 ResourceRegistry 实例，允许前缀为 ("forge://workspace/",)。注册 Resource("forge://workspace/lyrics.txt", "text/plain", b"La la la")，然后尝试注册 Resource("https://example.com/lyrics.txt", "text/plain", b"stolen")，并捕获异常。

完成标准

注册 forge://workspace/lyrics.txt 成功，且后续 read 返回的 MIME 类型为 text/plain。
尝试注册 https://example.com/lyrics.txt 时抛出 ValueError，错误消息中包含 "allowed"。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么在 register 和 read 方法中都要检查 URI 前缀？如果只在 register 中检查，而 read 不检查，会有什么安全风险？

一个有力回答应包含#

register 检查防止越界资源进入注册表，保证存储的数据都在授权命名空间内。
read 检查防止调用方直接请求未注册的越界 URI，即使注册表中没有该 URI，也能提前拒绝，避免信息泄露或错误访问。
如果只在 register 中检查，攻击者可能构造一个不在注册表中的 URI 直接调用 read，如果 read 不检查前缀，可能会返回意外结果或暴露内部实现细节。

掌握证据

ResourceRegistry.register 拒绝不在允许前缀内的 URI，测试 test_resource_outside_allowed_prefix_is_rejected 通过。
ResourceRegistry.read 返回已注册资源并保留正确的 MIME 类型，测试 test_registered_resource_preserves_media_type 通过。
运行 python -m pytest -q tests/test_stage.py 后，两个测试均通过，退出码为 0。

STAGE 03

声明模板变量并在渲染前验证缺失值和多余值

先看一个具体问题#

缺失变量被静默替换为空字符串，导致提示不完整#

在上一阶段，我们实现了 ResourceRegistry，但还没有提示模板。现在需要创建 forge/mcp/prompts.py 并实现 PromptTemplate 类。

测试 test_missing_prompt_variable_is_rejected 要求：当渲染提示模板时，如果缺少声明的变量，必须抛出 ValueError，而不是静默地用空字符串填充。

当前起始代码中 forge/mcp/prompts.py 不存在，运行测试会报 ModuleNotFoundError。我们需要从零实现 PromptTemplate，并确保变量验证逻辑正确。

先做判断#

如果 PromptTemplate 的 render 方法在缺少变量时用空字符串填充，会发生什么？

渲染结果不完整，但不会报错，测试会失败
渲染结果完整，测试通过
抛出 ValueError，测试通过
渲染结果不完整，但测试通过

判断依据: 正确答案是第一个选项。如果缺少变量时用空字符串填充，渲染结果会缺少必要信息，测试 test_missing_prompt_variable_is_rejected 会失败，因为它期望抛出 ValueError。

它是什么#

PromptTemplate 是变量合同与渲染器#

PromptTemplate 声明了渲染所需的变量集合，并在渲染前验证提供的值是否完全匹配。

它通过 variables 元组明确列出所有必需变量，render 方法检查缺失和多余变量，确保合同被遵守。

渲染使用 Python 的 str.format 机制，将模板中的占位符替换为提供的值。

它不是什么#

PromptTemplate 不负责提示内容质量或模型调用#

PromptTemplate 不评估提示文本是否有效或高质量，它只保证变量替换的完整性。

它不涉及模型调用、API 请求或任何外部副作用，仅是一个纯数据转换组件。

它不处理变量值的类型转换或默认值，所有变量必须显式提供。

它与相邻概念的关系#

PromptTemplate 与 ResourceRegistry 的关系#

ResourceRegistry 管理可读取的资源，而 PromptTemplate 管理可渲染的提示模板，两者都是 MCP 能力的一部分。

PromptTemplate 可以引用 ResourceRegistry 中的资源内容作为变量值，但变量验证独立于资源存在性。

两者都遵循明确的 Schema 约束，但 PromptTemplate 专注于变量合同，ResourceRegistry 专注于资源发现。

本阶段的边界决定: PromptTemplate 只负责变量合同和渲染，不负责提示内容质量或模型调用；缺失或多余变量必须抛出 ValueError，而不是静默处理。

实验 ID: mcp-saas-streaming-03-prompt-template-contract
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/03-prompt-template-contract/
验收: 待验收

教师示范#

目标: 演示 PromptTemplate 如何拒绝缺失变量：创建模板 “Summarize {topic} for {audience}.”，只提供 topic，观察 ValueError。

假设我们有一个提示模板 “Summarize {topic} for {audience}.”，它声明了变量 topic 和 audience。

如果我们只提供 topic 的值，而缺少 audience，那么渲染时应该抛出 ValueError，而不是生成 “Summarize MCP for .” 这样的不完整提示。

在实现中，我们首先计算缺失变量集合 missing = set(self.variables) - set(values)，然后检查是否为空。

如果 missing 非空，抛出 ValueError，错误消息中包含 “missing” 关键字，以便测试断言。

实施步骤#

创建 PromptTemplate 实例：prompt = PromptTemplate(“brief”, “Summarize {topic} for {audience}.”, (“topic”, “audience”))
调用 prompt.render({“topic”: “MCP”})，只提供 topic，缺少 audience。
在 render 方法中，计算 missing = {“audience”}，发现缺失变量。
抛出 ValueError，消息如 “prompt variables mismatch missing=[‘audience’] extra=[]”。

真实 solution 代码: forge/mcp/prompts.py

from dataclasses import dataclass

@dataclass(frozen=True)
class PromptTemplate:
    name: str
    template: str
    variables: tuple[str, ...]

    def render(self, values: dict[str, str]) -> str:
        missing = set(self.variables) - set(values)
        extra = set(values) - set(self.variables)
        if missing or extra:
            raise ValueError(f"prompt variables mismatch missing={sorted(missing)} extra={sorted(extra)}")
        return self.template.format(**values)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

缺失变量检查确保所有声明的变量都被提供，避免生成不完整的提示。

使用集合差集计算缺失变量，简单高效，且能清晰报告哪些变量缺失。

错误消息中包含 ‘missing’ 关键字，满足测试的断言要求。

带提示练习#

在 PromptTemplate.render 中添加缺失和多余变量检查，使测试 test_missing_prompt_variable_is_rejected 通过。

创建 forge/mcp/prompts.py 文件，定义 PromptTemplate 数据类，包含 name、template、variables 字段。
实现 render 方法，首先计算缺失变量集合 missing = set(self.variables) - set(values)。
计算多余变量集合 extra = set(values) - set(self.variables)。
如果 missing 或 extra 非空，抛出 ValueError，消息中包含 ‘missing’ 和 ‘extra’ 信息。
否则，使用 self.template.format(**values) 渲染模板并返回结果。

逐级提示#

使用集合操作可以方便地找出缺失和多余的键。
错误消息中必须包含 ‘missing’ 字符串，因为测试断言了这一点。
渲染时使用 format 方法，而不是 format_map，以便在变量不匹配时抛出 KeyError。

完成标准

运行 python -m pytest -q tests/test_stage.py 时，test_missing_prompt_variable_is_rejected 通过。
test_declared_variables_render_deterministically 也通过，确保正常渲染不受影响。

故障注入与排查#

故障夹具把 render 中的严格变量校验替换成了空字符串默认值。于是 audience 缺失时，模板仍然返回内容不完整的提示，test_missing_prompt_variable_is_rejected 会失败。

本阶段刚刚建立了变量合同：模板声明的变量必须与调用方提供的变量完全匹配。这个故障正好破坏该合同，因此可以通过比较声明集合与实际输入，定位到 render 的边界。

已验证的故障合同#

目标测试: test_missing_prompt_variable_is_rejected
故障文件: forge/mcp/prompts.py

诊断顺序#

运行 python -m pytest -q ../tests/test_stage.py::test_missing_prompt_variable_is_rejected，并保留失败断言。
检查 forge/mcp/prompts.py，确认 render 为缺失变量提供了空字符串，而不是拒绝输入。
比较 set(self.variables) 与 set(values)；缺少 audience 正是提示不完整的直接原因。
恢复缺失变量和多余变量的严格检查，只有两组集合都为空时才执行渲染。

修复原则: 在 render 开始时计算 missing = set(self.variables) - set(values) 和 extra = set(values) - set(self.variables)；任一集合非空就抛出 ValueError，否则调用 self.template.format(**values)。

python -m pytest -q ../tests/test_stage.py::test_missing_prompt_variable_is_rejected

迁移挑战#

办公研究：渲染一个带截止时间的简报提示，并验证变量合同的两种拒绝路径。

改变的条件: 场景从软件交付切换到办公研究，模板变量从 topic/audience 扩展为 topic/audience/deadline；PromptTemplate API 和校验规则保持不变。

使用现有 PromptTemplate API 创建模板“Prepare a brief on {topic} for {audience} by {deadline}.”，并声明 topic、audience、deadline 三个变量。先提供三个值并验证正常渲染；再省略 deadline，确认 ValueError 包含 missing；最后加入未声明的 location，确认 ValueError 包含 extra。

完成标准

提供完整输入时，结果是确定性的简报提示，并包含主题、受众和截止时间。
省略 deadline 必须抛出 ValueError，错误消息包含 missing。
加入未声明的 location 必须抛出 ValueError，错误消息包含 extra；使用现有 PromptTemplate API 和测试，不新增接口或测试文件。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么多余变量也必须被拒绝？请解释拒绝多余变量对提示模板合同完整性的意义。

一个有力回答应包含#

多余变量可能表明调用方误解了模板的变量合同，或者传递了不相关的数据。
拒绝多余变量可以防止意外忽略重要信息，确保渲染结果符合预期。
它强制调用方精确匹配声明的变量，提高代码的健壮性和可维护性。

掌握证据

PromptTemplate.render 对有效变量返回确定性字符串，如 ‘Summarize MCP for engineers.’。
PromptTemplate.render 对缺失变量抛出 ValueError，错误消息包含 ‘missing’。
PromptTemplate.render 对多余变量抛出 ValueError，错误消息包含 ‘extra’。

STAGE 04

按序组装流式响应，并拒绝重复、乱序和终止后的事件

先看一个具体问题#

流式事件必须按序到达且终止后不可继续#

在流式协议中，事件可能因为网络重传、并发推送或客户端错误而乱序到达，但最终拼接结果必须与发送顺序一致。

如果组装器盲目接受任何序列号，就会把缺失事件后的数据错误拼接，导致输出不可验证。

本阶段需要实现 StreamAssembler，它只接受连续递增的序列号，并在收到 done 事件后拒绝任何后续事件。

当前起始代码中 forge/mcp/streaming.py 文件不存在，测试 test_sequence_gap_is_rejected 会因导入错误而失败。

先做判断#

如果 StreamAssembler.push 收到序列号 3 而当前期望序列号是 2，应该发生什么？

接受事件并更新期望序列号为 4
抛出 ValueError，因为序列不连续
忽略该事件并等待序列号 2
自动补一个空事件再接受

判断依据: 正确行为是抛出 ValueError，因为流式协议要求事件严格按序到达，任何间隙都意味着数据丢失或乱序，必须显式拒绝而不是静默修复。

它是什么#

StreamAssembler 是什么#

StreamAssembler 是一个有状态对象，维护 next_sequence（下一个期望的序列号）、chunks（已接收的文本块列表）和 finished（是否已收到终止事件）。

它的 push 方法负责验证事件序列连续性、事件类型合法性以及终止状态，然后才应用事件数据。

text 方法只有在 finished 为真时才返回拼接结果，否则抛出异常，防止返回不完整内容。

它不是什么#

StreamAssembler 不是什么#

它不负责网络传输、重传或缓存乱序事件，这些属于传输层或更高层协议。

它不解析事件内容或验证业务语义，只关注序列号和终止状态。

它不是线程安全的，如果多个线程同时推送事件，需要外部同步机制。

它与相邻概念的关系#

与其他组件的关系#

StreamAssembler 使用 next_sequence 作为游标，与 push 中的序列号比较，确保每次只接受期望的下一个事件。

finished 状态由 done 事件设置，一旦为真，后续任何 push 都会立即抛出异常。

chunks 列表按接收顺序存储文本块，text 方法在终止后将其拼接为最终字符串。

本阶段的边界决定: StreamAssembler 只负责顺序和终止状态，不负责网络传输或重传；任何乱序或终止后的事件都必须被拒绝，而不是尝试修复。

实验 ID: mcp-saas-streaming-04-ordered-stream-events
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/04-ordered-stream-events/
验收: 待验收

教师示范#

目标: 演示 StreamAssembler 如何拒绝序列间隙：推送序列 1 后直接推送序列 3，观察 ValueError。

我们从一个空的 StreamAssembler 开始，它的 next_sequence 初始为 1。

首先推送序列 1 的 chunk 事件，数据为 “hello”，此时 next_sequence 更新为 2。

然后尝试推送序列 3 的 done 事件，由于期望序列是 2，push 方法检测到不连续并抛出 ValueError。

实施步骤#

创建 StreamAssembler 实例。
调用 push(1, "chunk", "hello")，成功添加数据并更新 next_sequence 为 2。
调用 push(3, "done")，触发序列连续性检查，抛出 ValueError。

真实 solution 代码: forge/mcp/streaming.py

from dataclasses import dataclass, field

@dataclass
class StreamAssembler:
    next_sequence: int = 1
    chunks: list[str] = field(default_factory=list)
    finished: bool = False

    def push(self, sequence: int, kind: str, data: str = "") -> None:
        if self.finished:
            raise ValueError("stream is already terminal")
        if sequence != self.next_sequence:
            raise ValueError("stream sequence must be contiguous")
        if kind not in {"chunk", "done"}:
            raise ValueError("unknown stream event")
        if kind == "chunk":
            self.chunks.append(data)
        else:
            self.finished = True
        self.next_sequence += 1

    def text(self) -> str:
        if not self.finished:
            raise ValueError("stream is incomplete")
        return "".join(self.chunks)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

push 方法首先检查 finished 状态，然后比较传入的 sequence 与 self.next_sequence。

如果不相等，立即抛出 ValueError，阻止事件被应用，从而保护内部状态不被破坏。

带提示练习#

在 StreamAssembler.push 中添加序列连续性检查，使测试 test_sequence_gap_is_rejected 通过。

创建 forge/mcp/streaming.py 文件，定义 StreamAssembler 数据类。
在 push 方法中，在检查 finished 之后添加序列连续性检查：如果 sequence != self.next_sequence，抛出 ValueError。
添加事件类型检查：如果 kind 不是 chunk 或 done，抛出 ValueError。
根据 kind 更新状态：chunk 追加数据，done 设置 finished 为 True。
更新 next_sequence 为 sequence + 1。

逐级提示#

序列检查必须在任何状态修改之前进行。
错误消息应包含 ‘contiguous’ 以通过测试断言。
不要忘记处理未知事件类型。

完成标准

test_sequence_gap_is_rejected 测试通过。
test_stream_assembles_only_after_done 测试通过。

故障注入与排查#

当前 forge/mcp/streaming.py 中的 StreamAssembler.push 方法缺少序列连续性检查，导致测试 test_sequence_gap_is_rejected 失败。该测试期望在推送序列 1 后直接推送序列 3 时抛出包含 ‘contiguous’ 的 ValueError，但实际代码接受了序列 3 并更新了游标，没有抛出异常。

本阶段的核心机制就是序列连续性检查，学习者已经通过 worked example 和 guided practice 理解了如何比较 sequence 与 next_sequence。现在需要诊断实际代码中缺失该检查的原因，并应用相同的修复模式。

已验证的故障合同#

目标测试: test_sequence_gap_is_rejected
故障文件: forge/mcp/streaming.py

诊断顺序#

运行测试 python -m pytest -q tests/test_stage.py，观察失败输出。
检查 forge/mcp/streaming.py 中的 push 方法，发现没有比较 sequence 和 self.next_sequence 的代码。
确认缺陷：代码直接接受任何序列号并更新 next_sequence，导致间隙被忽略。

修复原则: 在 push 方法中，在检查 finished 之后添加 if sequence != self.next_sequence: raise ValueError("stream sequence must be contiguous")，确保只有连续序列才能被接受。

python -m pytest -q ../tests/test_stage.py::test_sequence_gap_is_rejected

迁移挑战#

在办公研究场景中，你需要同时处理两个独立的流式响应，分别组装简报的背景和结论部分。每个流使用独立的 StreamAssembler 实例，按序推送文本块和终止事件，验证各自拼接结果，并确认终止后推送额外块会抛出异常且不影响另一个流。

改变的条件: 将使用场景从单一流扩展到同时处理多个独立的流，每个流代表一个办公研究简报的不同部分，要求学习者管理多个 StreamAssembler 实例并确保每个实例独立维护顺序和终止状态。

编写一个 Python 脚本，创建两个 StreamAssembler 实例，分别用于组装办公研究简报的两个部分：背景和结论。对于第一个实例，依次推送序列 1、2 的 chunk 事件，数据分别为 ‘办公研究’ 和 ‘：背景’，然后推送序列 3 的 done 事件。对于第二个实例，依次推送序列 1、2 的 chunk 事件，数据分别为 ‘决策’ 和 ‘简报’，然后推送序列 3 的 done 事件。分别调用 text() 验证返回 ‘办公研究：背景’ 和 ‘决策简报’。最后，尝试向第一个实例推送序列 4 的 chunk 事件，捕获并打印 ValueError 消息。然后调用第二个实例的 text() 并确认返回 ‘决策简报’。

完成标准

脚本成功运行，两个实例的 text() 分别返回 ‘办公研究：背景’ 和 ‘决策简报’。
在第一个实例的 done 后推送额外 chunk 时抛出 ValueError，且消息表明流已终止（例如包含 ‘terminal’）。
两个实例的状态相互独立，第二个实例不受第一个实例终止后推送的影响，其 text() 仍正确返回 ‘决策简报’。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么终止后的事件必须被拒绝？如果允许终止后继续推送数据，会对流式响应的完整性和可验证性造成什么具体影响？

一个有力回答应包含#

终止事件表示流已经完成，任何后续事件都意味着协议违规或数据错乱。
允许终止后推送会破坏 finished 状态的语义，导致 text() 可能返回不完整或错误拼接的内容。
拒绝终止后事件可以及早暴露客户端或传输层的错误，避免静默产生不可信结果。

掌握证据

StreamAssembler.text() 仅在收到 done 后返回拼接结果，否则抛出 ValueError。
StreamAssembler.push 对非连续序列抛出包含 ‘contiguous’ 的 ValueError。
StreamAssembler.push 在 finished 为真时拒绝任何后续事件，抛出包含 ‘terminal’ 的 ValueError。

STAGE 05

验证令牌有效期和权限作用域，并只构造最小授权请求头

先看一个具体问题#

只读令牌被误用于写操作#

在 SaaS 集成中，我们经常拿到一个访问令牌（AccessToken），它带有有效期和一组作用域（scopes）。

当前代码中，AccessToken.headers_for 只检查了令牌是否过期，却没有检查请求的操作是否在令牌允许的作用域内。

这导致一个只有 issues:read 权限的令牌，也能为 issues:write 操作生成 Authorization 头，从而越权调用写接口。

我们需要在生成授权头之前，同时验证令牌未过期且具备所需作用域，否则抛出 PermissionError。

先做判断#

在 AccessToken.headers_for 中，如果令牌未过期但缺少所需作用域，当前代码会怎样？

抛出 PermissionError，因为作用域检查已经存在
返回 Authorization 头，因为只检查了过期时间
返回空字典，因为缺少作用域
抛出 ValueError，因为作用域格式错误

判断依据: 当前代码只检查了 now >= self.expires_at，没有检查 required_scope 是否在 self.scopes 中。因此，只要令牌未过期，就会返回 Authorization 头，即使作用域不匹配。

它是什么#

AccessToken 是一个最小授权决策单元#

AccessToken 是一个不可变数据类，封装了令牌值、作用域集合和过期时间。

它的 headers_for 方法在生成授权头之前执行两个检查：先检查过期，再检查作用域。

这种顺序确保我们不会为一个已经无效的令牌浪费作用域检查，也避免泄露令牌值。

它不是什么#

AccessToken 不负责获取或刷新令牌#

AccessToken 只负责验证和头部生成，不负责从 OAuth 服务器获取新令牌或刷新过期令牌。

它也不负责实际的 HTTP 调用或权限策略的全局管理；它只针对单个请求的所需作用域做局部判断。

它与相邻概念的关系#

过期检查与作用域检查的顺序关系#

过期检查是前提：如果令牌已过期，无论作用域如何都不能使用，因此先抛出过期错误。

作用域检查是授权：只有令牌未过期且包含所需作用域时，才生成 Authorization 头。

sanitized_headers 独立于 AccessToken，用于在日志中脱敏授权头，防止令牌值泄露。

本阶段的边界决定: 在生成 Authorization 头之前，必须同时满足令牌未过期且包含所需作用域；任何一项不满足都应抛出 PermissionError，而不是返回部分头部或空头部。

实验 ID: mcp-saas-streaming-05-scoped-saas-auth
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/05-scoped-saas-auth/
验收: 待验收

教师示范#

目标: 演示 AccessToken 如何拒绝缺失作用域：创建只读令牌，尝试为 issues:write 生成头部，观察 PermissionError。

假设我们有一个只读令牌，其作用域为 {"issues:read"}，过期时间为 200。

现在时间点为 100，令牌未过期。我们尝试为 issues:write 操作生成头部。

由于 issues:write 不在令牌的作用域集合中，headers_for 应该抛出 PermissionError，而不是返回头部。

实施步骤#

创建 AccessToken("secret", frozenset({"issues:read"}), 200)。
调用 token.headers_for("issues:write", 100)。
观察抛出 PermissionError，且错误消息包含 “scope”。

真实 solution 代码: forge/mcp/auth.py

from dataclasses import dataclass

@dataclass(frozen=True)
class AccessToken:
    value: str
    scopes: frozenset[str]
    expires_at: int

    def headers_for(self, required_scope: str, now: int) -> dict[str, str]:
        if now >= self.expires_at:
            raise PermissionError("access token is expired")
        if required_scope not in self.scopes:
            raise PermissionError("required scope is missing")
        return {"Authorization": f"Bearer {self.value}"}

def sanitized_headers(headers: dict[str, str]) -> dict[str, str]:
    return {key: ("<redacted>" if key.lower() == "authorization" else value) for key, value in headers.items()}

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

在 headers_for 中，先检查 now >= self.expires_at，这里 100 < 200，所以不触发过期错误。

然后检查 required_scope not in self.scopes，因为 issues:write 不在 {"issues:read"} 中，所以抛出 PermissionError。

这个检查防止了只读令牌被用于写操作，实现了最小权限原则。

带提示练习#

在 AccessToken.headers_for 中添加作用域检查，使测试 test_missing_scope_is_rejected 通过。

打开 forge/mcp/auth.py 文件。
在 headers_for 方法中，在过期检查之后添加作用域检查：如果 required_scope not in self.scopes，则抛出 PermissionError("required scope is missing")。
运行测试 python -m pytest -q tests/test_stage.py::test_missing_scope_is_rejected，确认通过。

逐级提示#

作用域检查应该在过期检查之后，因为过期令牌无论如何都不能使用。
使用 not in 操作符检查 required_scope 是否在 self.scopes 中。
错误消息应包含 “scope”，以满足测试断言。

完成标准

test_missing_scope_is_rejected 测试通过。
headers_for 在缺少作用域时抛出 PermissionError，且消息包含 “scope”。

故障注入与排查#

故障夹具 fixtures/failure.json 中注入了一个缺陷：AccessToken.headers_for 缺少作用域检查。运行测试 test_missing_scope_is_rejected 时，测试失败，因为只读令牌成功生成了写操作的授权头。我们需要定位并修复这个缺陷。

本阶段刚刚学习了如何在 headers_for 中添加作用域检查。故障夹具中的缺陷正是缺少这个检查，因此我们可以直接应用所学知识来修复。通过观察测试失败，我们可以确认缺陷位置并添加正确的检查。

已验证的故障合同#

目标测试: test_missing_scope_is_rejected
故障文件: forge/mcp/auth.py

诊断顺序#

运行 python -m pytest -q ../tests/test_stage.py::test_missing_scope_is_rejected，观察测试失败。
查看测试代码：它创建了一个只读令牌，然后尝试为 issues:write 生成头部，期望抛出 PermissionError。
检查 forge/mcp/auth.py 中的 headers_for 方法，发现只有过期检查，没有作用域检查。
确认缺陷：缺少 if required_scope not in self.scopes: raise PermissionError(...)。

修复原则: 在 headers_for 方法中，在过期检查之后添加作用域检查：if required_scope not in self.scopes: raise PermissionError("required scope is missing")。这样，当所需作用域不在令牌作用域中时，会抛出 PermissionError，测试将通过。

python -m pytest -q ../tests/test_stage.py::test_missing_scope_is_rejected

迁移挑战#

在音乐创作场景中，你有一个只读令牌，其作用域为 {"music:read"}，过期时间为 300。现在时间点为 150。你需要验证为 music:write 生成头部时抛出 PermissionError，并检查 sanitized_headers 对 Authorization 的脱敏。

改变的条件: 将作用域从 issues:read 改为 music:read，并将所需操作从 issues:write 改为 music:write，同时调整过期时间和当前时间。

创建一个 AccessToken 实例，其值为 "music-secret"，作用域为 frozenset({"music:read"})，过期时间为 300。然后调用 headers_for("music:write", 150)，确认抛出 PermissionError。接着，使用一个有效的只读令牌（例如作用域为 {"music:read"}）为 music:read 生成头部，并调用 sanitized_headers 检查 Authorization 值是否被替换为 <redacted>。

完成标准

调用 headers_for("music:write", 150) 时抛出 PermissionError，且错误消息包含 “scope”。
对有效头部调用 sanitized_headers 后，返回的字典中 Authorization 键的值为 <redacted>。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么作用域检查必须在过期检查之后？请解释这种顺序如何影响安全性和性能。

一个有力回答应包含#

过期检查是前提：如果令牌已过期，无论作用域如何都不能使用，因此先抛出过期错误可以避免不必要的作用域检查。
如果先检查作用域，可能会为一个过期令牌泄露作用域信息或浪费计算资源。
顺序确保只有在令牌有效时才进行授权判断，符合最小权限原则。

掌握证据

AccessToken.headers_for 在令牌过期或缺少所需作用域时抛出 PermissionError，测试 test_missing_scope_is_rejected 通过。
sanitized_headers 将 Authorization 头的值替换为 <redacted>，测试 test_authorization_is_redacted_from_logs 通过。
在音乐场景的转移挑战中，成功验证了作用域拒绝和头部脱敏。

STAGE 06

只对可安全重试的请求执行退避，并为写请求绑定幂等键

先看一个具体问题#

为什么不能对所有请求都重试？#

在接入外部 SaaS 服务时，网络抖动或服务端瞬时错误很常见，因此我们可能想自动重试失败的请求。

但并非所有请求都适合重试：例如一个创建订单的 POST 请求，如果第一次请求已经成功但响应丢失，重试就会创建两个订单。

本阶段我们要实现一个 plan_request 函数，它根据 HTTP 方法和是否提供幂等键来决定重试策略，从而避免重复副作用。

当前代码中还没有 forge/mcp/saas.py 文件，测试 test_get_retries_are_bounded 和 test_post_retry_requires_idempotency_key 会因导入错误而失败。

先做判断#

在实现 plan_request 之前，请预测：对于 plan_request("POST", 1)（没有提供幂等键），函数应该返回什么？

返回一个 RequestPlan，其中 attempts 为 2
抛出 ValueError，提示需要幂等键
返回一个 RequestPlan，其中 attempts 为 1
返回 None

判断依据: 正确答案是抛出 ValueError。因为 POST 请求可能产生副作用，在没有幂等键的情况下重试是不安全的，所以函数必须拒绝规划重试。

它是什么#

重试策略规划器#

plan_request 是一个纯函数，它根据 HTTP 方法、瞬时失败次数和可选的幂等键，计算出一个 RequestPlan 数据类。

该函数只负责决定“是否重试”以及“重试多少次”，不执行实际的网络请求，也不进行重试操作。

对于 GET 请求，因为它是幂等的，可以安全地根据瞬时失败次数增加尝试次数，但尝试次数有上限。

对于 POST 请求，只有在提供了幂等键时才允许重试，否则抛出 ValueError。

它不是什么#

不是网络客户端或执行器#

plan_request 不发送 HTTP 请求，也不处理响应，它只是生成一个计划对象。

它不负责生成幂等键，也不验证幂等键的格式，只检查其是否存在。

它不是一个通用的重试装饰器，不会自动包裹任意函数进行重试。

它与相邻概念的关系#

与其他组件的关系#

RequestPlan 数据类包含 method、attempts 和 headers 字段，其中 headers 会包含 Idempotency-Key（如果提供了幂等键）。

后续的 SaaS 客户端可以使用 RequestPlan 来实际执行请求，并按照 attempts 进行重试。

plan_request 依赖于调用者传入正确的参数，它本身不维护任何状态。

本阶段的边界决定: 边界决策：plan_request 只负责规划重试策略，不负责实际网络调用或重试执行；对于写请求，必须要求调用者提供幂等键，否则拒绝规划。

实验 ID: mcp-saas-streaming-06-idempotent-saas-retries
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/06-idempotent-saas-retries/
验收: 待验收

教师示范#

目标: 演示 plan_request 如何拒绝无幂等键的 POST 重试：调用 plan_request("POST", 1)，观察 ValueError。

假设我们正在实现 plan_request，首先处理 GET 请求：对于 plan_request("GET", 9)，因为 GET 是幂等的，我们可以根据瞬时失败次数计算尝试次数。

根据公式 attempts = 1 + min(transient_failures, 2)，当 transient_failures=9 时，min(9,2)=2，所以 attempts=3。

对于 POST 请求，如果没有提供幂等键，函数应该抛出 ValueError，因为重试写操作可能导致重复副作用。

实施步骤#

创建 forge/mcp/saas.py 文件，定义 RequestPlan 数据类和 plan_request 函数。
在 plan_request 中，将方法转换为大写，并检查是否为支持的方法（GET 或 POST）。
如果方法是 POST 且没有提供幂等键，抛出 ValueError。
对于 GET 请求，计算尝试次数为 1 + min(transient_failures, 2)。
对于 POST 请求且提供了幂等键，也计算尝试次数，并在 headers 中包含 Idempotency-Key。

真实 solution 代码: forge/mcp/saas.py

from dataclasses import dataclass

@dataclass(frozen=True)
class RequestPlan:
    method: str
    attempts: int
    headers: dict[str, str]

def plan_request(method: str, transient_failures: int, idempotency_key: str | None = None) -> RequestPlan:
    method = method.upper()
    if method not in {"GET", "POST"}:
        raise ValueError("unsupported method")
    if method == "POST" and not idempotency_key:
        raise ValueError("write retries require an idempotency key")
    retryable = method == "GET" or idempotency_key is not None
    attempts = 1 + min(transient_failures, 2) if retryable else 1
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    return RequestPlan(method, attempts, headers)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

这个例子展示了 plan_request 对不安全重试的拒绝：当方法为 POST 且没有幂等键时，函数抛出 ValueError。

通过这种方式，我们强制调用者在规划写请求重试时提供幂等键，从而避免重复副作用。

带提示练习#

在 plan_request 中添加对 POST 请求幂等键的检查，使测试 test_post_retry_requires_idempotency_key 通过。

打开 forge/mcp/saas.py 文件。
在 plan_request 函数中，添加对方法的检查：如果方法不是 GET 或 POST，抛出 ValueError。
添加对 POST 请求的幂等键检查：如果方法是 POST 且 idempotency_key 为 None，抛出 ValueError。
对于 GET 请求，计算尝试次数为 1 + min(transient_failures, 2)。
对于 POST 请求且提供了幂等键，也计算尝试次数，并在 headers 中包含 Idempotency-Key。
运行测试 python -m pytest -q tests/test_stage.py 验证。

逐级提示#

提示 1：使用 if method == "POST" and not idempotency_key: 来检查。
提示 2：尝试次数公式为 1 + min(transient_failures, 2)。
提示 3：headers 字典在提供幂等键时包含 "Idempotency-Key": idempotency_key。

完成标准

plan_request("GET", 9) 返回 attempts == 3。
plan_request("POST", 1) 抛出 ValueError，且错误信息包含 "idempotency"。

故障注入与排查#

在故障实验室中，我们提供了一个有缺陷的 forge/mcp/saas.py 实现，它没有检查 POST 请求的幂等键，导致测试 test_post_retry_requires_idempotency_key 失败。运行 python -m pytest -q ../tests/test_stage.py::test_post_retry_requires_idempotency_key 会看到 AssertionError: unsafe POST retry was planned。

本阶段我们学习了如何区分安全和不安全的请求，并强制要求写请求提供幂等键。这个故障正是由于缺少该检查导致的，因此我们可以通过添加幂等键检查来修复它。

已验证的故障合同#

目标测试: test_post_retry_requires_idempotency_key
故障文件: forge/mcp/saas.py

诊断顺序#

运行测试 python -m pytest -q ../tests/test_stage.py::test_post_retry_requires_idempotency_key，观察失败信息。
查看 forge/mcp/saas.py 中的 plan_request 函数，发现它没有对 POST 请求检查幂等键。
对比预期行为：POST 请求在没有幂等键时应该抛出 ValueError。
定位到缺陷：函数对所有方法都计算了重试次数，没有区分 GET 和 POST。

修复原则: 在 plan_request 函数中添加对 POST 请求的幂等键检查：如果方法是 POST 且 idempotency_key 为 None，抛出 ValueError。同时，对于不支持的方法也抛出 ValueError。

python -m pytest -q ../tests/test_stage.py::test_post_retry_requires_idempotency_key

迁移挑战#

在办公研究场景中，你需要使用 plan_request 规划一个带幂等键的 POST 请求，并验证返回的 RequestPlan 包含 Idempotency-Key 头。同时，尝试不带幂等键时抛出 ValueError。

改变的条件: 将场景从通用 SaaS 调用改为办公研究中的文档上传请求，使用不同的幂等键值。

编写一个测试或脚本，调用 plan_request("POST", 2, idempotency_key="office-upload-123")，断言返回的 RequestPlan 的 attempts 为 3，且 headers 包含 {"Idempotency-Key": "office-upload-123"}。然后调用 plan_request("POST", 2)，断言抛出 ValueError。

完成标准

对于带幂等键的 POST 请求，返回的 RequestPlan 的 attempts 等于 3，且 headers 包含正确的 Idempotency-Key。
对于不带幂等键的 POST 请求，抛出 ValueError，错误信息包含 "idempotency"。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么 GET 请求可以安全重试而 POST 不行？请解释幂等键如何防止重复副作用。

一个有力回答应包含#

GET 请求是幂等的，多次执行不会改变服务器状态，因此可以安全重试。
POST 请求通常用于创建资源或执行操作，可能产生副作用，重复执行会导致重复创建或重复操作。
幂等键是一个唯一标识符，服务器可以用它来识别重复请求，从而只执行一次操作，避免重复副作用。

掌握证据

plan_request 对 GET 请求返回有界尝试次数：plan_request("GET", 9).attempts == 3。
plan_request 对没有幂等键的 POST 请求抛出 ValueError：plan_request("POST", 1) 抛出包含 "idempotency" 的异常。
测试 test_get_retries_are_bounded 和 test_post_retry_requires_idempotency_key 均通过。

STAGE 07

协商客户端与服务端能力，并只从已确认游标恢复兼容会话

先看一个具体问题#

会话恢复时协议版本可能被悄悄改变#

当前 forge/mcp/session.py 文件尚未创建，测试 test_negotiation_uses_intersection 和 test_resume_rejects_protocol_change 在导入阶段就会失败，因为 forge.mcp.session 模块不存在。

即使补上模块，如果 resume 函数无条件信任调用方传入的协议版本，一个原本协商为 2026-01 的会话就可能被恢复成 2025-06，导致后续流处理使用错误的消息格式。

本阶段需要实现 negotiate 和 resume 两个函数，并确保恢复时协议版本必须与已协商会话一致，否则抛出 ValueError。

先做判断#

在实现 resume 之前，你认为以下哪种行为是正确的？

恢复时可以重新协商协议版本，只要客户端支持即可。
恢复时必须使用原会话的协议版本，任何不一致都应拒绝。
恢复时只需检查游标是否大于已确认序列，版本无关紧要。
恢复时应该自动选择客户端和服务端都支持的最新版本。

判断依据: 正确选项是第二个：恢复时必须使用原会话的协议版本。因为会话已经建立，协议版本是双方约定的契约，改变版本意味着消息格式可能不兼容，必须拒绝。

它是什么#

会话协商与恢复的本质#

会话协商（negotiate）是在客户端和服务端之间确定一个共同的协议版本和一组共同支持的能力，并生成一个不可变的 Session 对象。

会话恢复（resume）是在连接中断后，使用已确认的游标继续处理事件，但必须保证协议版本不变且游标不回退。

它不是什么#

会话恢复不是重新协商#

恢复不是重新协商：它不能改变协议版本或能力集合，否则会破坏已建立的通信契约。

恢复也不负责实际流传输或断线重连，它只负责状态检查并返回新的会话对象。

它与相邻概念的关系#

组件之间的关系#

negotiate 产生 Session，resume 消费 Session 并返回更新后的 Session。

Session 的 protocol_version 和 capabilities 在恢复时必须保持不变，只有 acknowledged_sequence 可以更新。

本阶段的边界决定: 恢复时如果协议版本不一致或游标小于已确认序列，必须抛出 ValueError，而不是静默接受或尝试重新协商。

实验 ID: mcp-saas-streaming-07-session-negotiation-resume
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/mcp-saas-streaming/07-session-negotiation-resume/
验收: 待验收

教师示范#

目标: 演示 resume 如何拒绝协议版本变更：创建 2026-01 会话，尝试用 2025-06 恢复，观察 ValueError。

假设我们已经有一个协商好的会话，协议版本为 2026-01，能力为 {"tools"}，已确认序列为 8。

现在连接中断后，客户端尝试用协议版本 2025-06 和游标 9 恢复会话。

正确的实现应该检查协议版本是否一致，发现不一致后抛出 ValueError，并包含 “version” 字样。

实施步骤#

创建 Session("2026-01", frozenset({"tools"}), acknowledged_sequence=8)。
调用 resume(session, "2025-06", 9)。
断言抛出 ValueError 且错误消息包含 “version”。

真实 solution 代码: forge/mcp/session.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Session:
    protocol_version: str
    capabilities: frozenset[str]
    acknowledged_sequence: int = 0

def negotiate(client_versions: tuple[str, ...], server_versions: tuple[str, ...], client_caps: set[str], server_caps: set[str]) -> Session:
    common = [version for version in server_versions if version in client_versions]
    if not common:
        raise ValueError("no compatible protocol version")
    return Session(common[0], frozenset(client_caps & server_caps))

def resume(session: Session, protocol_version: str, cursor: int) -> Session:
    if protocol_version != session.protocol_version:
        raise ValueError("protocol version changed")
    if cursor < session.acknowledged_sequence:
        raise ValueError("resume cursor would replay acknowledged events")
    return Session(session.protocol_version, session.capabilities, cursor)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现#

resume 函数首先比较传入的 protocol_version 与 session.protocol_version，如果不相等则抛出 ValueError("protocol version changed")。

这样测试中的 except ValueError 分支会被触发，断言错误消息包含 “version” 成功。

带提示练习#

在 resume 中添加协议版本一致性检查，使测试 test_resume_rejects_protocol_change 通过。

创建 forge/mcp/session.py 文件。
定义 Session 数据类，包含 protocol_version、capabilities 和 acknowledged_sequence 字段。
实现 negotiate 函数：计算共同版本，若无则抛出 ValueError；返回 Session 对象，能力为交集。
实现 resume 函数：检查协议版本是否一致，若不一致抛出 ValueError；检查游标是否小于已确认序列，若是则抛出 ValueError；否则返回新 Session。

逐级提示#

使用 if protocol_version != session.protocol_version: 进行检查。
错误消息应包含 “version” 以通过测试断言。
游标检查使用 if cursor < session.acknowledged_sequence:。

完成标准

test_negotiation_uses_intersection 通过。
test_resume_rejects_protocol_change 通过。

故障注入与排查#

故障夹具 fixtures/failure.json 中的缺陷是：resume 函数被替换为无条件重建会话，忽略了协议版本检查。当测试 test_resume_rejects_protocol_change 运行时，它期望 resume 对版本不匹配抛出 ValueError，但实际却返回了一个新会话，导致测试失败。

本阶段学习的核心机制就是协议版本一致性检查。通过实现 resume 中的版本比较，可以修复该缺陷，使测试通过。

已验证的故障合同#

目标测试: test_resume_rejects_protocol_change
故障文件: forge/mcp/session.py

诊断顺序#

运行 python -m pytest -q ../tests/test_stage.py::test_resume_rejects_protocol_change，观察测试失败。
查看 forge/mcp/session.py 中的 resume 函数，发现它直接返回 Session(protocol_version, session.capabilities, cursor)，没有版本检查。
对比预期行为：应该检查 protocol_version != session.protocol_version 并抛出 ValueError。

修复原则: 在 resume 函数开头添加版本一致性检查：if protocol_version != session.protocol_version: raise ValueError("protocol version changed")。同时保留游标检查。

python -m pytest -q ../tests/test_stage.py::test_resume_rejects_protocol_change

迁移挑战#

在办公研究场景中，客户端支持协议版本 2026-01 和 2025-06，服务端支持 2026-01；客户端能力为 {"tools", "resources"}，服务端能力为 {"tools", "prompts"}。使用 negotiate 协商后，验证返回的 Session 协议版本为 2026-01，能力为 {"tools"}。然后尝试用 2025-06 恢复该会话，期望抛出 ValueError。

改变的条件: 将客户端和服务端的版本及能力集合改为办公研究场景的具体值，但保持协商逻辑不变。

编写一个测试或脚本，调用 negotiate(("2026-01", "2025-06"), ("2026-01",), {"tools", "resources"}, {"tools", "prompts"})，断言返回的会话协议版本为 2026-01 且能力为 frozenset({"tools"})；然后调用 resume(session, "2025-06", 0)，断言抛出 ValueError。

完成标准

协商结果协议版本为 2026-01，能力为 frozenset({"tools"})。
使用不同版本恢复时抛出 ValueError，且错误消息包含 “version”。

运行真实测试#

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么恢复时必须检查协议版本？如果允许版本变化，会导致什么具体后果？

一个有力回答应包含#

协议版本决定了消息的序列化格式和字段语义，版本不一致会导致解析错误或数据丢失。
恢复时改变版本相当于单方面修改契约，另一方无法正确解释消息，可能引发安全漏洞或状态不一致。

掌握证据

negotiate 返回的 Session 具有共同协议版本和能力交集，测试 test_negotiation_uses_intersection 通过。
resume 在协议版本不匹配或游标回退时抛出 ValueError，测试 test_resume_rejects_protocol_change 通过。
在办公研究场景的转移挑战中，正确协商并拒绝版本不匹配的恢复。

完成本章#

让 Forge 通过 MCP 访问外部服务，三个场景分别接入不同工具。

FORGE / LOCAL CHECK

本地实验自检

未开始
2阅读中
3实验已下载
4测试结果已读取
5本地自检通过

verification.json 只在当前浏览器中解析，不会上传。这里验证的是实验合同，不是服务器认证或第三方背书。

选择 verification.json

主要层

依赖能力

为后续提供

受这些层约束

概念校准

MCP

工具

Harness

MCP Server、SaaS 与流式协议#

先看一个具体问题#

为什么只检查必填字段还不够？#

先做判断#

它是什么#

工具 Schema 是参数形状的合同#

它不是什么#

工具 Schema 不是安全边界#

它与相邻概念的关系#

Schema 与 Harness、Agent Loop 的关系#

先看一个具体问题#

如何防止资源注册越界，确保只能访问允许的 URI 前缀？#

先做判断#

它是什么#

ResourceRegistry 是什么#

它不是什么#

ResourceRegistry 不是什么#

它与相邻概念的关系#

与其他组件的关系#

先看一个具体问题#

缺失变量被静默替换为空字符串，导致提示不完整#

先做判断#

它是什么#

PromptTemplate 是变量合同与渲染器#

它不是什么#

PromptTemplate 不负责提示内容质量或模型调用#

它与相邻概念的关系#

PromptTemplate 与 ResourceRegistry 的关系#

先看一个具体问题#

流式事件必须按序到达且终止后不可继续#

先做判断#

它是什么#

StreamAssembler 是什么#

它不是什么#

StreamAssembler 不是什么#

它与相邻概念的关系#

与其他组件的关系#

先看一个具体问题#

只读令牌被误用于写操作#

先做判断#

它是什么#

AccessToken 是一个最小授权决策单元#

它不是什么#

AccessToken 不负责获取或刷新令牌#

它与相邻概念的关系#

过期检查与作用域检查的顺序关系#

先看一个具体问题#

为什么不能对所有请求都重试？#

先做判断#

它是什么#

重试策略规划器#

它不是什么#

不是网络客户端或执行器#

它与相邻概念的关系#

与其他组件的关系#

先看一个具体问题#

会话恢复时协议版本可能被悄悄改变#

先做判断#

它是什么#

会话协商与恢复的本质#

它不是什么#

会话恢复不是重新协商#

它与相邻概念的关系#

组件之间的关系#