尧图网站建设 
CAI 的 MCP 工具互操作层MCPUtil完全解析从 MCP 服务器到 FunctionTool 的转换、调用与容错机制【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/caiCAICybersecurity AI在 MCP 支持 的基础上通过cai.sdk.agents.mcp.util模块中的MCPUtil类实现了 MCPModel Context Protocol与 CAI 原生工具体系之间的完整桥接将任意 MCP 服务器暴露的工具批量转换为 CAI 的FunctionTool并负责调用执行、结果格式化与异常容错。本文以 docs/ref/mcp/util.md 所指向的模块为核心结合源码实现与测试用例逐层拆解这一互操作层的设计帮助读者理解 CAI 中 MCP 工具从枚举到执行再到追踪的完整链路并掌握多服务器聚合、重连恢复与错误处理等关键实战细节。一、MCPUtil 在整个 CAI MCP 架构中的位置在 CAI 中MCP 能力分为两个层面连接层由 src/cai/sdk/agents/mcp/server.py 中的MCPServer抽象基类、MCPServerStdiostdio 传输与MCPServerSseHTTP SSE 传输负责封装了connect()、list_tools()、call_tool()、cleanup()等会话级原语并通过cache_tools_list参数支持工具列表缓存以降低延迟。互操作层由 src/cai/sdk/agents/mcp/util.py 中的MCPUtil承担把连接层拿到的原生 MCP 工具对象转换为 CAI 的FunctionTool从而让模型执行循环Runner以统一的方式调度 MCP 工具与本地函数工具。从模块导出关系看src/cai/sdk/agents/mcp/init.py 在try/except ImportError中导出MCPServer、MCPServerSse、MCPServerSseParams、MCPServerStdio、MCPServerStdioParams并无条件导出MCPUtil说明互操作层是整个 MCP 子包对外承诺的核心能力之一。二、MCPUtil 类结构总览MCPUtil是一个只包含类方法classmethod的工具类源码注释将其定义为 Set of utilities for interop between MCP and CAI tools。全部方法如下方法作用get_all_function_tools(servers)聚合多个 MCP 服务器的全部工具并检测跨服务器的工具重名get_function_tools(server)枚举单个 MCP 服务器的工具并转换为FunctionTool同时记录追踪 spanto_function_tool(tool, server)将单个 MCP 工具对象转换为 CAI 的FunctionToolinvoke_mcp_tool(server, tool, context, input_json)执行 MCP 工具调用包含 JSON 校验、会话重连、异常分级处理_format_tool_result(result, tool, server)将 MCP 的CallToolResult内容列表格式化为字符串并写入当前 span 数据其中invoke_mcp_tool通过functools.partial绑定到每个转换后的FunctionTool上作为其on_invoke_tool回调——这是两个工具体系衔接的关键设计。三、批量转换get_all_function_tools 与多服务器聚合get_all_function_tools是 Agent 获取 MCP 工具的统一入口。源码实现src/cai/sdk/agents/mcp/util.py依次遍历传入的servers列表对每个服务器调用get_function_tools同时维护一个已见工具名集合classmethod async def get_all_function_tools(cls, servers: list[MCPServer]) - list[Tool]: tools [] tool_names: set[str] set() for server in servers: server_tools await cls.get_function_tools(server) server_tool_names {tool.name for tool in server_tools} if len(server_tool_names tool_names) 0: raise UserError( fDuplicate tool names found across MCP servers: f{server_tool_names tool_names} ) tool_names.update(server_tool_names) tools.extend(server_tools) return tools两个关键行为值得注意重名检测如果两个 MCP 服务器暴露了同名的工具会直接抛出UserError。这是因为 LLM 工具调度依赖工具名唯一性重名会导致调用歧义。该行为在 tests/mcp/test_mcp_util.py 中有直接测试覆盖test_get_all_function_tools使用三个FakeMCPServer来自tests/helpers共注册 5 个工具断言转换后工具数量、名称与params_json_schema均与预期一致。顺序保序结果按服务器列表顺序拼接同一个服务器内的工具顺序由list_tools()返回顺序决定。该方法是 Agent 获取工具链的必经之路。在 src/cai/sdk/agents/agent.py 中async def get_mcp_tools(self) - list[Tool]: Fetches the available tools from the MCP servers. return await MCPUtil.get_all_function_tools(self.mcp_servers) async def get_all_tools(self) - list[Tool]: All agent tools, including MCP tools and function tools. mcp_tools await self.get_mcp_tools() return mcp_tools self.tools也就是说每次运行前 Runner 都会把 Agent 配置的mcp_servers中的工具动态并入工具集与self.tools中的本地函数工具并列参与模型工具选择。四、单服务器枚举get_function_tools 与追踪埋点get_function_toolssrc/cai/sdk/agents/mcp/util.py负责单个服务器的枚举并在外层包了一个追踪 spanwith mcp_tools_span(serverserver.name) as span: tools await server.list_tools() span.span_data.result [tool.name for tool in tools] return [cls.to_function_tool(tool, server) for tool in tools]这里使用的mcp_tools_span定义于 src/cai/sdk/agents/tracing/create.py其 span 数据类型为MCPListToolsSpanData用于记录某个 MCP 服务器被枚举了哪些工具这一过程。这为后续在追踪系统中排查工具是否成功加载、来自哪个服务器提供了可观测性依据。五、核心转换to_function_tool 的字段映射to_function_toolsrc/cai/sdk/agents/mcp/util.py是互操作层的核心映射逻辑classmethod def to_function_tool(cls, tool: MCPTool, server: MCPServer) - FunctionTool: invoke_func functools.partial(cls.invoke_mcp_tool, server, tool) return FunctionTool( nametool.name, descriptiontool.description or , params_json_schematool.inputSchema, on_invoke_toolinvoke_func, strict_json_schemaFalse, )映射关系清晰name→ 直接采用 MCP 工具名这也是重名检测的意义所在description→ 采用 MCP 工具描述为空时兜底为空字符串params_json_schema→ 采用 MCP 工具声明的inputSchemaJSON Schema 格式on_invoke_tool→ 通过functools.partial预先绑定服务器与工具对象使每次调用都携带上下文strict_json_schemaFalse→ 显式关闭严格 JSON Schema 校验兼容 MCP 生态中宽松声明的工具。由此模型侧无需感知工具来自 MCP 还是本地函数调度层面完全统一。六、调用执行invoke_mcp_tool 的完整容错链路invoke_mcp_toolsrc/cai/sdk/agents/mcp/util.py是最复杂的部分承担输入解析、会话检查、断线重连、异常分级与重试。整体流程如下6.1 输入 JSON 解析json_data: dict[str, Any] json.loads(input_json) if input_json else {}若模型产出的input_json非法则记录 debug 日志并抛出ModelBehaviorError视为模型行为异常而非基础设施故障。日志细节受_debug.DONT_LOG_TOOL_DATA开关控制开启时只记录工具名、不记录输入载荷用于敏感环境下的脱敏。6.2 会话有效性检查与主动重连调用前先检查服务器对象上是否存在session且非None若会话丢失例如进程重启或连接被远端关闭会主动调用server.connect()重连成功后继续调用重连失败则抛出AgentsException并提示用户移除并重新加载该 MCP 服务器。6.3 异常分级处理call_tool抛出的异常被分类处理src/cai/sdk/agents/mcp/util.py异常情形判定依据处理策略服务器对象未正确初始化AttributeError抛出AgentsException提示使用/mcp remove/mcp load重连连接被关闭ClosedResourceError、ExceptionGroup或错误信息含closedresourceerror/taskgroup类型名 错误文本清除旧 session、静默重连并重试一次工具调用重连失败则抛出带修复指令的AgentsException错误信息含session/connection/closed等关键词错误文本抛出AgentsException提示用/mcp status检查健康状态其他一般错误兜底抛出带完整异常类型与信息的AgentsException这一设计把可恢复的瞬时连接故障与不可恢复的调用错误区分开前者自动重连重试后者直接上抛给上层处理。在 tests/mcp/test_mcp_util.py 中test_mcp_invocation_crash_causes_error通过一个CrashingFakeMCPServercall_tool直接抛异常验证了崩溃会被包装为AgentsException且日志包含 Error invoking MCP tool test_tool_1。6.4 结果格式化与追踪回写调用成功后进入_format_tool_resultsrc/cai/sdk/agents/mcp/util.py。MCP 的CallToolResult.content是内容项列表而 CAI 工具输出约定为单一字符串因此需要转换内容项只有 1 个取result.content[0].model_dump_json()内容项多个序列化为 JSON 数组字符串内容为空记录错误日志并返回Error running tool.。随后若当前追踪中存在FunctionSpanData会将输出字符串写入span_data.output并把mcp_data记为{server: server.name}从而在追踪系统中可以溯源该输出来自哪个 MCP 服务器。关于 MCP 追踪体系的整体设计可进一步阅读 MCP 追踪文档。七、REPL 层的扩展GlobalMCPUtil除 SDK 外REPL 命令层还基于MCPUtil派生了GlobalMCPUtil见 src/cai/repl/commands/mcp.py其注释说明它uses global registry——即工具转换时绑定的是全局服务器注册表而非单个实例从而支持 REPL 中/mcp load、/mcp remove等命令动态增删服务器后工具仍能通过全局注册表找到正确的会话。这印证了to_function_tool是一个可被继承覆写的扩展点不同的上层环境SDK 单实例 vs REPL 全局注册表可以注入不同的服务器解析策略。八、测试覆盖与可靠性保障互操作层的核心行为均有自动化测试保障tests/mcp/test_mcp_util.pytest_get_all_function_tools验证多服务器聚合、数量、名称与 JSON Schema 的完整传递test_invoke_mcp_tool验证空输入场景下调用不崩溃test_mcp_invoke_bad_json_errors验证非法 JSON 会抛ModelBehaviorError并记录日志test_mcp_invocation_crash_causes_error验证服务器崩溃会被包装为AgentsException。配合tests/helpers.py中的FakeMCPServer内存模拟无需真实启动外部 MCP 进程即可对互操作层做单元级验证这也为开发者贡献新的容错分支提供了现成的测试范式。九、实战要点与建议命名规划先行由于get_all_function_tools会跨服务器做重名检测接入多个 MCP 服务器时应在服务器端或通过别名避免工具名冲突否则 Agent 运行会直接报UserError。善用工具列表缓存连接层server.py的cache_tools_listTrue可避免每次运行都向服务器发起list_tools往返对工具集固定的服务器能显著降低启动延迟工具集可能动态变化时保持False或调用invalidate_tools_cache()手动失效缓存。连接故障先自查会话invoke_mcp_tool已内置会话丢失自动重连 一次重试逻辑若仍报AgentsException且提示重载服务器应按提示执行 REPL 的/mcp remove name与/mcp load ...恢复。敏感环境开启脱敏日志设置_debug.DONT_LOG_TOOL_DATA后工具输入输出载荷将不被写入日志仅记录工具名适合处理敏感数据的网络安全场景。通过MCPUtil这层互操作设计CAI 将外部 MCP 生态的工具无缝接入自身的 Agent 工具调度与追踪体系既保持了工具调用的统一抽象又通过重连、重试与异常分级保证了跨进程、跨网络工具调用的健壮性。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 CAI 的 MCP 工具互操作层（MCPUtil）完全解析：从 MCP 服务器到 FunctionTool 的转换、调用与容错机制
CAI 的 MCP 工具互操作层MCPUtil完全解析从 MCP 服务器到 FunctionTool 的转换、调用与容错机制【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/caiCAICybersecurity AI在 MCP 支持 的基础上通过cai.sdk.agents.mcp.util模块中的MCPUtil类实现了 MCPModel Context Protocol与 CAI 原生工具体系之间的完整桥接将任意 MCP 服务器暴露的工具批量转换为 CAI 的FunctionTool并负责调用执行、结果格式化与异常容错。本文以 docs/ref/mcp/util.md 所指向的模块为核心结合源码实现与测试用例逐层拆解这一互操作层的设计帮助读者理解 CAI 中 MCP 工具从枚举到执行再到追踪的完整链路并掌握多服务器聚合、重连恢复与错误处理等关键实战细节。一、MCPUtil 在整个 CAI MCP 架构中的位置在 CAI 中MCP 能力分为两个层面连接层由 src/cai/sdk/agents/mcp/server.py 中的MCPServer抽象基类、MCPServerStdiostdio 传输与MCPServerSseHTTP SSE 传输负责封装了connect()、list_tools()、call_tool()、cleanup()等会话级原语并通过cache_tools_list参数支持工具列表缓存以降低延迟。互操作层由 src/cai/sdk/agents/mcp/util.py 中的MCPUtil承担把连接层拿到的原生 MCP 工具对象转换为 CAI 的FunctionTool从而让模型执行循环Runner以统一的方式调度 MCP 工具与本地函数工具。从模块导出关系看src/cai/sdk/agents/mcp/init.py 在try/except ImportError中导出MCPServer、MCPServerSse、MCPServerSseParams、MCPServerStdio、MCPServerStdioParams并无条件导出MCPUtil说明互操作层是整个 MCP 子包对外承诺的核心能力之一。二、MCPUtil 类结构总览MCPUtil是一个只包含类方法classmethod的工具类源码注释将其定义为 Set of utilities for interop between MCP and CAI tools。全部方法如下方法作用get_all_function_tools(servers)聚合多个 MCP 服务器的全部工具并检测跨服务器的工具重名get_function_tools(server)枚举单个 MCP 服务器的工具并转换为FunctionTool同时记录追踪 spanto_function_tool(tool, server)将单个 MCP 工具对象转换为 CAI 的FunctionToolinvoke_mcp_tool(server, tool, context, input_json)执行 MCP 工具调用包含 JSON 校验、会话重连、异常分级处理_format_tool_result(result, tool, server)将 MCP 的CallToolResult内容列表格式化为字符串并写入当前 span 数据其中invoke_mcp_tool通过functools.partial绑定到每个转换后的FunctionTool上作为其on_invoke_tool回调——这是两个工具体系衔接的关键设计。三、批量转换get_all_function_tools 与多服务器聚合get_all_function_tools是 Agent 获取 MCP 工具的统一入口。源码实现src/cai/sdk/agents/mcp/util.py依次遍历传入的servers列表对每个服务器调用get_function_tools同时维护一个已见工具名集合classmethod async def get_all_function_tools(cls, servers: list[MCPServer]) - list[Tool]: tools [] tool_names: set[str] set() for server in servers: server_tools await cls.get_function_tools(server) server_tool_names {tool.name for tool in server_tools} if len(server_tool_names tool_names) 0: raise UserError( fDuplicate tool names found across MCP servers: f{server_tool_names tool_names} ) tool_names.update(server_tool_names) tools.extend(server_tools) return tools两个关键行为值得注意重名检测如果两个 MCP 服务器暴露了同名的工具会直接抛出UserError。这是因为 LLM 工具调度依赖工具名唯一性重名会导致调用歧义。该行为在 tests/mcp/test_mcp_util.py 中有直接测试覆盖test_get_all_function_tools使用三个FakeMCPServer来自tests/helpers共注册 5 个工具断言转换后工具数量、名称与params_json_schema均与预期一致。顺序保序结果按服务器列表顺序拼接同一个服务器内的工具顺序由list_tools()返回顺序决定。该方法是 Agent 获取工具链的必经之路。在 src/cai/sdk/agents/agent.py 中async def get_mcp_tools(self) - list[Tool]: Fetches the available tools from the MCP servers. return await MCPUtil.get_all_function_tools(self.mcp_servers) async def get_all_tools(self) - list[Tool]: All agent tools, including MCP tools and function tools. mcp_tools await self.get_mcp_tools() return mcp_tools self.tools也就是说每次运行前 Runner 都会把 Agent 配置的mcp_servers中的工具动态并入工具集与self.tools中的本地函数工具并列参与模型工具选择。四、单服务器枚举get_function_tools 与追踪埋点get_function_toolssrc/cai/sdk/agents/mcp/util.py负责单个服务器的枚举并在外层包了一个追踪 spanwith mcp_tools_span(serverserver.name) as span: tools await server.list_tools() span.span_data.result [tool.name for tool in tools] return [cls.to_function_tool(tool, server) for tool in tools]这里使用的mcp_tools_span定义于 src/cai/sdk/agents/tracing/create.py其 span 数据类型为MCPListToolsSpanData用于记录某个 MCP 服务器被枚举了哪些工具这一过程。这为后续在追踪系统中排查工具是否成功加载、来自哪个服务器提供了可观测性依据。五、核心转换to_function_tool 的字段映射to_function_toolsrc/cai/sdk/agents/mcp/util.py是互操作层的核心映射逻辑classmethod def to_function_tool(cls, tool: MCPTool, server: MCPServer) - FunctionTool: invoke_func functools.partial(cls.invoke_mcp_tool, server, tool) return FunctionTool( nametool.name, descriptiontool.description or , params_json_schematool.inputSchema, on_invoke_toolinvoke_func, strict_json_schemaFalse, )映射关系清晰name→ 直接采用 MCP 工具名这也是重名检测的意义所在description→ 采用 MCP 工具描述为空时兜底为空字符串params_json_schema→ 采用 MCP 工具声明的inputSchemaJSON Schema 格式on_invoke_tool→ 通过functools.partial预先绑定服务器与工具对象使每次调用都携带上下文strict_json_schemaFalse→ 显式关闭严格 JSON Schema 校验兼容 MCP 生态中宽松声明的工具。由此模型侧无需感知工具来自 MCP 还是本地函数调度层面完全统一。六、调用执行invoke_mcp_tool 的完整容错链路invoke_mcp_toolsrc/cai/sdk/agents/mcp/util.py是最复杂的部分承担输入解析、会话检查、断线重连、异常分级与重试。整体流程如下6.1 输入 JSON 解析json_data: dict[str, Any] json.loads(input_json) if input_json else {}若模型产出的input_json非法则记录 debug 日志并抛出ModelBehaviorError视为模型行为异常而非基础设施故障。日志细节受_debug.DONT_LOG_TOOL_DATA开关控制开启时只记录工具名、不记录输入载荷用于敏感环境下的脱敏。6.2 会话有效性检查与主动重连调用前先检查服务器对象上是否存在session且非None若会话丢失例如进程重启或连接被远端关闭会主动调用server.connect()重连成功后继续调用重连失败则抛出AgentsException并提示用户移除并重新加载该 MCP 服务器。6.3 异常分级处理call_tool抛出的异常被分类处理src/cai/sdk/agents/mcp/util.py异常情形判定依据处理策略服务器对象未正确初始化AttributeError抛出AgentsException提示使用/mcp remove/mcp load重连连接被关闭ClosedResourceError、ExceptionGroup或错误信息含closedresourceerror/taskgroup类型名 错误文本清除旧 session、静默重连并重试一次工具调用重连失败则抛出带修复指令的AgentsException错误信息含session/connection/closed等关键词错误文本抛出AgentsException提示用/mcp status检查健康状态其他一般错误兜底抛出带完整异常类型与信息的AgentsException这一设计把可恢复的瞬时连接故障与不可恢复的调用错误区分开前者自动重连重试后者直接上抛给上层处理。在 tests/mcp/test_mcp_util.py 中test_mcp_invocation_crash_causes_error通过一个CrashingFakeMCPServercall_tool直接抛异常验证了崩溃会被包装为AgentsException且日志包含 Error invoking MCP tool test_tool_1。6.4 结果格式化与追踪回写调用成功后进入_format_tool_resultsrc/cai/sdk/agents/mcp/util.py。MCP 的CallToolResult.content是内容项列表而 CAI 工具输出约定为单一字符串因此需要转换内容项只有 1 个取result.content[0].model_dump_json()内容项多个序列化为 JSON 数组字符串内容为空记录错误日志并返回Error running tool.。随后若当前追踪中存在FunctionSpanData会将输出字符串写入span_data.output并把mcp_data记为{server: server.name}从而在追踪系统中可以溯源该输出来自哪个 MCP 服务器。关于 MCP 追踪体系的整体设计可进一步阅读 MCP 追踪文档。七、REPL 层的扩展GlobalMCPUtil除 SDK 外REPL 命令层还基于MCPUtil派生了GlobalMCPUtil见 src/cai/repl/commands/mcp.py其注释说明它uses global registry——即工具转换时绑定的是全局服务器注册表而非单个实例从而支持 REPL 中/mcp load、/mcp remove等命令动态增删服务器后工具仍能通过全局注册表找到正确的会话。这印证了to_function_tool是一个可被继承覆写的扩展点不同的上层环境SDK 单实例 vs REPL 全局注册表可以注入不同的服务器解析策略。八、测试覆盖与可靠性保障互操作层的核心行为均有自动化测试保障tests/mcp/test_mcp_util.pytest_get_all_function_tools验证多服务器聚合、数量、名称与 JSON Schema 的完整传递test_invoke_mcp_tool验证空输入场景下调用不崩溃test_mcp_invoke_bad_json_errors验证非法 JSON 会抛ModelBehaviorError并记录日志test_mcp_invocation_crash_causes_error验证服务器崩溃会被包装为AgentsException。配合tests/helpers.py中的FakeMCPServer内存模拟无需真实启动外部 MCP 进程即可对互操作层做单元级验证这也为开发者贡献新的容错分支提供了现成的测试范式。九、实战要点与建议命名规划先行由于get_all_function_tools会跨服务器做重名检测接入多个 MCP 服务器时应在服务器端或通过别名避免工具名冲突否则 Agent 运行会直接报UserError。善用工具列表缓存连接层server.py的cache_tools_listTrue可避免每次运行都向服务器发起list_tools往返对工具集固定的服务器能显著降低启动延迟工具集可能动态变化时保持False或调用invalidate_tools_cache()手动失效缓存。连接故障先自查会话invoke_mcp_tool已内置会话丢失自动重连 一次重试逻辑若仍报AgentsException且提示重载服务器应按提示执行 REPL 的/mcp remove name与/mcp load ...恢复。敏感环境开启脱敏日志设置_debug.DONT_LOG_TOOL_DATA后工具输入输出载荷将不被写入日志仅记录工具名适合处理敏感数据的网络安全场景。通过MCPUtil这层互操作设计CAI 将外部 MCP 生态的工具无缝接入自身的 Agent 工具调度与追踪体系既保持了工具调用的统一抽象又通过重连、重试与异常分级保证了跨进程、跨网络工具调用的健壮性。【免费下载链接】caiCybersecurity AI (CAI), the framework for AI Security项目地址: https://gitcode.com/GitHub_Trending/cai3/cai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考