AI Agent 开始控制现实世界这句口号背后其实是一套工程链路。模型负责生成意图系统负责执行动作而连接这两端的往往不是模型本身而是一个容易被忽略的中间层。在围绕 Anthropic 生态的开发讨论中这个中间层经常被简称为 MHS。这篇文章就从这个缩写入手讲清楚它承担什么职责、为什么 Agent 需要它以及实际项目中如何用代码实现一个类似能力的服务。如果你已经接触过 LLM API、正在开发会调用外部工具的 Agent或者正在排查 Claude API 403、模型路由等错误这篇文章可以直接帮你把概念和实操串起来。读完后你会得到一份可运行的 MHS 风格 Agent 服务也会掌握一套至少覆盖连接错误、权限错误、路由错误的排查链路。更重要的是一句话AI Agent 控制现实世界控制权必须落在受控的中间层而不是模型本身。1. AI Agent 控制现实世界的底层逻辑1.1 从“生成文本”到“生成动作”的关键跳跃传统大模型聊天接口做的事情是生成文本。用户输入一句话模型返回一段回答。这不会对现实世界产生任何副作用最多只是信息呈现。AI Agent 不一样。它要执行动作比如查数据库、调用接口、发消息、操控设备。要做到这一点模型不再只输出自然语言而是输出一种结构化指令通常是tool_use类型的消息块里面包含工具名例如query_device_status输入参数例如{ device_id: DEV-1001 }意图描述模型为什么要调用这个工具当 Agent 运行时收到这样的输出它不会直接把这段 JSON 拿去执行而是先交给某个具备执行能力的中间系统。这个中间系统负责找到对应工具、校验参数、运行工具、拿回结果再把结果回传给模型。这个流程就是 AI Agent 从对话模型变成行动系统的本质变化。模型负责决定“做什么”中间系统负责确认“能不能做”以及“做成什么样”。1.2 为什么模型不能直接触碰系统一个很容易踩坑的设计是让模型直接生成一条系统命令然后由 Agent 主进程直接执行。这在 demo 里看起来很快但进入真实环境几乎不可用。第一个原因是参数不可信。模型输出是概率采样结果同一个指令可能出现参数类型错误、字段缺失甚至幻觉出的工具名。如果不经过校验一个/bin/rm工具可能因为参数拼错而删掉不该删的文件。第二个原因是目标系统语义不同。数据库、HTTP API、消息队列、文件系统各自的调用方式完全不同。模型只需要声明“我要查 DEV-1001 的状态”中间层必须把它翻译成具体系统的请求格式。第三个原因是权限边界。模型没有用户身份它只是输出了一段建议。真正决定这个调用是否被允许的应该是当前会话的权限模型。如果模型可以直接执行任何工具任何通过 Prompt 注入方式发送的恶意工具调用都会被无差别执行。所以模型永远不应该直接握着系统权限。中间层必须承担参数校验、权限校验、超时控制、异常隔离和执行记录。1.3 AI Agent 执行动作的完整链路一个比较标准的 Agent 动作链路可以拆成七个环节环节负责方典型产出用户发起请求前端/聊天入口用户消息意图理解大模型自然语言回答或tool_use指令指令接收Agent Runtime提取工具名和参数指令校验中间层合法的执行计划动作执行工具适配器数据库记录、HTTP 响应、命令输出结果回填Agent Runtimetool_result消息结果总结大模型面向用户的最终回答这个链路里第三个到第五个环节就是 MHS 这类中间层最应该发挥作用的地方。很多初学 Agent 的开发者只关注大模型如何调用工具却忽略了中间层对工具调用结果的封装、错误处理和日志记录。一旦 Agent 开始控制真实系统这部分投入会直接决定系统是否安全、是否可维护。2. 理解 MHS它是模型与世界之间的翻译层2.1 MHS 并不是一个固定的官方产品名称在围绕 Anthropic 生态的讨论中MHS 这个缩写并不像 Claude API、Model Context Protocol 那样有非常明确的官方文档定义。不同团队在内部会用它指代不同模块比较常见的拆解有以下几种缩写展开典型场景Model Handling System面向模型输入输出的一整套处理流程重点是工具调用处理Message Handling Service消息协议转换、事件转发强调服务间通信Managed Hosted Service部署形态指由平台托管运行的一种服务在本文中我们取“Model Handling System”这个角度来理解它。从工程语义看MHS 是一套位于模型和真实工具之间的处理系统它接收模型生成的指令把指令翻译成可执行的工具调用再把执行结果变成模型能继续理解的上下文。如果你的项目文档里出现了 MHS建议先确认它到底指什么不要直接套用本文的全称。技术文章的落地能力来自概念清晰而概念清晰的前提是术语在团队内有明确共识。2.2 MHS 的核心职责四个“必须做”一个能被称作 MHS 的系统至少要承担四个职责。第一协议转换。大模型输出的是统一格式的tool_use但每个外部工具都有自己的请求格式。MHS 需要把模型的标准指令转换成目标工具真正需要的结构。第二工具注册与路由。系统里可能挂了十几个工具模型声明了要调用query_device_statusMHS 必须快速找到对应实现。这个过程不能靠硬编码 if-else而应该有一张工具注册表用名称做路由。第三执行与结果规范化。工具执行可能成功也可能失败。失败可能是超时、网络异常、权限不足、业务规则拒绝。MHS 要负责把这些异常统一封装成模型能理解的tool_result而不是让模型看到一长串堆栈。第四执行轨迹记录。每个工具调用都要留下 audit log包括谁触发的、模型给什么参数、工具实际返回什么、耗时多少。这不是锦上添花而是 Agent 出现误操作后回溯问题的唯一线索。2.3 MHS 与 MCP、Claude Agent SDK 的分工很多人在学习 Anthropic 生态时会被多个名词搞混尤其是 Model Context Protocol、Claude Agent SDK 和 MHS 的关系。Model Context Protocol 是一种开放协议解决的是“外部工具如何标准化暴露给模型”的问题。它把文件系统、数据库、HTTP API 包装成 MCP Server让模型通过统一协议去访问。Claude Agent SDK 是构建 Agent 主程序的开发工具帮助开发者编排模型调用、工具调用循环和输入输出。MHS 则更接近运行期的一个中间层角色。它使用 MCP 或普通工具定义来获取工具清单但它更关注执行控制本身工具是否被允许调用、参数是否合法、执行结果如何回填、失败如何降级。可以这样理解组件解决问题典型产出MCP工具接入协议标准化的 Tool InterfaceClaude Agent SDKAgent 开发体验Agent 主循环MHS执行安全与治理工具路由、权限检查、日志、故障隔离三者可以组合使用用 MCP 接入工具用 Claude Agent SDK 构建 Agent用 MHS 做中间层的治理能力。如果你从零开发也可以不用 MCP直接在代码里注册工具函数这并不影响理解 MHS 的核心思想。3. 动手实现一个 MHS 风格的 Agent 服务3.1 环境准备与依赖安装代码示例采用 Python 和 Anthropic 官方 SDK。先准备 Python 3.10 或 3.11 环境建议使用虚拟环境隔离依赖。在项目目录下创建requirements.txtfastapi0.115.6 uvicorn0.30.6 anthropic0.34.2 python-dotenv1.0.1安装依赖pip install -r requirements.txt还需要配置 API Key。不要把它写死在代码里。创建.env文件ANTHROPIC_API_KEYyour_api_key_here加载环境变量可以使用python-dotenv。生产环境建议由容器平台或密钥管理服务注入环境变量而不是提交到代码仓库。注意不同时间发布的 Anthropic SDK 在参数名称上可能略有差异。如果安装的是更新版本以官方文档为准。这里给出的版本组合用于说明核心链路。3.2 项目结构设计一个最小的 MHS 风格服务可以设计成两个核心文件一个负责工具注册与路由一个负责 HTTP 入口和模型调用。mhs-demo/ ├── app/ │ ├── __init__.py │ ├── main.py │ └── mhs_core.py ├── .env ├── .env.example └── requirements.txtmhs_core.py放工具注册表和执行逻辑main.py放 FastAPI 应用和 Anthropic 调用。工具数量少时这样分层足够清晰如果工具数量多再把具体工具实现拆到独立模块中。3.3 工具注册表MHS 的“路由内核”工具注册表的核心价值是让新增工具变得简单。每增加一个工具只需要注册一个名称、一个描述、一个 JSON Schema和一个函数不需要修改模型调用主逻辑。# app/mhs_core.py from dataclasses import dataclass from typing import Any, Callable, Awaitable ToolFunc Callable[..., Awaitable[dict]] dataclass class ToolSpec: name: str description: str input_schema: dict func: ToolFunc class ToolRegistry: def __init__(self): self._tools {} def register(self, spec: ToolSpec) - None: self._tools[spec.name] spec def list_tools(self) - list[dict]: return [ { name: spec.name, description: spec.description, input_schema: spec.input_schema, } for spec in self._tools.values() ] async def call(self, name: str, arguments: dict) - dict: spec self._tools.get(name) if spec is None: raise KeyError(ftool not found: {name}) try: return await spec.func(arguments) except Exception as exc: return { ok: False, error: f{type(exc).__name__}: {str(exc)}, }这个注册表做的事情很直观。list_tools输出给模型的工具定义call按名称找到工具并执行。异常被统一捕获后封装成结构化返回避免模型拿到不可解析的错误堆栈。3.4 主服务调用 Anthropic 模型并处理工具指令在main.py中创建 FastAPI 应用注册一个查询设备状态的模拟工具并通过 Anthropic 的messages.create接口发送携带工具定义的消息。# app/main.py from fastapi import FastAPI from dotenv import load_dotenv import anthropic from app.mhs_core import ToolRegistry, ToolSpec load_dotenv() app FastAPI() registry ToolRegistry() async def query_device_status(arguments: dict) - dict: device_id arguments.get(device_id) return { device_id: device_id, status: online, temperature: 32.5, last_seen: 2025-01-01T08:30:00Z, } registry.register( ToolSpec( namequery_device_status, description查询IoT设备当前状态, input_schema{ type: object, properties: { device_id: {type: string} }, required: [device_id] }, funcquery_device_status, ) ) client anthropic.AsyncAnthropic() app.post(/agent/run) async def agent_run(payload: dict): user_message payload.get(message) model_name payload.get(model, claude-3-5-sonnet-latest) response await client.messages.create( modelmodel_name, max_tokens1024, system你是一个智能助手可以使用工具查询设备信息。, toolsregistry.list_tools(), messages[ {role: user, content: user_message} ], ) tool_blocks [ block for block in response.content if block.type tool_use ] if not tool_blocks: texts [ block.text for block in response.content if block.type text ] return { stop_reason: response.stop_reason, content: .join(texts), tool_results: [], } results [] for block in tool_blocks: result await registry.call(block.name, block.input) results.append({ tool_name: block.name, tool_use_id: block.id, result: result, }) return { stop_reason: response.stop_reason, tool_results: results, note: 该响应仅展示了工具执行结果实际 Agent 还需要将 tool_result 回传给模型继续生成总结。 }这段代码里有两个关键点。第一tools参数必须是符合 Anthropic 工具定义格式的列表注册表里的list_tools()可以直接复用。第二response.stop_reason是tool_use时说明模型要调用工具。此时不能直接把内容返回给用户而要把工具结果构造成tool_result消息再调用一次模型接口让模型基于真实结果完成最终回复。3.5 启动服务并验证最小闭环启动服务uvicorn app.main:app --reload --port 8000发送测试请求curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {message:请查询设备 DEV-1001 的状态}如果配置正确响应里会看到stop_reason为tool_use并且tool_results中包含模拟的设备状态。这说明模型正确识别了意图注册表正确路由了工具调用。3.6 补全 tool_result 回填真正的 Agent 闭环上面的实现只展示了工具执行结果模型没有基于结果生成总结。要形成完整闭环需要在拿到工具结果后追加一条tool角色的消息。关键代码片段if tool_blocks: messages [ {role: user, content: user_message} ] for block in tool_blocks: result await registry.call(block.name, block.input) messages.append({ role: assistant, content: [ {type: tool_use, id: block.id, name: block.name, input: block.input} ], }) messages.append({ role: user, content: [ { type: tool_result, tool_use_id: block.id, content: str(result), } ], }) response await client.messages.create( modelmodel_name, max_tokens1024, system你是一个智能助手可以使用工具查询设备信息。, toolsregistry.list_tools(), messagesmessages, )这段代码展示了 tool_use 与 tool_result 的配对方式assistant 消息必须以原始 tool_use 结构携带tool_result 必须通过tool_use_id关联到对应工具调用。这套协议是模型继续理解执行结果的基础。4. 常见连接与路由错误按现象定位根因4.1unable to connect to anthropic services failed to connect to api.anthropic.com: status 403这个现象在 Claude API 集成过程中非常常见。严格说status 403说明请求已经到达 API 服务器但服务器拒绝了请求因此问题不在域名解析或基础网络不可达而在认证、授权或出口策略。排查时先区分两层如果连网络都无法建立会看到ConnectionError或超时。如果出现 403优先怀疑 API Key、请求头、模型访问权限。检查 API Key 是否生效curl -v https://api.anthropic.com/v1/models \ -H x-api-key: $ANTHROPIC_API_KEY \ -H anthropic-version: 2023-06-01如果返回 401或403说明凭证无效或对应账号没有访问权限。常见原因包括可能原因说明解决方向API Key 写错复制时多了空格或缺失前缀重新生成并确认环境变量账号未开通模型权限某些模型或 beta 能力需要单独申请查看账号权限页面确认模型可用区域或出口网络受限企业网络策略拦截请求联系网络管理员确认出口白名单SDK 版本过旧请求头缺少新版要求参数升级 anthropic SDK 到最新版不要在生产日志里打印完整 API Key。排查时可以用掩码方式确认前几位和后几位。4.2expected a gateway model route referenced这类模型路由错误如果你通过企业网关或模型网关访问 Anthropic而不是直连官方 API请求中的model字段需要匹配网关里配置的路由名称。常见的报错是“doesnt look like an anthropic model: expected a gateway model route reference”。翻译成工程语言就是你传的模型名在网关路由表里找不到。排查顺序查看代码或配置中model参数的值。登录网关管理端确认已发布的路由名称列表。确认请求是否经过网关。如果本应直连官方 API却被环境变量指向了网关地址也会出现不相符的情况。如果网关按模型名做转发要确认路由规则是否区分大小写。修复示例# 假设网关允许的路由名是 anthropic/claude-sonnet modelpayload.get(model, anthropic/claude-sonnet)如果直连官方 API不要随意加前缀使用官方文档给出的模型 ID。4.3 VS Code 加载 Claude Code 扩展失败这个问题属于本地开发环境范畴。很多用户在用 VS Code 加载 Claude Code 扩展时遇到无法连接或登录态丢失。建议按以下顺序处理先在终端中执行claude --version确认命令行工具已经安装。执行claude完成一次登录流程确认 API Key 或 OAuth 状态正常。回到 VS Code重新加载窗口。打开输出面板查看 Claude Code 扩展日志定位是认证失败还是网络请求失败。不要把扩展加载失败和模型调用失败混在一起。命令行能跑通扩展一般也能跑通命令行都报 403就需要先解决认证问题。4.4 错误排查速查表错误现象可能原因检查方式处理建议api.anthropic.com 返回 403API Key 无效或权限不足curl 带 key 访问 /v1/models确认 key 和模型权限连接超时或 connection error出口网络不通curl -v 查看连接阶段联系网络管理员检查出口gateway model route 报错模型名未匹配网关路由查看网关路由表修改 model 参数工具调用后模型不总结未回填 tool_result查看请求消息结构按 tool_result 格式回传工具执行报错但模型无感知异常被吞掉查看 MHS 日志和 tool_result将异常封装到 tool_result5. 生产环境里MHS 必须补上的安全与控制5.1 工具白名单宁可少注册不要全注册开发环境可以为了演示随意注册工具。生产环境里工具注册表必须是一份受控的白名单。每个工具在注册时应该附带所属业务域允许的操作范围最大可变参数是否属于危险操作是否需要对用户二次确认不要在 Agent 启动时自动扫描目录下所有函数并全部注册。模型可能因为上下文引导调用一个不该调用的工具工具越少风险面越小。5.2 工具调用必须做身份映射大模型没有用户身份。MHS 在执行工具前要把当前请求的用户身份和工具所需权限做映射。比如设备查询工具可能允许工程师调用但不允许外部访客调用。模型只负责生成工具名和参数MHS 要判断“当前用户有没有权限执行这个工具”。这个判断不能放在工具函数内部重复写。建议在注册表调用链中增加一层权限拦截器async def call_with_permission(user_id: str, tool_name: str, arguments: dict): if not await permission_service.check(user_id, tool_name): return {ok: False, error: permission denied} return await registry.call(tool_name, arguments)身份映射是 Agent 从 demo 走向生产的关键门槛。5.3 限流、审计与错误隔离MHS 需要记录每一条工具调用。最少要包含请求 ID用户 ID会话 ID工具名称输入参数执行结果耗时异常信息限流不能只做在 HTTP 入口还要做在工具级别。某些工具调用代价很高比如对外发送短信、调用第三方付费 API必须按用户、按工具设置独立配额。错误隔离方面外部工具超时不能拖垮整个 Agent。每个工具调用都要单独设置超时时间默认可以取 5 到 15 秒。对于长时间任务应该采用任务提交加结果查询的方式而不是让 HTTP 请求长期阻塞。5.4 学习环境与生产环境对照维度学习环境生产环境API Key写在 .env由密钥管理服务注入工具注册手动函数注册配置化注册带权限标识日志控制台打印集中日志平台结构化输出限流不关注按用户、按工具分别限流错误处理try-except 返回封装异常类型支持重试和降级模型名称写死默认值通过配置中心下发审计无全链路 trace6. 从 MHS 到可控 Agent最佳实践与扩展方向6.1 落地 MHS 的可复用检查清单如果你准备在自己的项目里实现类似 MHS 的中间层建议对照这份清单逐项确认。[ ] 模型返回的tool_use是否经过 schema 校验[ ] 工具注册表是否有唯一命名不允许重复注册[ ] 工具调用是否带用户身份并做了权限校验[ ] 危险操作是否有二次确认机制[ ] 每个工具是否有独立超时配置[ ] 工具执行结果是严格按tool_result协议回填吗[ ] 是否有完整的调用日志能回溯到具体用户和参数[ ] 测试环境和生产环境是否使用不同的模型路由和网关配置[ ] 工具异常是否会被模型感知而不仅是记录在日志里[ ] 新增工具时是否需要走代码评审和权限评审6.2 扩展方向从单工具到多 Agent 编排MHS 不只是单 Agent 的工具网关。当系统有多个 Agent 协作时MHS 可以升级为服务间的消息处理中枢负责把 Agent A 的输出转换成 Agent B 能理解的指令。这种场景下MHS 需要额外支持消息队列事件订阅任务状态机Agent 间上下文传递跨租户的权限隔离把协议层继续外延就是 MCP 或类似标准协议可以发挥作用的地方。MHS 关注执行治理MCP 关注接入标准两者可以共存。6.3 给开发者的一个练习建议如果你刚开始接触这个方向不要一上来就接太多外部系统。先实现一个只调用本地模拟工具的 MHS比如设备状态查询、订单号查询、时间戳格式化。把三种情况跑通模型直接返回文本没有调用工具。模型调用一个工具并且你正确回填结果。工具执行抛出异常你封装后回传模型看模型如何处理。跑完这个练习再接入真实 HTTP API。这时候你就能直观体会到AI Agent 控制现实世界的关键不是模型有多聪明而是模型和真实系统之间这层 MHS 是否把安全、路由、日志和错误处理都补齐了。把这一层做扎实Agent 才不会在 demo 和真实世界之间翻车。