MCP 表单回传之后：处理 accept、decline、cancel 与缺字段
用十二个真实 stdio 案例检查补充输入：动作决定是否继续，字段验证决定能否计算。
- 拒绝与取消直接结束，不使用夹带内容继续计算
- accept 仍需要字段验证，也不等于外部操作授权
- 区分低层结果容器与高层模型校验
- 能力守卫应阻止不受支持的请求
- 用独立预期和正常退出记录验收真实 stdio
1. 一声 accept 之后，服务端还欠哪些检查
一个工具需要两个整数才能完成计算。客户端已经发出了 tools/call，服务端却还不知道 units 和 unit_value。MCP 的 form elicitation 允许服务端在这次调用中发起一份结构化补充输入请求；客户端返回动作和内容后，外层工具调用继续完成。真正容易写错的地方在回传之后：只检查 action 为 accept，就直接读取字段。
本文把问题缩小到一项虚构整数乘法。两个字段分别限定为 1～10 和 1～100，合法输入 3、7 的结果是 21。没有真实账户、消息发送、支付、资料保存或其他业务写操作。测试程序会在本地写合成协议记录，这些测试文件不代表任何外部行为已经获准。
所有选择由固定客户端回调模拟，没有真人点击，也没有模型判断。accept 仅表示这份表单被接受并回传输入；它不授予外部操作权限。拒绝和取消都结束当前调用，不自动重试，也不重新询问。
实测日期为 2026-10-07，使用环境中已有的 Python 3.12.14、MCP Python SDK 1.29.0、Pydantic 2.13.4 和 AnyIO 4.14.2。每案都启动真实 stdio 子进程，协商到的协议版本是 2025-11-25。本文没有验证最新 SDK、2026-07-28 协议或任何宿主界面；运行时不会安装依赖或访问 HTTP。
2. 把内层请求和外层结果分开读
在这次实测中，客户端以 id=1 发起 tools/call，服务端随后以自己这一方向的 id=0 发起 elicitation/create。客户端必须先回复内层请求，服务端才能把计算结果作为 id=1 的工具结果返回。请求 ID 按请求方向区分，不能看到数字相同就把两种方向的会话关联混在一起。下面的图省略了 initialize 和 SDK 自动进行的工具列表查询，保留决定结果的四条消息。
form 请求携带 mode、message 和 requestedSchema；回复的 action 是 accept、decline 或 cancel。decline 表示明确拒绝该请求，cancel 表示未作明确选择便取消了交互。依照2025-11-25 版 elicitation 规范，content 随接受动作返回，拒绝与取消通常省略 content。实验额外注入两份仍附带合法 content 的拒绝或取消响应，用来检查服务端能否守住停止分支；即使内容可用，本例也不继续计算。
客户端 → 服务端 tools/call, id=1
服务端 → 客户端 elicitation/create, id=0
客户端 → 服务端 {"id":0,"result":{"action":"accept","content":{"units":3,"unit_value":7}}}
服务端 → 客户端 工具结果：{"status":"calculated","computed":true,"value":21}, id=1这里展示的是便于阅读的消息摘要，不是可直接发送的完整 JSON-RPC 数据。运行后 results.json 中的 wire 保存实际完整消息，包括 jsonrpc、请求参数和响应。客户端的 initialize 能力也记录在那里。
3. 先让能力与动作决定是否继续
服务端先检查客户端有没有声明 form 支持。没有声明时，正常路径立即返回 unsupported，而且不发送 elicitation/create。这比等待一个“不支持”的错误更清楚：调用者没有承诺能够处理表单，就不应该收到表单请求。
这个固定 SDK 的 ClientSession 在提供回调时会同时声明 form 和 url，本例回调只处理 form。它只是受控实验客户端，不能直接当作完整支持 URL 模式的生产客户端。代码也兼容规范中的空对象 legacy form 声明，并拒绝仅声明 url 的组合，但这两种能力组合没有在本实验单独启动 stdio 案例，不计入已测结果。
服务端模型显式开启 strict=True，所以字符串 "3" 不会被转换成整数；extra="forbid" 禁止多余字段。字段缺失、类型不符和范围越界都会成为 invalid_content。多余字段策略虽写在模型中，当前 12 案没有单独测试它，不能把配置声明当成测试覆盖。
4. 服务端：只把验证通过的数据送进乘法
在一个空目录中按文中的相对路径保存文件。source/server.py 注册一个 calculate 工具，默认使用高层 ctx.elicit。status、computed 和 value 是本例工具自定义的业务结果字段，不属于表单回传的 action/content 契约。raw 和 diagnostic_unchecked 只用于后文两个隔离诊断案例，永远不会执行乘法。不要把这个带诊断参数的实验服务直接开放给真实用户。
source/server.py · 完整文件
"""Synthetic pure calculation; no user data, files, purchases or messages."""
from mcp.server.fastmcp import FastMCP, Context
from mcp.shared.exceptions import McpError
from pydantic import BaseModel, ConfigDict, Field, ValidationError
app = FastMCP('synthetic-elicitation-lab')
class Form(BaseModel):
model_config = ConfigDict(strict=True, extra='forbid')
units: int = Field(ge=1, le=10)
unit_value: int = Field(ge=1, le=100)
@app.tool()
async def calculate(ctx: Context, policy: str = 'validated') -> dict:
caps = ctx.session.client_params.capabilities.elicitation
# In this pinned SDK, an empty elicitation capability is legacy form support.
supports_form = caps is not None and (caps.form is not None or caps.url is None)
if not supports_form and policy != 'diagnostic_unchecked':
return {'status': 'unsupported', 'computed': False, 'requested': False}
try:
if policy in ('raw', 'diagnostic_unchecked'):
result = await ctx.session.elicit_form('Synthetic arithmetic inputs only', Form.model_json_schema())
# Diagnostic path deliberately does not calculate from unvalidated content.
return {'status': 'raw_response', 'computed': False, 'action': result.action, 'content': result.content}
result = await ctx.elicit('Synthetic arithmetic inputs only', Form)
except ValidationError as exc:
return {'status': 'invalid_content', 'computed': False, 'errors': [list(e['loc']) for e in exc.errors()]}
except ValueError as exc:
return {'status': 'invalid_response', 'computed': False, 'error': str(exc)}
except McpError as exc:
return {'status': 'protocol_error', 'computed': False, 'code': exc.error.code, 'error': exc.error.message}
if result.action == 'decline':
return {'status': 'declined', 'computed': False}
if result.action == 'cancel':
return {'status': 'cancelled', 'computed': False}
# accept means inputs were returned, never permission for external effects.
return {'status': 'calculated', 'computed': True, 'value': result.data.units * result.data.unit_value}
if __name__ == '__main__':
app.run(transport='stdio')
高层 helper 会在 accept 且存在 content 时用 Form 验证字段，再把模型实例放到 result.data 中。decline 和 cancel 分别形成自己的返回类型。业务代码为这两种动作直接 return，所以即使响应夹带了两个合法整数，也不会落入最后的乘法。
accept 没有 content 是另一个边界。SDK 1.29.0 的高层 helper 在这里抛 ValueError，消息是 Unexpected elicitation action: accept。本例把它归为 invalid_response。错误文案容易令人误会：合法动作 accept 本身没有被协议禁止，出问题的是缺少接受分支所需内容的响应组合。
5. 低层返回容器，不替你完成字段验证
给 session.elicit_form 传入 requestedSchema，能把结构要求送到客户端。但在本次固定版本实测中，accept 加 {"units":3} 仍成功返回 ElicitResult。低层方法验证协议结果容器，没有替应用按这份 schema 验证缺少的 unit_value。高层 ctx.elicit 则会用传入的 Pydantic 模型检查它。需要查看 API 实现时，可对照SDK v1.29.0 的 ServerSession。
诊断路径把原始动作和内容显示为 raw_response，同时固定 computed=false。它的用途是让读者观察验证在哪一层发生。若应用自行调用低层 API，应自行建立严格模型验证并明确处理动作分支；不能把 content 已经是字典当作字段都有效的证明。
另一个诊断案例模拟完全没有 elicitation capability 的客户端，并故意略过服务端守卫。该客户端没有自定义回调，SDK 默认处理器实际返回 -32600、Elicitation not supported。固定版规范要求未声明相应模式的客户端返回 -32602，因此这里记录的是 SDK 1.29.0 的实测防御行为，不把它写成规范规定，更不建议真实服务忽略能力声明。正常无能力案例仍是零请求并立即结束。
6. 客户端：用固定夹具回答，记录真实管道消息
source/client.py 用当前 sys.executable 启动独立服务端进程，再把其 stdin/stdout 接到 SDK ClientSession。每一行 stdout 都按 JSON-RPC 消息解析，任何普通打印或不完整尾行都会记为错误。服务端日志单独从 stderr 读取；客户端的测试摘要是父进程输出，不会混入服务端协议管道。
source/client.py · 完整文件
"""Pinned SDK session over a real local subprocess. All input is synthetic."""
import json
import sys
from contextlib import asynccontextmanager
from datetime import timedelta
from pathlib import Path
import anyio
from anyio.streams.text import TextReceiveStream
from mcp import ClientSession, types
from mcp.shared.message import SessionMessage
@asynccontextmanager
async def transport(server, evidence):
incoming_send, incoming_receive = anyio.create_memory_object_stream(0)
outgoing_send, outgoing_receive = anyio.create_memory_object_stream(0)
proc = await anyio.open_process([sys.executable, str(server)], cwd=server.parent)
evidence['pid'] = proc.pid
evidence['forced_kill'] = False
evidence['wire'] = []
evidence['stdout_errors'] = []
evidence['stderr'] = ''
stdout_done, stderr_done = anyio.Event(), anyio.Event()
async def read_stdout():
buffer = ''
async for chunk in TextReceiveStream(proc.stdout):
buffer += chunk
while '\n' in buffer:
line, buffer = buffer.split('\n', 1)
try:
message = types.JSONRPCMessage.model_validate_json(line)
except Exception as exc:
evidence['stdout_errors'].append(str(exc))
await incoming_send.send(exc)
else:
evidence['wire'].append({'from': 'server', 'message': json.loads(line)})
await incoming_send.send(SessionMessage(message))
if buffer:
evidence['stdout_errors'].append('Incomplete stdout frame: ' + buffer)
stdout_done.set()
async def write_stdin():
async for item in outgoing_receive:
line = item.message.model_dump_json(by_alias=True, exclude_none=True)
evidence['wire'].append({'from': 'client', 'message': json.loads(line)})
await proc.stdin.send((line + '\n').encode())
async def read_stderr():
async for chunk in TextReceiveStream(proc.stderr):
evidence['stderr'] += chunk
stderr_done.set()
async with proc, anyio.create_task_group() as tasks:
tasks.start_soon(read_stdout)
tasks.start_soon(write_stdin)
tasks.start_soon(read_stderr)
try:
yield incoming_receive, outgoing_send
finally:
# Close only this child's stdin; wait before killing this child.
with anyio.CancelScope(shield=True):
await proc.stdin.aclose()
with anyio.move_on_after(5) as timeout:
await proc.wait()
if timeout.cancel_called:
evidence['forced_kill'] = True
proc.kill()
await proc.wait()
evidence['returncode'] = proc.returncode
with anyio.move_on_after(1) as drain:
await stdout_done.wait()
await stderr_done.wait()
evidence['stream_drain_timeout'] = drain.cancel_called
tasks.cancel_scope.cancel()
for stream in (incoming_send, incoming_receive, outgoing_send, outgoing_receive):
await stream.aclose()
async def run_case(case, server):
evidence = {'case': case['name'], 'callback_count': 0}
async def simulated_user(context, params):
# Fixed fixture selection, not a human approval or a model decision.
evidence['callback_count'] += 1
if not isinstance(params, types.ElicitRequestFormParams):
raise ValueError('This laboratory only handles form mode')
evidence['form_request'] = params.model_dump(mode='json', exclude_none=True)
return types.ElicitResult.model_validate(case['response'])
callback = simulated_user if case['response'] is not None else None
async with transport(server, evidence) as (reader, writer):
async with ClientSession(reader, writer,
read_timeout_seconds=timedelta(seconds=10),
elicitation_callback=callback) as session:
initialized = await session.initialize()
evidence['protocol_version'] = initialized.protocolVersion
result = await session.call_tool('calculate', {'policy': case['policy']})
if result.isError:
raise RuntimeError(result.content)
evidence['result'] = json.loads(result.content[0].text)
return evidence
模拟回调只返回 fixture 中预先写好的动作与内容，不修改输入来争取通过。没有 response 的两个案例不配置回调，也就没有声明 elicitation。客户端保留握手、内层请求、固定回复和工具结果，测试可以核对请求方向以及每案到底询问了几次。
transport 的 finally 只关闭本次创建子进程的 stdin，给它五秒正常退出时间。逾时才终止这个子进程并记下 forced_kill=true，不扫描或终止其他进程。正常验收要求退出码为 0 且没有强制终止。十秒响应上限是防止测试卡住的保护措施，本次没有模拟超时或断线，不把这段保护代码算作相应场景已通过。
7. 十二份输入与独立预期
fixtures/cases.json 只保存输入。expected/results.json 是独立写定的结果，不从当前服务端输出自动生成；否则实现把计算分支写错时，预期也可能跟着变化。比较完整结果，能发现“状态看起来对，但 computed 或 value 错了”的情况。
fixtures/cases.json · 完整文件
[
{
"name": "accept_valid",
"policy": "validated",
"response": {
"action": "accept",
"content": {
"units": 3,
"unit_value": 7
}
}
},
{
"name": "accept_missing_field",
"policy": "validated",
"response": {
"action": "accept",
"content": {
"units": 3
}
}
},
{
"name": "accept_no_content",
"policy": "validated",
"response": {
"action": "accept"
}
},
{
"name": "accept_wrong_type",
"policy": "validated",
"response": {
"action": "accept",
"content": {
"units": "3",
"unit_value": 7
}
}
},
{
"name": "accept_out_of_range",
"policy": "validated",
"response": {
"action": "accept",
"content": {
"units": 11,
"unit_value": 7
}
}
},
{
"name": "decline",
"policy": "validated",
"response": {
"action": "decline"
}
},
{
"name": "cancel",
"policy": "validated",
"response": {
"action": "cancel"
}
},
{
"name": "decline_with_spurious_data",
"policy": "validated",
"response": {
"action": "decline",
"content": {
"units": 3,
"unit_value": 7
}
}
},
{
"name": "cancel_with_spurious_data",
"policy": "validated",
"response": {
"action": "cancel",
"content": {
"units": 3,
"unit_value": 7
}
}
},
{
"name": "raw_accept_missing_field",
"policy": "raw",
"response": {
"action": "accept",
"content": {
"units": 3
}
}
},
{
"name": "unsupported_guarded",
"policy": "validated",
"response": null
},
{
"name": "unsupported_unchecked_diagnostic",
"policy": "diagnostic_unchecked",
"response": null
}
]
expected/results.json · 完整文件
{
"accept_valid": {
"status": "calculated",
"computed": true,
"value": 21
},
"accept_missing_field": {
"status": "invalid_content",
"computed": false,
"errors": [
[
"unit_value"
]
]
},
"accept_no_content": {
"status": "invalid_response",
"computed": false,
"error": "Unexpected elicitation action: accept"
},
"accept_wrong_type": {
"status": "invalid_content",
"computed": false,
"errors": [
[
"units"
]
]
},
"accept_out_of_range": {
"status": "invalid_content",
"computed": false,
"errors": [
[
"units"
]
]
},
"decline": {
"status": "declined",
"computed": false
},
"cancel": {
"status": "cancelled",
"computed": false
},
"decline_with_spurious_data": {
"status": "declined",
"computed": false
},
"cancel_with_spurious_data": {
"status": "cancelled",
"computed": false
},
"raw_accept_missing_field": {
"status": "raw_response",
"computed": false,
"action": "accept",
"content": {
"units": 3
}
},
"unsupported_guarded": {
"status": "unsupported",
"computed": false,
"requested": false
},
"unsupported_unchecked_diagnostic": {
"status": "protocol_error",
"computed": false,
"code": -32600,
"error": "Elicitation not supported"
}
}
只有 accept_valid 的 computed 为 true，结果必须是 21。正常拒绝、取消以及两份夹带数据的响应都要求 computed=false。回调次数同时受断言约束：有固定回调的案例必须恰好一次，无回调案例必须零次。检查线上 elicitation/create 消息后，能进一步区别“没发请求”和“发了请求但由 SDK 默认处理器拒绝”，避免仅凭回调次数下结论。
8. 测试：连请求数量、Schema 和退出状态一起验收
2026-10-20 补测发现了一种假通过：把服务端每个 computed 的 true/false 改成整数 1/0，原来的十二案仍全部通过。原因是 Python 的 True == 1、False == 0，字典的 == 也沿用元素相等规则。比较完整对象不等于比较字段类型。以下修改只收紧测试判据，source/server.py 和两份独立预期均不变。
先保存共用比较器，再保存测试文件。它递归比较本实验 JSON 解码后的值，同时要求类型、对象键、列表长度与值相同；用显式异常保留 python -O 下的类型检查。它不是通用 JSON Schema 验证器，也不是 MCP 要求所有数字具有相同词法表示：这里刻意按固定夹具区分 bool/int 和 int/float。Python 的数值比较规则可核对官方内置类型文档。
source/result_contract.py · 完整文件
"""Type-aware comparison for this lab's JSON-decoded, fixed fixtures."""
def require_same_json(actual, expected, path='$'):
if type(actual) is not type(expected):
raise AssertionError(f'{path}: {type(actual).__name__} != {type(expected).__name__}')
if isinstance(expected, dict):
if actual.keys() != expected.keys():
raise AssertionError(f'{path}: object keys differ')
for key in expected:
require_same_json(actual[key], expected[key], f'{path}.{key}')
elif isinstance(expected, list):
if len(actual) != len(expected):
raise AssertionError(f'{path}: list lengths differ')
for index, (left, right) in enumerate(zip(actual, expected)):
require_same_json(left, right, f'{path}[{index}]')
elif actual != expected:
raise AssertionError(f'{path}: values differ: {actual!r} != {expected!r}')
tests/test_contract.py · 完整文件
"""Independent expectations; never regenerate them from observed output."""
import importlib.metadata
import json
import platform
import shutil
import sys
import tempfile
from pathlib import Path
import anyio
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'source'))
from client import run_case
from result_contract import require_same_json
async def main():
versions = {'python': platform.python_version(),
'mcp': importlib.metadata.version('mcp'),
'pydantic': importlib.metadata.version('pydantic'),
'anyio': importlib.metadata.version('anyio')}
assert versions == {'python': '3.12.14', 'mcp': '1.29.0',
'pydantic': '2.13.4', 'anyio': '4.14.2'}, versions
fixtures = json.loads((ROOT / 'fixtures/cases.json').read_text())
expected = json.loads((ROOT / 'expected/results.json').read_text())
assert len(fixtures) == 12
assert {c['name'] for c in fixtures} == set(expected)
rows = []
with tempfile.TemporaryDirectory(prefix='form-contract-') as tmp:
server = Path(tmp) / 'server.py'
shutil.copyfile(ROOT / 'source/server.py', server)
for case in fixtures:
row = await run_case(case, server)
rows.append(row)
assert row['protocol_version'] == '2025-11-25', row
require_same_json(row['result'], expected[case['name']])
assert row['returncode'] == 0 and not row['forced_kill'], row
assert row['stdout_errors'] == [] and not row['stream_drain_timeout'], row
wire = row['wire']
calls = [i for i, x in enumerate(wire) if x['from'] == 'client'
and x['message'].get('method') == 'tools/call']
assert len(calls) == 1, row
call_id = wire[calls[0]]['message']['id']
completions = [i for i, x in enumerate(wire) if x['from'] == 'server'
and x['message'].get('id') == call_id
and 'result' in x['message']]
assert len(completions) == 1, row
requests = [x['message'] for x in row['wire'] if x['from'] == 'server'
and x['message'].get('method') == 'elicitation/create']
count = 0 if case['name'] == 'unsupported_guarded' else 1
assert len(requests) == count, row
assert row['callback_count'] == (0 if case['response'] is None else 1), row
if requests:
request = requests[0]
position = next(i for i, x in enumerate(wire)
if x['from'] == 'server' and x['message'] == request)
assert calls[0] < position < completions[0], row
assert request['params']['mode'] == 'form', request
schema = request['params']['requestedSchema']
assert schema['required'] == ['units', 'unit_value'], schema
assert schema['additionalProperties'] is False, schema
assert schema['properties']['units']['type'] == 'integer', schema
assert schema['properties']['units']['minimum'] == 1, schema
assert schema['properties']['units']['maximum'] == 10, schema
assert schema['properties']['unit_value']['type'] == 'integer', schema
assert schema['properties']['unit_value']['minimum'] == 1, schema
assert schema['properties']['unit_value']['maximum'] == 100, schema
replies = [x['message'] for x in row['wire'] if x['from'] == 'client'
and x['message'].get('id') == request['id']
and ('result' in x['message'] or 'error' in x['message'])]
assert len(replies) == 1, row
if case['response'] is not None:
assert replies[0]['result'] == case['response'], replies
else:
assert replies[0]['error']['code'] == -32600, replies
out = ROOT / 'run-output'
out.mkdir(exist_ok=True)
(out / 'results.json').write_text(json.dumps({'versions': versions, 'cases': rows},
ensure_ascii=False, indent=2) + '\n')
print(json.dumps({'passed': len(rows), 'protocol': '2025-11-25',
'normal_exits': sum(r['returncode'] == 0 for r in rows),
'forced_kills': sum(r['forced_kill'] for r in rows),
'stdout_errors': sum(len(r['stdout_errors']) for r in rows)}, indent=2))
if __name__ == '__main__':
anyio.run(main)
每案使用一个全新服务端子进程。服务端源文件复制到自动清理的合成临时目录，客户端不读取用户文件或账户。测试先核对固定依赖版本，版本不同就停止，避免把环境差异造成的结果误报为本文已经验证。正常完成后，run-output/results.json 保留每案 PID、退出码、完整合成协议消息和 stderr。PID 每次会改变，不能把某个 PID 数字当作验收条件。
这次环境的 stderr 中包含 Pydantic settings 关于 lifespan 前向引用的警告，以及 SDK 的请求处理日志；它们没有进入 stdout。测试并不要求 stderr 为空，而是严格检查协议 stdout 与正常退出。警告的存在也是固定环境的实际情况，不能省略成“全程没有日志”。
下面再检查测试本身：先验收十二份真实运行结果，然后逐份把 computed 换成等值整数。旧式 == 必须仍认为它们相等，而新比较器必须全部拒绝。这个回放脚本不启动服务端，不把固定回调或测试数据当作真人授权。
test_result_types.py · 完整文件
"""Replay fixed JSON evidence; reject bool/int drift even under python -O."""
import copy
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT / 'source'))
from result_contract import require_same_json
rows = json.loads((ROOT / 'run-output/results.json').read_text())['cases']
expected = json.loads((ROOT / 'expected/results.json').read_text())
if len(rows) != 12 or {r['case'] for r in rows} != set(expected):
raise AssertionError('Missing or duplicate contract cases')
rejected = 0
for row in rows:
original = row['result']
golden = expected[row['case']]
require_same_json(original, golden)
mutant = copy.deepcopy(original)
mutant['computed'] = int(mutant['computed'])
# Establish the old comparator's false positive, not merely a differing value.
if mutant != golden:
raise AssertionError('Fixture does not reproduce the bool/int equality gap')
try:
require_same_json(mutant, golden)
except AssertionError:
rejected += 1
else:
raise RuntimeError('Integer computed flag was accepted')
if rejected != 12:
raise AssertionError(rejected)
print(json.dumps({'valid_results': len(rows), 'rejected_integer_flags': rejected}))
python tests/test_contract.py
python test_result_types.py
python -O test_result_types.py2026-10-20 实测：原服务端十二案通过，十二个整数假冒布尔值的污染在普通模式与 -O 下都被拒绝。另将服务端所有 computed 改为 0/1 的真实 stdio 负例，修复后的测试在两种模式下均以 $.computed: int != bool 失败。只用 -O 直接运行旧测试会移除其他 assert；需要完整优化模式验收时，使用后文的 verify_all.py。
9. 在已有依赖的空目录复跑
source/requirements.txt · 完整文件
# Observed versions only; no installation or upgrade was performed.
# Python 3.12.14
mcp==1.29.0
pydantic==2.13.4
anyio==4.14.2
requirements.txt 只是本次直接依赖版本记录。本教程不执行安装或升级，也没有做干净环境安装验证。请在已经具备这些版本的 Python 环境中运行；缺依赖或版本不同，就先停下核对环境，不要把未运行的结果写成通过。无需浏览器、模型、远程 HTTP 或密钥。
python tests/test_contract.py本地最终版本连续两轮各 12 案通过。每轮摘要如下，完整结果在运行生成的 run-output/results.json 中。
{
"passed": 12,
"protocol": "2025-11-25",
"normal_exits": 12,
"forced_kills": 0,
"stdout_errors": 0
}正文已经给出复跑所需的全部源文件、输入夹具、独立预期和命令。三张 SVG 为本文绘制，内嵌到文章中即可阅读，不依赖外部图片服务。
10. 这份结果可以证明到哪里
这些记录证明固定 SDK 的真实 stdio 请求可以在工具调用中来回完成，并且本例按动作、字段和能力分别限制了纯计算分支。它没有证明某款宿主会怎样展示表单，也没有证明真实用户理解了提示，或模型能正确决定何时询问。
当前没有覆盖 URL 模式、敏感数据表单、远程 HTTP、并发请求、跨会话串线、超时取消、断线恢复、畸形 JSON、超大输入或所有 schema 子集。此处的 cancel 是表单结果动作，不是取消协议通知的演示。面向真实业务扩展时，仍要单独检查调用身份、权限和业务规则；外部操作的授权无法从本例的 accept 或 computed=true 推导出来。
以后修改服务端时，先保持十二份独立预期不动再复跑。若确实改变了契约，应明确说明改变的是动作处理、字段约束还是能力判断，然后单独审阅预期更新。这样失败输出才能指出语义变化，而不是仅仅告诉你程序还能启动。
11. 补充实验：表单回调为什么不该等待同会话的元数据请求
原来的固定回调可以直接返回夹具，不需要再访问服务端。本节只检查一种扩展：为了显示工具资料，回调先在同一个 ClientSession 内 await tools/list，再决定怎样返回表单。此处仍没有真人、模型、账户或业务副作用；accept 与 cancel 都由合成测试明确选定。
在本文固定的 MCP Python SDK 1.29.0 普通回调路径中，接收循环会等待回调返回。回调内发出的请求又要等同一个循环派发结果，因此形成相互等待。BaseSession 的接收循环与ClientSession 的回调入口可用于核对这条路径。这个结论不推广到更新 SDK、其他语言或 task-augmented 回调。
测试没有仅凭“超时”猜测原因。它在真实子进程 stdout 收到完整结果时，先记录一个不包含 await 的观察事件；再把消息交给 SDK。对嵌套 tools/list 的确切请求 ID，断言其响应到达事件早于回调里的超时事件。这样可以区分“服务端尚未回答”和“回答已到达，但当前等待链无法派发”。请求 ID 的对应规则见固定版协议基础规范。
三个场景分别是：nested 在回调内请求元数据，超时后明确返回合成 cancel；prefetch 在 tools/call 之前读好工具名称，回调只读本地快照；fixed_fixture 保留原来的纯固定回调作为阴性对照。每个场景随后在同一个会话中再发一次调用，第二次固定返回 accept，检查能否得到 21。prefetch 只准备元数据，不代表任何用户已经批准操作，也不证明缓存中的权限仍有效。
请先按正文保存原有全部文件，再补充以下五个文件。observed_client.py 是原客户端的一份完整观察副本，增加消息到达事件，并为强制终止后的等待加上两秒上限；没有改变 SDK 路由。主测试在失败时也保存已收集的记录。独立预期是人工写定的常量，不根据本次输出生成。
source/observed_client.py
"""Pinned SDK session over a real local subprocess. All input is synthetic."""
import json
import sys
from contextlib import asynccontextmanager
from datetime import timedelta
from pathlib import Path
import anyio
from anyio.streams.text import TextReceiveStream
from mcp import ClientSession, types
from mcp.shared.message import SessionMessage
@asynccontextmanager
async def transport(server, evidence):
incoming_send, incoming_receive = anyio.create_memory_object_stream(0)
outgoing_send, outgoing_receive = anyio.create_memory_object_stream(0)
proc = await anyio.open_process([sys.executable, str(server)], cwd=server.parent)
evidence['pid'] = proc.pid
evidence['forced_kill'] = False
evidence['wire'] = []
evidence['stdout_errors'] = []
evidence['stderr'] = ''
stdout_done, stderr_done = anyio.Event(), anyio.Event()
async def read_stdout():
buffer = ''
async for chunk in TextReceiveStream(proc.stdout):
buffer += chunk
while '\n' in buffer:
line, buffer = buffer.split('\n', 1)
try:
message = types.JSONRPCMessage.model_validate_json(line)
except Exception as exc:
evidence['stdout_errors'].append(str(exc))
await incoming_send.send(exc)
else:
evidence['wire'].append({'from': 'server', 'message': json.loads(line)})
evidence.setdefault('timeline', []).append({'event': 'server_frame', 'wire_index': len(evidence['wire']) - 1})
await incoming_send.send(SessionMessage(message))
if buffer:
evidence['stdout_errors'].append('Incomplete stdout frame: ' + buffer)
stdout_done.set()
async def write_stdin():
async for item in outgoing_receive:
line = item.message.model_dump_json(by_alias=True, exclude_none=True)
evidence['wire'].append({'from': 'client', 'message': json.loads(line)})
await proc.stdin.send((line + '\n').encode())
async def read_stderr():
async for chunk in TextReceiveStream(proc.stderr):
evidence['stderr'] += chunk
stderr_done.set()
async with proc, anyio.create_task_group() as tasks:
tasks.start_soon(read_stdout)
tasks.start_soon(write_stdin)
tasks.start_soon(read_stderr)
try:
yield incoming_receive, outgoing_send
finally:
# Close only this child's stdin; wait before killing this child.
with anyio.CancelScope(shield=True):
await proc.stdin.aclose()
with anyio.move_on_after(5) as timeout:
await proc.wait()
if timeout.cancel_called:
evidence['forced_kill'] = True
proc.kill()
with anyio.fail_after(2):
await proc.wait()
evidence['returncode'] = proc.returncode
with anyio.move_on_after(1) as drain:
await stdout_done.wait()
await stderr_done.wait()
evidence['stream_drain_timeout'] = drain.cancel_called
tasks.cancel_scope.cancel()
for stream in (incoming_send, incoming_receive, outgoing_send, outgoing_receive):
await stream.aclose()
async def run_case(case, server):
evidence = {'case': case['name'], 'callback_count': 0}
async def simulated_user(context, params):
# Fixed fixture selection, not a human approval or a model decision.
evidence['callback_count'] += 1
if not isinstance(params, types.ElicitRequestFormParams):
raise ValueError('This laboratory only handles form mode')
evidence['form_request'] = params.model_dump(mode='json', exclude_none=True)
return types.ElicitResult.model_validate(case['response'])
callback = simulated_user if case['response'] is not None else None
async with transport(server, evidence) as (reader, writer):
async with ClientSession(reader, writer,
read_timeout_seconds=timedelta(seconds=10),
elicitation_callback=callback) as session:
initialized = await session.initialize()
evidence['protocol_version'] = initialized.protocolVersion
result = await session.call_tool('calculate', {'policy': case['policy']})
if result.isError:
raise RuntimeError(result.content)
evidence['result'] = json.loads(result.content[0].text)
return evidence
expected/reentrancy.json
{
"nested": {"first":{"status":"cancelled","computed":false},"recovery":{"status":"calculated","computed":true,"value":21},"list_calls":2,"actions":["cancel","accept"],"nested_error_code":408},
"prefetch": {"first":{"status":"calculated","computed":true,"value":21},"recovery":{"status":"calculated","computed":true,"value":21},"list_calls":1,"actions":["accept","accept"]},
"fixed_fixture": {"first":{"status":"calculated","computed":true,"value":21},"recovery":{"status":"calculated","computed":true,"value":21},"list_calls":1,"actions":["accept","accept"]}
}
test_reentrancy.py
"""SDK 1.29.0 callback reentrancy probe; synthetic inputs, no HTTP or models."""
import sys, json, importlib.metadata, platform
from pathlib import Path
from datetime import timedelta
import anyio
from mcp import ClientSession, types
from mcp.shared.exceptions import McpError
P = Path(__file__).parent
sys.path.insert(0, str(P / 'source'))
from observed_client import transport
from result_contract import require_same_json
async def scenario(mode, evidence):
cached = None
callbacks = 0
async def callback(context, params):
nonlocal callbacks
callbacks += 1
if not isinstance(params, types.ElicitRequestFormParams):
raise AssertionError('Contract check failed')
evidence['timeline'].append({'event': 'callback_enter', 'number': callbacks})
if mode == 'nested' and callbacks == 1:
try:
await context.session.send_request(types.ClientRequest(types.ListToolsRequest(method='tools/list')), types.ListToolsResult, request_read_timeout_seconds=timedelta(seconds=0.5))
except McpError as exc:
evidence['nested_error_code'] = exc.error.code
evidence['timeline'].append({'event': 'nested_timeout'})
return types.ElicitResult(action='cancel')
raise AssertionError('Nested call unexpectedly completed; recheck SDK behavior')
if mode == 'prefetch':
if not (cached is not None and 'calculate' in cached):
raise AssertionError('Contract check failed')
return types.ElicitResult(action='accept', content={'units': 3, 'unit_value': 7})
async with transport(P / 'source/server.py', evidence) as (reader, writer):
async with ClientSession(reader, writer, read_timeout_seconds=timedelta(seconds=10), elicitation_callback=callback) as session:
evidence['protocol'] = (await session.initialize()).protocolVersion
cached = tuple((t.name for t in (await session.list_tools()).tools))
for name in ('first', 'recovery'):
result = await session.call_tool('calculate', {'policy': 'validated'}, read_timeout_seconds=timedelta(seconds=3))
if not not result.isError:
raise AssertionError('Contract check failed')
evidence[name] = json.loads(result.content[0].text)
evidence['callbacks'] = callbacks
return evidence
def verify(row, expected):
for k in ('first', 'recovery'):
require_same_json(row[k], expected[k], '$.' + k)
if not row['protocol'] == '2025-11-25':
raise AssertionError('Contract check failed')
if not row['callbacks'] == 2:
raise AssertionError('Contract check failed')
if not (row['returncode'] == 0 and (not row['forced_kill'])):
raise AssertionError('Contract check failed')
if not (not row['stdout_errors'] and (not row['stream_drain_timeout'])):
raise AssertionError('Contract check failed')
wire = row['wire']
lists = [x['message'] for x in wire if x['from'] == 'client' and x['message'].get('method') == 'tools/list']
if not len(lists) == expected['list_calls']:
raise AssertionError('Contract check failed')
elic = [x['message'] for x in wire if x['from'] == 'server' and x['message'].get('method') == 'elicitation/create']
if not len(elic) == 2:
raise AssertionError('Contract check failed')
replies = [x['message'] for x in wire if x['from'] == 'client' and 'result' in x['message']]
if not [x['result']['action'] for x in replies] == expected['actions']:
raise AssertionError('Contract check failed')
if not [x['id'] for x in replies] == [x['id'] for x in elic]:
raise AssertionError('Contract check failed')
if row['mode'] == 'nested':
if not row['nested_error_code'] == expected['nested_error_code']:
raise AssertionError('Contract check failed')
nested_id = lists[-1]['id']
index = next((i for i, x in enumerate(wire) if x['from'] == 'server' and x['message'].get('id') == nested_id and ('result' in x['message'])))
timeline = row['timeline']
arrived = next((i for i, e in enumerate(timeline) if e.get('wire_index') == index))
timeout = next((i for i, e in enumerate(timeline) if e['event'] == 'nested_timeout'))
if not arrived < timeout:
raise AssertionError((arrived, timeout))
row['reply_arrived_before_timeout'] = True
async def main():
versions = {'python': platform.python_version(), **{n: importlib.metadata.version(n) for n in ('mcp', 'pydantic', 'anyio')}}
if not versions == {'python': '3.12.14', 'mcp': '1.29.0', 'pydantic': '2.13.4', 'anyio': '4.14.2'}:
raise AssertionError(versions)
expected = json.loads((P / 'expected/reentrancy.json').read_text())
rows = []
try:
with anyio.fail_after(90):
for repeat in range(2):
for mode in ('nested', 'prefetch', 'fixed_fixture'):
row = {'mode': mode, 'timeline': [], 'repeat': repeat + 1}
rows.append(row)
await scenario(mode, row)
verify(row, expected[mode])
finally:
out = P / 'run-output'
out.mkdir(exist_ok=True)
(out / 'reentrancy.json').write_text(json.dumps({'versions': versions, 'cases': rows}, ensure_ascii=False, indent=2) + '\n')
print(json.dumps({'passed_scenarios': len(rows), 'normal_exits': sum((r['returncode'] == 0 for r in rows)), 'forced_kills': sum((r['forced_kill'] for r in rows)), 'nested_replies_before_timeout': sum((r.get('reply_arrived_before_timeout', False) for r in rows)), 'same_session_recoveries': len(rows)}, indent=2))
if __name__ == '__main__':
anyio.run(main)
verify_all.py 以 optimize=0 编译正文测试，使 python -O 也执行其中的断言；本节的检查使用显式异常，不依赖 assert。两处结果比较都复用前文的 require_same_json。test_oracle.py 除了翻转 computed 和颠倒响应到达与超时事件，还把 computed 换成等值整数；共 26 次污染都必须被拒绝。独立预期文件始终是写定的常量。
verify_all.py
"""Run original checks with assertions retained, then optimization-safe supplement."""
import runpy
from pathlib import Path
ROOT = Path(__file__).resolve().parent
path = ROOT / "tests/test_contract.py"
# Keep the tutorial's assertion checks under -O.
exec(compile(path.read_text(), str(path), "exec", optimize=0),
{"__name__": "__main__", "__file__": str(path)})
runpy.run_path(str(ROOT / "test_reentrancy.py"), run_name="__main__")
test_oracle.py
"""Deliberately corrupt results; the verifier must reject each mutation under -O too."""
import copy, json
from pathlib import Path
from test_reentrancy import verify
ROOT = Path(__file__).resolve().parent
rows = json.loads((ROOT / "run-output/reentrancy.json").read_text())["cases"]
expected = json.loads((ROOT / "expected/reentrancy.json").read_text())
rejected = 0
for original in rows:
verify(copy.deepcopy(original), expected[original["mode"]])
for field in ("first", "recovery"):
row = copy.deepcopy(original)
row[field]["computed"] = not row[field]["computed"]
try:
verify(row, expected[row["mode"]])
except AssertionError:
rejected += 1
else:
raise RuntimeError("Verifier accepted a corrupted computed flag")
for original in rows:
for field in ("first", "recovery"):
row = copy.deepcopy(original)
row[field]["computed"] = int(row[field]["computed"])
try:
verify(row, expected[row["mode"]])
except AssertionError:
rejected += 1
else:
raise RuntimeError("Verifier accepted an integer computed flag")
for original in rows:
if original["mode"] != "nested":
continue
row = copy.deepcopy(original)
row["timeline"] = sorted(row["timeline"], key=lambda e: e["event"] != "nested_timeout")
try:
verify(row, expected["nested"])
except AssertionError:
rejected += 1
else:
raise RuntimeError("Verifier accepted response arrival after timeout")
if rejected != 26:
raise RuntimeError(rejected)
print(json.dumps({"rejected_mutations": rejected}))
在原有固定依赖环境中运行；脚本不安装或升级软件。先回归原文，再运行本节：
python verify_all.py
python test_result_types.py
python test_oracle.py
python -O verify_all.py
python -O test_result_types.py
python -O test_oracle.py2026-10-13 本地实测：原有 12 案通过；新增三个场景各运行两遍，共 6 次场景运行。全部子进程正常退出，没有强制终止。两个 nested 场景均证实响应先到、嵌套等待后超时，首个工具结果为 cancelled 且 computed=false；prefetch 和 fixed_fixture 首次结果均为 21。六次同会话后续调用都返回 21。完整合成消息与事件顺序由脚本写入 run-output/reentrancy.json。
{
"passed_scenarios": 6,
"normal_exits": 6,
"forced_kills": 0,
"nested_replies_before_timeout": 2,
"same_session_recoveries": 6
}
嵌套请求的 0.5 秒上限只用于结束这次实验；tools/call 仍单独限制为 3 秒。2026-10-20 复跑时，旧版把初始化也放进 3 秒响应上限，连续两次在 initialize 阶段超时，尚未进入回调。现在会话默认响应上限设为 10 秒，整套实验上限设为 90 秒，给六次冷启动留出空间；这不延长嵌套请求的 0.5 秒等待。子进程沿用原文的有限退出与清理。增加等待时间不能消除同一接收循环上的依赖关系。示例把本地元数据失败后的动作明确选为 cancel，不能据此把真实网络超时说成“用户取消”。生产界面还需单独呈现错误来源。
本节没有运行真实宿主界面、HTTP、URL 模式、取消通知或实际审批，也没有证明所有并发路径均可恢复。版本断言不同就停止；原文的单次固定回调并不存在本节主动加入的嵌套请求。保留这些范围，才能把一次 SDK 接线实验与协议保证分开。
2026-10-20 按上述最终代码复验：普通模式与 -O 均通过原有十二案、六个重入场景、十二个整数标记回放污染与二十六个重入判据污染。六个场景均正常退出且无强制终止，两个 nested 响应均先到达再超时，六次同会话恢复均返回 21。此处的版本、协议、纯计算与模拟选择范围不变，未增加真实宿主界面、远程 HTTP、模型、凭据或生产验收。
accept 可以直接作为发送消息或付款的授权吗？
不能。它表示接受并回传这份输入，外部操作的身份、权限和确认需要另行设计与验证。本文只有虚构整数纯计算。
收到 cancel 会重新询问吗？
不会。本例立即返回 cancelled，拒绝也同样结束当前调用，没有自动重试。
elicit_form 会按 requestedSchema 检查所有回传字段吗？
固定 SDK 1.29.0 的本次实验中不会：缺字段的 accept 仍返回 ElicitResult。高层 ctx.elicit 会用传入的模型验证接受内容。
为什么不支持的诊断案例是 -32600？
这是 SDK 1.29.0 默认客户端处理器的实测值，固定版规范要求的是 -32602。正常路径在服务端能力守卫处结束，不发请求。