Host 如何把 MCP 接进模型工具循环
从发现目录到平台结果
沿用MCP 1.12.4环境并安装jsonschema==4.26.0。两个独立Server提供同名search；Host为模型生成唯一名称，注册表保存名称到真实session的映射。以下只用固定模型意图隔离宿主逻辑，真实MCP初始化、目录、工具调用与错误返回均运行。不会调用云模型。
保存host_server.py。crash_for_test只用于故障实验，会退出自己的临时Server进程；Host不会把它放进模型可见目录。
python
import os
from mcp.server.fastmcp import FastMCP
mcp=FastMCP('scope lab')
@mcp.tool()
def search(query:str)->str:
if query not in {'public','private'}:raise ValueError('unknown query')
if query=='private':raise PermissionError('service policy denies private data')
return os.environ.get('LAB_SOURCE','docs')+': public evidence'
@mcp.tool()
def crash_for_test()->str:
"""Local fault injection only; terminate this disposable test server."""
os._exit(7)
if __name__=='__main__':mcp.run(transport='stdio')同目录保存host.py，执行python host.py：
python
import asyncio,json,sys
from contextlib import AsyncExitStack
from mcp import ClientSession,StdioServerParameters
from mcp.client.stdio import stdio_client
from jsonschema import Draft202012Validator
from jsonschema.exceptions import ValidationError
async def open_server(stack,alias):
read,write=await stack.enter_async_context(stdio_client(StdioServerParameters(
command=sys.executable,args=['host_server.py'],env={'LAB_SOURCE':alias})))
session=await stack.enter_async_context(ClientSession(read,write))
await session.initialize()
return session
async def run():
async with AsyncExitStack() as stack:
sessions={a:await open_server(stack,a) for a in ('docs','tickets')}
registry={};declarations=[]
for alias,session in sessions.items():
cursor=None
while True:
page=await session.list_tools(cursor=cursor)
for tool in page.tools:
if tool.name!='search':continue # test-only crash is not exposed to model
name=alias+'__'+tool.name
registry[name]=(session,tool)
declarations.append({'type':'function','name':name,'description':tool.description or '',
'parameters':tool.inputSchema})
cursor=page.nextCursor
if not cursor:break
assert {d['name'] for d in declarations}=={'docs__search','tickets__search'}
executed=[]
async def execute(intent,actor):
name=intent['name'];session,tool=registry[name]
args=json.loads(intent['arguments'])
Draft202012Validator(tool.inputSchema).validate(args)
# actor is authenticated host context, not part of model arguments.
if actor=='reader' and name.startswith('tickets__'):raise PermissionError('host scope')
result=await session.call_tool(tool.name,args)
executed.append(name)
data={'ok':not result.isError,'content':[x.model_dump() for x in result.content]}
return {'type':'function_call_output','call_id':intent['call_id'],'output':json.dumps(data)}
intent={'call_id':'call-1','name':'docs__search','arguments':'{"query":"public"}'}
output=await execute(intent,'reader')
assert output['call_id']=='call-1' and json.loads(output['output'])['ok']
# Deterministic model adapter consumes the actual protocol result; no paid model call.
final='answer with evidence' if json.loads(output['output'])['ok'] else 'cannot answer'
assert final=='answer with evidence'
for bad in ({**intent,'arguments':'{"query":3}'},{**intent,'name':'tickets__search'}):
try:await execute(bad,'reader')
except (ValueError,PermissionError,ValidationError):pass
else:raise AssertionError('invalid intention accepted')
assert executed==['docs__search']
failed=await execute({**intent,'call_id':'call-2','arguments':'{"query":"private"}'},'reader')
assert not json.loads(failed['output'])['ok']
print('discovery, schema, scope, service failure and correlated result passed')
async def reconnect():
try:
async with AsyncExitStack() as stack:
session=await open_server(stack,'docs')
await session.call_tool('crash_for_test',{})
except Exception:pass # transport closes; new session must prove recovery
else:raise AssertionError('crash should close transport')
async with AsyncExitStack() as stack:
session=await open_server(stack,'docs')
result=await session.call_tool('search',{'query':'public'})
assert not result.isError
print('crashed process replaced and session initialized again')
asyncio.run(run())
asyncio.run(reconnect())实际过程是list_tools分页→平台function声明→固定模型意图→schema与身份校验→session.call_tool→平台function_call_output→下一轮回答。function_call_output使用模型call_id，MCP请求ID由ClientSession内部维护；不能把两种身份互换。伪造参数类型或跨scope不会触发服务函数；已授权调用也可能被Server业务政策拒绝，此时isError=True需要进入失败观察。
进程退出后正在等待的RPC抛错，Host建立新进程并重新initialize才能查询。这里只对只读查询重建会话；写调用中断时不能立即重放，因为外部副作用可能已发生。重连不是恢复旧会话，也不保证重放旧事件；业务幂等与任务查询另行实现。
练习
给reader增加tickets只读授权，并保留对private查询的服务端拒绝。证明两个同名search实际返回不同来源，模型call_id分别保留；给不存在的映射名输入应明确拒绝而不能落到任意默认Server。