先查询服务，再读本地策略
这次用户的问题是：“查docs服务的运行许可规则，再读policy.txt，说明两者是否一致。”思想篇已经追过目录如何变成模型可见工具；这里把开发时真正要接好的三个边界展开。
公开源码在第十九课。适配核心使用Python 3.10+标准库，真实MCP通信固定使用官方mcp==1.26.0。2026-10-20本机以Python 3.14.6验证，协议实际协商为2025-11-25。
bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 -m venv .venv
.venv/bin/python -m pip install -r examples/s19-mcp-tools/requirements.txt
.venv/bin/python examples/s19-mcp-tools/user_code/practice.py --decision allow
.venv/bin/python examples/s19-mcp-tools/user_code/practice.py模型为固定替身，MCP客户端、服务子进程、stdio通信和本地文件读取都真实执行；没有在线模型或外部业务系统。正常路径四次逻辑模型请求：发现目录、查询服务、读取文件、回答。默认拒绝外部查询时，三次请求后如实报告缺少证据，服务端没有tools/call。
固定模型按预设选择工具和回答，验证的是报文传递与许可分支，不证明真实模型能正确比较文本。服务本身只返回本课拥有的固定运行许可规则，不连接公司的文档库，也不运行用户程序。
先把会话建好，别把连接与工具发布混为一谈
宿主只启动本仓库的server.py，命令与路径不能由模型填写。官方SDK的stdio_client管理子进程，ClientSession负责初始化、协商和消息匹配；宿主确认tools能力后才进入原来的Agent循环。
python
# user_code/practice.py：demo中的实际会话边界
async with stdio_client(params) as (read, write):
async with ClientSession(
read, write, read_timeout_seconds=timedelta(seconds=5)
) as session:
initialized = await session.initialize()
# 确认 initialized.capabilities.tools 后再运行本课流程初始工具只有read_file和load_docs_tools。后者的名字刻意说明它做的是“读取并接入目录”：服务会话已经由宿主建好，这次工具请求不安装软件、不接受任意启动命令。
SDK会话是异步的，前课唯一run_loop是同步的。SessionBridge让旧循环在asyncio.to_thread中运行，用run_coroutine_threadsafe把list_tools和call_tool投回创建session的事件循环。不要在那个事件循环线程里直接等待这个同步适配器，否则会自己阻塞自己。
SDK请求读取限制为五秒，适配等待为六秒；退出同一组async上下文时关闭会话与stdio。连接生命周期没有另写一套协议栈，也没有把普通Python函数调用假装成真实MCP通信。
discover_tools：拿到第一页，还不能发布目录
函数接收一个list_page回调，输入当前cursor，取得MCP解析后的页面。它把工具放入局部候选，全部读取与验证完成后才返回列表。
这一层不关心stdio还是HTTP，但关心目录是否完整：每页必须有tools列表，分页游标不能重复；最多读取四页、八项工具。单页按解析后JSON重新编码计量，限24000字节，这是教学预算，并非传输层的消息大小限制。
真实执行轨迹回放；原样适配函数，分页与服务调用为可展开的内存替身；不启动MCP进程或请求在线模型。
Pys19/core/runtime.py · discover_tools
未运行
def discover_tools(list_page):
"""最多四页八项；完整验证后才返回目录，不发布半份结果。"""
tools, seen, cursors, cursor = [], set(), set(), None
for _ in range(4):
page = list_page(cursor)
if not isinstance(page, dict) or not isinstance(page.get('tools'), list):
raise ValueError('工具目录页面不合法。')
if len(json.dumps(page, ensure_ascii=False).encode()) > 24000:
raise ValueError('单页目录超过24000字节。')
for raw in page['tools']:
tool = checked_tool(raw)
if tool['name'] in seen or len(tools) >= 8:
raise ValueError('工具重复或超过八项。')
seen.add(tool['name'])
tools.append(tool)
cursor = page.get('nextCursor')
if cursor is None:
return tools
if not isinstance(cursor, str) or not cursor or len(cursor) > 120 or cursor in cursors:
raise ValueError('分页游标无效或重复。')
cursors.add(cursor)
raise ValueError('工具目录超过四页。')输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
import re
def checked_tool(raw):
"""只接入本课支持的必填短字符串对象；不冒充通用JSON Schema验证器。"""
if not isinstance(raw, dict) or not isinstance(raw.get('name'),str) or not re.fullmatch(r'[A-Za-z0-9_.-]{1,32}', raw.get('name', '')):
raise ValueError('外部工具名称不受支持。')
schema = raw.get('inputSchema')
if not isinstance(schema, dict) or set(schema) != {'type','properties','required','additionalProperties'}:
raise ValueError('本课只支持显式字段的字符串对象schema。')
props = schema['properties']
if (schema['type'] != 'object' or schema['additionalProperties'] is not False
or not isinstance(props, dict) or len(props) > 4
or not isinstance(schema['required'], list)
or any(type(key) is not str for key in schema['required'])
or len(schema['required']) != len(props) or set(schema['required']) != set(props)
or any(not isinstance(key,str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,31}', key) or value != {'type':'string'} for key,value in props.items())):
raise ValueError('schema超出本课支持范围。')
if not isinstance(raw.get('description',''), str) or len(raw.get('description','')) > 400:
raise ValueError('工具说明过长或不是文字。')
return deepcopy(raw)
def result_text(reply):
if not isinstance(reply, dict) or type(reply.get('isError',False)) is not bool:
raise ValueError('MCP结果格式不合法。')
blocks = reply.get('content')
if not isinstance(blocks, list) or not blocks or len(blocks) > 4 or reply.get('structuredContent') is not None:
raise ValueError('本课只接收一到四块纯文本结果。')
if any(not isinstance(b,dict) or b.get('type') != 'text' or not isinstance(b.get('text'),str) for b in blocks):
raise ValueError('结果包含本课不支持的内容类型。')
text = '\n'.join(block['text'] for block in blocks)
if not text.strip() or len(text) > 2000:
raise ValueError('外部结果为空或超过2000字符。')
return text, reply.get('isError', False)
def discover_tools(list_page):
"""最多四页八项；完整验证后才返回目录，不发布半份结果。"""
tools, seen, cursors, cursor = [], set(), set(), None
for _ in range(4):
page = list_page(cursor)
if not isinstance(page, dict) or not isinstance(page.get('tools'), list):
raise ValueError('工具目录页面不合法。')
if len(json.dumps(page, ensure_ascii=False).encode()) > 24000:
raise ValueError('单页目录超过24000字节。')
for raw in page['tools']:
tool = checked_tool(raw)
if tool['name'] in seen or len(tools) >= 8:
raise ValueError('工具重复或超过八项。')
seen.add(tool['name'])
tools.append(tool)
cursor = page.get('nextCursor')
if cursor is None:
return tools
if not isinstance(cursor, str) or not cursor or len(cursor) > 120 or cursor in cursors:
raise ValueError('分页游标无效或重复。')
cursors.add(cursor)
raise ValueError('工具目录超过四页。')
def assemble_tool_pool(builtins, servers):
"""同一次组装同时产生模型定义和调用路由；失败不改旧池。"""
definitions, routes = deepcopy(builtins), {}
names = {item['name'] for item in definitions}
if len(names) != len(definitions) or len(servers) > 2:
raise ValueError('内置名称重复或服务超过两项。')
for server, client in sorted(servers.items()):
if not re.fullmatch(r'[a-z][a-z0-9-]{0,15}', server) or not callable(client['call']):
raise ValueError('宿主服务名或调用接口不合法。')
if len(client['tools']) > 8:
raise ValueError('服务工具超过八项。')
for raw in client['tools']:
tool = checked_tool(raw)
alias = 'mcp__' + server + '__' + tool['name'].replace('.', '_')
if alias in names:
raise ValueError('映射后的工具名冲突，保留旧池。')
names.add(alias)
definitions.append({'name':alias, 'description':tool.get('description',''),
'input_schema':deepcopy(tool['inputSchema'])})
routes[alias] = {'server':server, 'remote':tool['name'],
'schema':deepcopy(tool['inputSchema']), 'invoke':client['call']}
return {'definitions':definitions, 'routes':routes}
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}
scenario='pages'
schema={'type':'object','properties':{'query':{'type':'string'}},'required':['query'],'additionalProperties':False}
def tool(name):return {'name':name,'description':'本地固定文档查询','inputSchema':deepcopy(schema),'annotations':{'readOnlyHint':True}}
a,b=tool('search.docs'),tool('read.policy')
pages=[{'tools':[a],'nextCursor':'p2'},{'tools':[b]}];page_calls=[];remote_calls=[];approvals=[]
def list_page(cursor):
page_calls.append(cursor)
if scenario in ('repeat','limit'):return {'tools':[],'nextCursor':'again' if scenario=='repeat' else str(len(page_calls))}
return deepcopy(pages[0 if cursor is None else 1])
def invoke(remote,arguments):
remote_calls.append({'name':remote,'arguments':deepcopy(arguments)})
if scenario=='transport':raise TimeoutError('教学超时；不重发调用')
if scenario=='mixed':return {'content':[{'type':'image','data':'omitted'}]}
return {'content':[{'type':'text','text':'x'*2001 if scenario=='oversize' else '运行需要本次许可。'}],'isError':scenario=='tool_error'}
def ask_user(snapshot,server):
approvals.append(server);snapshot['input'].clear()
return scenario!='deny'
builtins=[{'name':'read_file','description':'读取本地短文件','input_schema':{'type':'object','properties':{'path':{'type':'string'}},'required':['path'],'additionalProperties':False}}]
servers={'docs':{'tools':[a],'call':invoke}}
if 'discover_tools'=='discover_tools':
if scenario=='empty':pages=[{'tools':[]}]
if scenario=='duplicate':pages[1]={'tools':[a]}
if scenario=='bad_schema':pages[1]={'tools':[{'name':'later','inputSchema':{'type':'array'}}]}
if scenario=='many':pages=[{'tools':[tool('t'+str(i)) for i in range(9)]}]
if 'discover_tools'=='assemble_tool_pool':
if scenario=='two':servers['other']={'tools':[a],'call':invoke}
if scenario=='collision':servers['docs']['tools'].append(tool('search_docs'))
if scenario=='builtin_collision':builtins[0]['name']='mcp__docs__search_docs'
if scenario=='bad_server':servers={'../docs':servers['docs']}
if scenario=='unsupported':servers['docs']['tools'][0]['inputSchema']={'type':'array'}
before=deepcopy(builtins)
pool=assemble_tool_pool(builtins,{'docs':{'tools':[a],'call':invoke}}) if 'discover_tools'=='call_external' else None
call={'id':'call_2','name':'mcp__docs__search_docs','input':{'query':'运行许可'}}
if scenario=='bad_args':call['input']={'path':'main.py'}
if scenario=='unknown':call['name']='mcp__missing__search_docs'
# 调用入口
try:
observed=discover_tools(list_page)
except (ValueError,TypeError) as error:
observed={"error":str(error)}
print(observed)
- 分页读取并验证候选。
- 读完后返回完整目录，失败不发布部分结果。
选择“两页目录”，停在tools.append行：当前项已通过校验，还没有进入候选。继续单步后它才加入。第一次页面读完仍有p2，不会提前return；下一次list_page收到的就是这个游标。
选择“后页不支持”，第一页已经成功，第二页的schema超出支持范围。函数会失败，调用者不会用半份目录替换旧工具池。选择“重复游标”或“页数用完”，则观察它如何有界停下，而不是一直问服务还有没有下一页。
checked_tool只接入本课能准确验证的schema：显式对象、最多四个必填字符串字段、禁止额外字段。字段名、工具名和说明也有长度限制。数组、可选字段和复杂schema整次拒绝；这些结构并非MCP不支持，而是超出了本例适配器的范围。
模型看到的输入契约必须和宿主实际验证、最终发到服务的参数一致；需要更完整的JSON Schema时，应接入成熟验证器并补相应结果适配。
assemble_tool_pool：定义与路由必须来自同一次组装
输入是内置工具定义和宿主登记的服务表。每个服务条目包含已发现的tools以及实际call接口。输出同时带definitions和routes，旧列表保持不变。
每项外部工具会生成一个模型可见名称，例如mcp__docs__search_docs；路由则保存docs、原名search.docs、输入schema和调用接口。发送给模型的是input_schema，调用服务时仍传原名与arguments。
真实执行轨迹回放；原样适配函数，分页与服务调用为可展开的内存替身；不启动MCP进程或请求在线模型。
Pys19/core/runtime.py · assemble_tool_pool
未运行
def assemble_tool_pool(builtins, servers):
"""同一次组装同时产生模型定义和调用路由；失败不改旧池。"""
definitions, routes = deepcopy(builtins), {}
names = {item['name'] for item in definitions}
if len(names) != len(definitions) or len(servers) > 2:
raise ValueError('内置名称重复或服务超过两项。')
for server, client in sorted(servers.items()):
if not re.fullmatch(r'[a-z][a-z0-9-]{0,15}', server) or not callable(client['call']):
raise ValueError('宿主服务名或调用接口不合法。')
if len(client['tools']) > 8:
raise ValueError('服务工具超过八项。')
for raw in client['tools']:
tool = checked_tool(raw)
alias = 'mcp__' + server + '__' + tool['name'].replace('.', '_')
if alias in names:
raise ValueError('映射后的工具名冲突，保留旧池。')
names.add(alias)
definitions.append({'name':alias, 'description':tool.get('description',''),
'input_schema':deepcopy(tool['inputSchema'])})
routes[alias] = {'server':server, 'remote':tool['name'],
'schema':deepcopy(tool['inputSchema']), 'invoke':client['call']}
return {'definitions':definitions, 'routes':routes}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
import re
def checked_tool(raw):
"""只接入本课支持的必填短字符串对象；不冒充通用JSON Schema验证器。"""
if not isinstance(raw, dict) or not isinstance(raw.get('name'),str) or not re.fullmatch(r'[A-Za-z0-9_.-]{1,32}', raw.get('name', '')):
raise ValueError('外部工具名称不受支持。')
schema = raw.get('inputSchema')
if not isinstance(schema, dict) or set(schema) != {'type','properties','required','additionalProperties'}:
raise ValueError('本课只支持显式字段的字符串对象schema。')
props = schema['properties']
if (schema['type'] != 'object' or schema['additionalProperties'] is not False
or not isinstance(props, dict) or len(props) > 4
or not isinstance(schema['required'], list)
or any(type(key) is not str for key in schema['required'])
or len(schema['required']) != len(props) or set(schema['required']) != set(props)
or any(not isinstance(key,str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,31}', key) or value != {'type':'string'} for key,value in props.items())):
raise ValueError('schema超出本课支持范围。')
if not isinstance(raw.get('description',''), str) or len(raw.get('description','')) > 400:
raise ValueError('工具说明过长或不是文字。')
return deepcopy(raw)
def result_text(reply):
if not isinstance(reply, dict) or type(reply.get('isError',False)) is not bool:
raise ValueError('MCP结果格式不合法。')
blocks = reply.get('content')
if not isinstance(blocks, list) or not blocks or len(blocks) > 4 or reply.get('structuredContent') is not None:
raise ValueError('本课只接收一到四块纯文本结果。')
if any(not isinstance(b,dict) or b.get('type') != 'text' or not isinstance(b.get('text'),str) for b in blocks):
raise ValueError('结果包含本课不支持的内容类型。')
text = '\n'.join(block['text'] for block in blocks)
if not text.strip() or len(text) > 2000:
raise ValueError('外部结果为空或超过2000字符。')
return text, reply.get('isError', False)
def discover_tools(list_page):
"""最多四页八项；完整验证后才返回目录，不发布半份结果。"""
tools, seen, cursors, cursor = [], set(), set(), None
for _ in range(4):
page = list_page(cursor)
if not isinstance(page, dict) or not isinstance(page.get('tools'), list):
raise ValueError('工具目录页面不合法。')
if len(json.dumps(page, ensure_ascii=False).encode()) > 24000:
raise ValueError('单页目录超过24000字节。')
for raw in page['tools']:
tool = checked_tool(raw)
if tool['name'] in seen or len(tools) >= 8:
raise ValueError('工具重复或超过八项。')
seen.add(tool['name'])
tools.append(tool)
cursor = page.get('nextCursor')
if cursor is None:
return tools
if not isinstance(cursor, str) or not cursor or len(cursor) > 120 or cursor in cursors:
raise ValueError('分页游标无效或重复。')
cursors.add(cursor)
raise ValueError('工具目录超过四页。')
def assemble_tool_pool(builtins, servers):
"""同一次组装同时产生模型定义和调用路由；失败不改旧池。"""
definitions, routes = deepcopy(builtins), {}
names = {item['name'] for item in definitions}
if len(names) != len(definitions) or len(servers) > 2:
raise ValueError('内置名称重复或服务超过两项。')
for server, client in sorted(servers.items()):
if not re.fullmatch(r'[a-z][a-z0-9-]{0,15}', server) or not callable(client['call']):
raise ValueError('宿主服务名或调用接口不合法。')
if len(client['tools']) > 8:
raise ValueError('服务工具超过八项。')
for raw in client['tools']:
tool = checked_tool(raw)
alias = 'mcp__' + server + '__' + tool['name'].replace('.', '_')
if alias in names:
raise ValueError('映射后的工具名冲突，保留旧池。')
names.add(alias)
definitions.append({'name':alias, 'description':tool.get('description',''),
'input_schema':deepcopy(tool['inputSchema'])})
routes[alias] = {'server':server, 'remote':tool['name'],
'schema':deepcopy(tool['inputSchema']), 'invoke':client['call']}
return {'definitions':definitions, 'routes':routes}
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}
scenario='one'
schema={'type':'object','properties':{'query':{'type':'string'}},'required':['query'],'additionalProperties':False}
def tool(name):return {'name':name,'description':'本地固定文档查询','inputSchema':deepcopy(schema),'annotations':{'readOnlyHint':True}}
a,b=tool('search.docs'),tool('read.policy')
pages=[{'tools':[a],'nextCursor':'p2'},{'tools':[b]}];page_calls=[];remote_calls=[];approvals=[]
def list_page(cursor):
page_calls.append(cursor)
if scenario in ('repeat','limit'):return {'tools':[],'nextCursor':'again' if scenario=='repeat' else str(len(page_calls))}
return deepcopy(pages[0 if cursor is None else 1])
def invoke(remote,arguments):
remote_calls.append({'name':remote,'arguments':deepcopy(arguments)})
if scenario=='transport':raise TimeoutError('教学超时；不重发调用')
if scenario=='mixed':return {'content':[{'type':'image','data':'omitted'}]}
return {'content':[{'type':'text','text':'x'*2001 if scenario=='oversize' else '运行需要本次许可。'}],'isError':scenario=='tool_error'}
def ask_user(snapshot,server):
approvals.append(server);snapshot['input'].clear()
return scenario!='deny'
builtins=[{'name':'read_file','description':'读取本地短文件','input_schema':{'type':'object','properties':{'path':{'type':'string'}},'required':['path'],'additionalProperties':False}}]
servers={'docs':{'tools':[a],'call':invoke}}
if 'assemble_tool_pool'=='discover_tools':
if scenario=='empty':pages=[{'tools':[]}]
if scenario=='duplicate':pages[1]={'tools':[a]}
if scenario=='bad_schema':pages[1]={'tools':[{'name':'later','inputSchema':{'type':'array'}}]}
if scenario=='many':pages=[{'tools':[tool('t'+str(i)) for i in range(9)]}]
if 'assemble_tool_pool'=='assemble_tool_pool':
if scenario=='two':servers['other']={'tools':[a],'call':invoke}
if scenario=='collision':servers['docs']['tools'].append(tool('search_docs'))
if scenario=='builtin_collision':builtins[0]['name']='mcp__docs__search_docs'
if scenario=='bad_server':servers={'../docs':servers['docs']}
if scenario=='unsupported':servers['docs']['tools'][0]['inputSchema']={'type':'array'}
before=deepcopy(builtins)
pool=assemble_tool_pool(builtins,{'docs':{'tools':[a],'call':invoke}}) if 'assemble_tool_pool'=='call_external' else None
call={'id':'call_2','name':'mcp__docs__search_docs','input':{'query':'运行许可'}}
if scenario=='bad_args':call['input']={'path':'main.py'}
if scenario=='unknown':call['name']='mcp__missing__search_docs'
# 调用入口
try:
observed=assemble_tool_pool(builtins,servers)
observed={'names':[x['name'] for x in observed['definitions']], 'routes':{key:{'server':value['server'],'remote':value['remote']} for key,value in observed['routes'].items()}}
except (ValueError,TypeError) as error:
observed={"error":str(error)}
print(observed)
- 从内置定义副本开始。
- 生成唯一名称，同时建立原服务名路由。
- 完整快照返回后才发布。
在definitions.append前停下，名称已经准备好，定义还没追加；到routes赋值行，模型定义已进入候选，但路由尚未加入。只有两个部分都完成，函数才返回整个候选。调用者收到返回值以后再替换当前池。
选择“两个服务同名工具”，docs和other都提供search.docs，外显名称不同，分别能路由回各自服务。这里直接保存调用接口和原名，没有用一个未绑定的循环变量去猜最后该调用哪一个服务。
选择“映射名称冲突”，同一服务的search.docs和search_docs会生成相同外显名称，整次组装被拒绝。与内置工具撞名也一样处理，不覆盖其中一个。模型工具名受当前宿主策略约束，不能把“加了前缀”当成永远不会冲突的证明。
本例最多接入两个服务，每个服务八项工具。组装器再次校验契约，防止调用者跳过discover_tools直接塞进一个不受支持的目录。新池只供当前会话使用，不自动分发给所有队友。
call_external：查到路由，还要验证和确认
输入是模型的工具请求、这次请求对应的工具快照和宿主确认回调。函数先复制请求，保留模型调用ID，再查路由和输入schema。
字段必须完全匹配，值只能是最长120字符的字符串。未知路由或参数不符都在本地结束，没有tools/call。宿主拿到确认副本，即使它修改显示对象，也不能改变随后发送的参数。
真实执行轨迹回放；原样适配函数，分页与服务调用为可展开的内存替身；不启动MCP进程或请求在线模型。
Pys19/core/runtime.py · call_external
未运行
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
import json
import re
def checked_tool(raw):
"""只接入本课支持的必填短字符串对象；不冒充通用JSON Schema验证器。"""
if not isinstance(raw, dict) or not isinstance(raw.get('name'),str) or not re.fullmatch(r'[A-Za-z0-9_.-]{1,32}', raw.get('name', '')):
raise ValueError('外部工具名称不受支持。')
schema = raw.get('inputSchema')
if not isinstance(schema, dict) or set(schema) != {'type','properties','required','additionalProperties'}:
raise ValueError('本课只支持显式字段的字符串对象schema。')
props = schema['properties']
if (schema['type'] != 'object' or schema['additionalProperties'] is not False
or not isinstance(props, dict) or len(props) > 4
or not isinstance(schema['required'], list)
or any(type(key) is not str for key in schema['required'])
or len(schema['required']) != len(props) or set(schema['required']) != set(props)
or any(not isinstance(key,str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,31}', key) or value != {'type':'string'} for key,value in props.items())):
raise ValueError('schema超出本课支持范围。')
if not isinstance(raw.get('description',''), str) or len(raw.get('description','')) > 400:
raise ValueError('工具说明过长或不是文字。')
return deepcopy(raw)
def result_text(reply):
if not isinstance(reply, dict) or type(reply.get('isError',False)) is not bool:
raise ValueError('MCP结果格式不合法。')
blocks = reply.get('content')
if not isinstance(blocks, list) or not blocks or len(blocks) > 4 or reply.get('structuredContent') is not None:
raise ValueError('本课只接收一到四块纯文本结果。')
if any(not isinstance(b,dict) or b.get('type') != 'text' or not isinstance(b.get('text'),str) for b in blocks):
raise ValueError('结果包含本课不支持的内容类型。')
text = '\n'.join(block['text'] for block in blocks)
if not text.strip() or len(text) > 2000:
raise ValueError('外部结果为空或超过2000字符。')
return text, reply.get('isError', False)
def discover_tools(list_page):
"""最多四页八项；完整验证后才返回目录，不发布半份结果。"""
tools, seen, cursors, cursor = [], set(), set(), None
for _ in range(4):
page = list_page(cursor)
if not isinstance(page, dict) or not isinstance(page.get('tools'), list):
raise ValueError('工具目录页面不合法。')
if len(json.dumps(page, ensure_ascii=False).encode()) > 24000:
raise ValueError('单页目录超过24000字节。')
for raw in page['tools']:
tool = checked_tool(raw)
if tool['name'] in seen or len(tools) >= 8:
raise ValueError('工具重复或超过八项。')
seen.add(tool['name'])
tools.append(tool)
cursor = page.get('nextCursor')
if cursor is None:
return tools
if not isinstance(cursor, str) or not cursor or len(cursor) > 120 or cursor in cursors:
raise ValueError('分页游标无效或重复。')
cursors.add(cursor)
raise ValueError('工具目录超过四页。')
def assemble_tool_pool(builtins, servers):
"""同一次组装同时产生模型定义和调用路由；失败不改旧池。"""
definitions, routes = deepcopy(builtins), {}
names = {item['name'] for item in definitions}
if len(names) != len(definitions) or len(servers) > 2:
raise ValueError('内置名称重复或服务超过两项。')
for server, client in sorted(servers.items()):
if not re.fullmatch(r'[a-z][a-z0-9-]{0,15}', server) or not callable(client['call']):
raise ValueError('宿主服务名或调用接口不合法。')
if len(client['tools']) > 8:
raise ValueError('服务工具超过八项。')
for raw in client['tools']:
tool = checked_tool(raw)
alias = 'mcp__' + server + '__' + tool['name'].replace('.', '_')
if alias in names:
raise ValueError('映射后的工具名冲突，保留旧池。')
names.add(alias)
definitions.append({'name':alias, 'description':tool.get('description',''),
'input_schema':deepcopy(tool['inputSchema'])})
routes[alias] = {'server':server, 'remote':tool['name'],
'schema':deepcopy(tool['inputSchema']), 'invoke':client['call']}
return {'definitions':definitions, 'routes':routes}
def call_external(call, pool, ask_user):
"""宿主确认独立于server标注；一次调用，无自动工具重试。"""
snapshot = deepcopy(call)
envelope = {'type':'tool_result', 'tool_use_id':snapshot['id']}
try:
route = pool['routes'].get(snapshot.get('name'))
if route is None:
raise ValueError('本次工具快照中没有这个路由。')
arguments = snapshot.get('input')
if (not isinstance(arguments,dict) or set(arguments) != set(route['schema']['properties'])
or any(type(value) is not str or len(value) > 120 for value in arguments.values())):
raise ValueError('外部工具参数字段、类型或长度不符。')
if ask_user(deepcopy(snapshot), route['server']) is not True:
return {**envelope, 'is_error':True, 'content':'未执行：本次外部调用未获批准。'}
reply = route['invoke'](route['remote'], deepcopy(arguments))
text, failed = result_text(reply)
return {**envelope, 'is_error':failed, 'content':text}
except Exception as error:
return {**envelope, 'is_error':True, 'content':'外部调用未能完成确认：' + type(error).__name__ + ': ' + str(error)[:180]}
scenario='success'
schema={'type':'object','properties':{'query':{'type':'string'}},'required':['query'],'additionalProperties':False}
def tool(name):return {'name':name,'description':'本地固定文档查询','inputSchema':deepcopy(schema),'annotations':{'readOnlyHint':True}}
a,b=tool('search.docs'),tool('read.policy')
pages=[{'tools':[a],'nextCursor':'p2'},{'tools':[b]}];page_calls=[];remote_calls=[];approvals=[]
def list_page(cursor):
page_calls.append(cursor)
if scenario in ('repeat','limit'):return {'tools':[],'nextCursor':'again' if scenario=='repeat' else str(len(page_calls))}
return deepcopy(pages[0 if cursor is None else 1])
def invoke(remote,arguments):
remote_calls.append({'name':remote,'arguments':deepcopy(arguments)})
if scenario=='transport':raise TimeoutError('教学超时；不重发调用')
if scenario=='mixed':return {'content':[{'type':'image','data':'omitted'}]}
return {'content':[{'type':'text','text':'x'*2001 if scenario=='oversize' else '运行需要本次许可。'}],'isError':scenario=='tool_error'}
def ask_user(snapshot,server):
approvals.append(server);snapshot['input'].clear()
return scenario!='deny'
builtins=[{'name':'read_file','description':'读取本地短文件','input_schema':{'type':'object','properties':{'path':{'type':'string'}},'required':['path'],'additionalProperties':False}}]
servers={'docs':{'tools':[a],'call':invoke}}
if 'call_external'=='discover_tools':
if scenario=='empty':pages=[{'tools':[]}]
if scenario=='duplicate':pages[1]={'tools':[a]}
if scenario=='bad_schema':pages[1]={'tools':[{'name':'later','inputSchema':{'type':'array'}}]}
if scenario=='many':pages=[{'tools':[tool('t'+str(i)) for i in range(9)]}]
if 'call_external'=='assemble_tool_pool':
if scenario=='two':servers['other']={'tools':[a],'call':invoke}
if scenario=='collision':servers['docs']['tools'].append(tool('search_docs'))
if scenario=='builtin_collision':builtins[0]['name']='mcp__docs__search_docs'
if scenario=='bad_server':servers={'../docs':servers['docs']}
if scenario=='unsupported':servers['docs']['tools'][0]['inputSchema']={'type':'array'}
before=deepcopy(builtins)
pool=assemble_tool_pool(builtins,{'docs':{'tools':[a],'call':invoke}}) if 'call_external'=='call_external' else None
call={'id':'call_2','name':'mcp__docs__search_docs','input':{'query':'运行许可'}}
if scenario=='bad_args':call['input']={'path':'main.py'}
if scenario=='unknown':call['name']='mcp__missing__search_docs'
# 调用入口
try:
observed=call_external(call,pool,ask_user)
except (ValueError,TypeError) as error:
observed={"error":str(error)}
print(observed)
- 固定调用ID并验证路由与参数。
- 宿主确认后只调用一次外部服务。
- 保留文本/错误标记，关联原调用ID。
在route['invoke']行暂停，本次许可已通过，但调用还没有发生。下一步才用search.docs这个原名和参数副本进入服务；模型可见的mcp__docs__search_docs不会原样发过去。
选择“拒绝外部调用”，服务虽然声明readOnlyHint，宿主仍可拒绝，调用接口不会被执行。这里的决定来自宿主配置，不从工具说明文字推导授权；原execute_hooked仍包住实际入口。
收到MCP结果以后，result_text检查内容和isError。本课支持一到四块纯文本，合并后非空且不超过2000字符；业务错误映射为is_error=True。多媒体、结构化结果和超长文本没有被悄悄丢掉，而是明确交回当前适配器不支持的错误。
结果最终成为原循环的tool_result，并关联原模型tool_use_id。MCP的JSON-RPC ID由SDK维护，不能拿它替换模型调用ID。选择“服务业务错误”，可以看到调用确实发生过，但反馈仍标记失败；选择“调用超时”，则不能据此推断服务根本没有执行。
函数不自动重发外部调用。模型请求恢复继续由第十一课处理；恢复一份已经包含工具反馈的模型请求，不会绕回来再查一次服务。
新工具为什么要等下一次模型请求？
run_example保存当前工具池；load_docs_tools完成发现和组装后，才发布新池。每次ask还会固定这一份响应可以使用的快照，definitions、路由和生成system的能力目录都来自它。
如果模型同一批回复里先load目录、再猜测调用新名称，第二项会被拒绝：这项工具还没出现在它刚才接收的定义里。下次ask才读取新池，模型这时才能按新契约调用。标准库测试专门验证这条路径，即使宿主允许查询，越过快照的调用也不会发到服务端。
新增工具后，s10的实际能力缓存键发生变化，system重组，reused为False；后续能力不变时继续复用。这里的复用只减少本地组装，模型请求仍会携带system与当前工具定义。
我们保留原来的有限run_loop、请求恢复和Hook入口，没有增加第二套Agent循环。本课只让一个会话接入已知服务，没有热处理list_changed、断线重连、OAuth或远程HTTP配置；新增工具能力也没有自动扩散给其他成员。
故意让客户端先超时，服务后来才完成
bash
python3 examples/s19-mcp-tools/user_code/verify_practice.py
python3 examples/s19-mcp-tools/user_code/verify_fixtures.py
.venv/bin/python examples/s19-mcp-tools/user_code/verify_mcp.py十二组标准库测试覆盖分页边界、目录不完整不发布、名称冲突、两个服务的原名路由、参数与确认副本，以及同一批次不能提前使用新工具。二十一种同源教学输入共407个真实执行快照，运行与调试stdout一致；浏览器只替换分页和服务调用接口，不启动MCP进程。
真实MCP验证另起本地服务进程，正常、默认拒绝、服务业务错误、坏schema四条路径分别产生4、3、3、2次逻辑模型请求。服务日志中tools/call分别为1、0、1、0次，能区分“没调用”和“调用后失败”。
超时实验再把服务延迟设为一秒，把这一次客户端等待设为0.1秒。服务先记录已接受请求，客户端返回关联错误；保持连接后，还能观察到服务随后完成。这说明取消等待不能证明业务没有发生，更不能成为自动重发的依据。
可选verify_sdk.py需要根requirements中的Anthropic SDK 1.13.0。允许与拒绝两条路径分别为五次、四次模拟HTTP，含已有工具反馈之后的一次529；MCP仍为真实stdio，服务实际调用保持一次或零次，没有在线模型调用。
这条最小链路让外部查询和本地读取进入同一份历史，同时保留了各自的输入契约与许可。要接入真实文档系统，还需要先决定服务信任、凭据、支持的schema与结果类型；标准协议解决通信方式，业务适配与执行边界仍由宿主明确承担。