1. 为什么 agno 多智能体跑 MCP 时Codex 通道最容易卡住agno 是一个偏工程化的多智能体框架它把 Agent、Model、Tools 拆得很清楚其中MultiMCPTools能一次性挂载多个 MCP 服务非常适合做「一个 Agent 同时调多个外部能力」的场景。原文案例里MCP 用的是在线免费摩搭社区的两个 SSE 端点FastAPI 暴露/stream-run前端用 requests 流式读取整体链路是通的。真正让人卡住的往往不是 agno 代码而是 Codex 这一侧的模型通道。Codex 默认走的是官方通道很多人在本地想让它按原文的 agno 代码重跑 MCP 调用和 FastAPI 流式响应时会发现请求发不出去、Base URL 不知道填什么、Key 从哪来。这篇就按「接入配置」的视角把 Codex 的通道换成 TaoToken再让 Codex 去驱动 agno MCP FastAPI 这套案例跑通。适合谁看已经在用 agno 写多智能体、手里有 MCP SSE 地址、想用 Codex 当编码助手把整条链路调通的人。你需要准备的东西不多一个 TaoToken 的 Key、能跑 Python 的环境、以及原文那两个 MCP 地址换成你自己的也行。先说清楚一个容易混的点Codex 负责的是「帮你写和改代码」的模型通道agno 里的OpenAILike负责的是「Agent 实际推理」的模型通道这两个是分开的。很多人只配了其中一个结果要么 Codex 用不了要么 Agent 跑起来报鉴权错。下面两处都会给到。2. 前置准备在 TaoToken 拿 Key把 Codex 通道接上第一步是去 TaoToken 官网创建 Key。打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册登录后进控制台在 API Keys 页面新建一个 Key复制出来先存好。这个 Key 后面既给 Codex 用也可以给 agno 的OpenAILike用省得你再去别处找。创建 Key 的直达入口在这里https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keys_ctautm_campaignrewrite 进去后点新建即可。如果你还没想好要用哪个模型可以先到模型对话页面看看有哪些可用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodels_ctautm_campaignrewrite 。关键配置就一条Codex 的 Base URL 填https://taotoken.net/api注意不带/v1。这是最容易填错的地方很多人习惯性加/v1结果 404。Key 就填你刚复制的那串。填完之后 Codex 的模型请求就会走 TaoToken 的通道你在本地让它读 agno 代码、改 MCP 调用逻辑、补 FastAPI 流式处理都能正常返回。如果你打算长期用 Codex 做编码和 Agent 调试可以考虑 Coding Plan额度更划算入口https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_plan_ctautm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc_ctautm_campaignrewrite 遇到参数不确定时对着看。3. 可复制配置agno MultiMCPTools FastAPI 流式接口原文的代码能跑但有两个可以优化的点一是 MCP 连接每次请求都新建再关闭开销大二是流式输出里工具调用日志和正文混在一起前端不好区分。下面这版把 MCP 做成常驻用 FastAPI 的 startup/shutdown 事件管理生命周期同时保留流式。先装依赖pip install agno fastapi uvicorn requests然后是服务端代码保存为agnoagent_mcp_stream1.pyfrom agno.agent import Agent from agno.models.openai.like import OpenAILike from agno.tools.mcp import MultiMCPTools from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() # Agent 实际推理用的模型通道同样走 TaoToken model OpenAILike( idqwen-plus-latest, base_urlhttps://taotoken.net/api, api_key你的TaoToken Key, ) multi_mcp_tools None # 全局变量存 MCP 连接 app.on_event(startup) async def startup_event(): global multi_mcp_tools multi_mcp_tools MultiMCPTools( urls[ https://mcp.api-inference.modelscope.net/9e8a6bb57a774b/sse, https://mcp.api-inference.modelscope.net/44e33ba901cc4f/sse, ], urls_transports[sse, sse], ) await multi_mcp_tools.connect() print(MultiMCPTools 已连接) app.on_event(shutdown) async def shutdown_event(): global multi_mcp_tools if multi_mcp_tools: await multi_mcp_tools.close() print(MultiMCPTools 已关闭) app.post(/stream-run) async def stream_run(input: dict): message input.get(text, ) async def generator(): agent Agent( modelmodel, tools[multi_mcp_tools], markdownTrue, ) stream_result await agent.arun(messagemessage, streamTrue, markdownTrue) async for chunk in stream_result: if chunk.content: yield chunk.content.encode(utf-8) return StreamingResponse(generator(), media_typetext/plain) if __name__ __main__: import uvicorn uvicorn.run(agnoagent_mcp_stream1:app, host0.0.0.0, port8002, reloadTrue)这里base_url填的是https://taotoken.net/api和 Codex 那侧保持一致不带/v1。id换成你在模型对话页面看到的可用模型名即可。MCP 地址是原文那两个你换成自己的 SSE 端点也行只要urls_transports对应写sse。启动服务python agnoagent_mcp_stream1.py看到MultiMCPTools 已连接就说明两个 MCP 都挂上了这一步不通的话先别急着发请求往下看第 5 节的排查。4. 验证请求流式读取工具日志和回答正文服务起来后用 requests 发一个流式请求。原文的客户端把工具日志和正文混着打印这里加一个正则把工具调用日志单独拎出来正文按行流式输出读起来清楚很多。import requests import re url http://localhost:8002/stream-run data {text: 推荐下重庆美食} pattern re.compile(r(\w)\((.*?)\) completed in ([\d\.])s\.) buffer print( 工具调用日志 ) body_started False with requests.post(url, jsondata, streamTrue) as response: if response.status_code 200: for chunk in response.iter_content(chunk_size1024): if not chunk: continue buffer chunk.decode(utf-8) # 1. 优先处理工具日志 while True: m pattern.search(buffer) if not m: break tool, params, duration m.groups() print(f工具: {tool}, 参数: {params}, 耗时: {duration}s) buffer buffer[m.end():] # 2. 正文开始第一次出现非空 buffer 时打印提示 if buffer.strip() and not body_started: print(\n 回答正文流式, flushTrue) body_started True # 3. 按行流式输出正文 while \n in buffer: line, buffer buffer.split(\n, 1) print(line, flushTrue) # 4. 最后一行可能没有换行符 if buffer.strip(): if not body_started: print(\n 回答正文流式, flushTrue) print(buffer.strip(), flushTrue) else: print(f请求失败状态码{response.status_code})跑起来后正常会先看到类似工具: xxx(...) completed in 1.23s.的日志说明 Agent 确实调用了 MCP接着是 回答正文流式后面逐行吐出模型回答。如果只看到正文没有工具日志说明 MCP 没被触发检查一下你的提问是否真的需要外部工具。想验证模型通道本身是否通可以到模型对话页面直接发一句话试试https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodels_ctautm_campaignrewrite 。这一步能返回就说明 Key 和 Base URL 没问题问题只可能在 agno 或 MCP 侧。5. 本篇常见错排查Codex 报 404 或找不到模型九成是 Base URL 多写了/v1。正确写法是https://taotoken.net/api不带/v1。改完重启 Codex 再试。agno 报鉴权失败检查OpenAILike里的api_key是不是你复制的 TaoToken Keybase_url是不是同样不带/v1。两个通道的 Key 可以共用一个但别把 Key 写错行。MCP 连接超时或 startup 卡住先单独测 MCP 地址能不能访问SSE 端点对网络环境有要求。如果某个地址不通先去掉它只留一个确认单个能连上再加回来。urls_transports的数量要和urls一一对应写错会直接抛异常。流式输出没有内容确认agent.arun的streamTrue传了并且chunk.content有值。有些模型返回的 chunk 里 content 为空、只有 role 信息加个if chunk.content过滤就行原文这版已经处理了。工具日志正则匹配不到不同版本的 agno 打印格式可能略有差异completed in前后如果有额外字符调整正则即可。匹配不到不影响正文输出只是日志没被单独拎出来。端口被占用8002被占就换一个客户端 URL 同步改。reloadTrue在开发时方便但会重复触发 startup调试 MCP 常驻时可以先关掉。6. 把 Codex 和 agno 两条通道都固定下来整套链路跑通后建议把两个通道的配置固定成环境变量别硬编码在代码里。Codex 那侧在配置里写死 Base URLhttps://taotoken.net/api和 Keyagno 这侧用os.environ读 Key 和模型名。这样换 Key 或换模型时只改一处。如果你后面要长期跑多智能体和 Agent 调试Coding Plan 会比按量更省心https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_plan_ctautm_campaignrewrite 。接入过程中遇到参数问题直接翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc_ctautm_campaignrewrite 比在群里问快。最后留一个我踩过的坑MCP 常驻之后multi_mcp_tools是全局单例多个请求并发时 Agent 共用同一个 tools 实例一般没问题但如果你在 Agent 里改了 tools 的状态记得加锁或者每个请求单独建 Agent。原文那版每次请求新建 Agent、复用全局 MCP是相对稳的写法可以直接用。
多智能体 agno 调用 MCP，让 Codex 用 TaoToken 走通 FastAPI 流式接口
1. 为什么 agno 多智能体跑 MCP 时Codex 通道最容易卡住agno 是一个偏工程化的多智能体框架它把 Agent、Model、Tools 拆得很清楚其中MultiMCPTools能一次性挂载多个 MCP 服务非常适合做「一个 Agent 同时调多个外部能力」的场景。原文案例里MCP 用的是在线免费摩搭社区的两个 SSE 端点FastAPI 暴露/stream-run前端用 requests 流式读取整体链路是通的。真正让人卡住的往往不是 agno 代码而是 Codex 这一侧的模型通道。Codex 默认走的是官方通道很多人在本地想让它按原文的 agno 代码重跑 MCP 调用和 FastAPI 流式响应时会发现请求发不出去、Base URL 不知道填什么、Key 从哪来。这篇就按「接入配置」的视角把 Codex 的通道换成 TaoToken再让 Codex 去驱动 agno MCP FastAPI 这套案例跑通。适合谁看已经在用 agno 写多智能体、手里有 MCP SSE 地址、想用 Codex 当编码助手把整条链路调通的人。你需要准备的东西不多一个 TaoToken 的 Key、能跑 Python 的环境、以及原文那两个 MCP 地址换成你自己的也行。先说清楚一个容易混的点Codex 负责的是「帮你写和改代码」的模型通道agno 里的OpenAILike负责的是「Agent 实际推理」的模型通道这两个是分开的。很多人只配了其中一个结果要么 Codex 用不了要么 Agent 跑起来报鉴权错。下面两处都会给到。2. 前置准备在 TaoToken 拿 Key把 Codex 通道接上第一步是去 TaoToken 官网创建 Key。打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册登录后进控制台在 API Keys 页面新建一个 Key复制出来先存好。这个 Key 后面既给 Codex 用也可以给 agno 的OpenAILike用省得你再去别处找。创建 Key 的直达入口在这里https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keys_ctautm_campaignrewrite 进去后点新建即可。如果你还没想好要用哪个模型可以先到模型对话页面看看有哪些可用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodels_ctautm_campaignrewrite 。关键配置就一条Codex 的 Base URL 填https://taotoken.net/api注意不带/v1。这是最容易填错的地方很多人习惯性加/v1结果 404。Key 就填你刚复制的那串。填完之后 Codex 的模型请求就会走 TaoToken 的通道你在本地让它读 agno 代码、改 MCP 调用逻辑、补 FastAPI 流式处理都能正常返回。如果你打算长期用 Codex 做编码和 Agent 调试可以考虑 Coding Plan额度更划算入口https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_plan_ctautm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc_ctautm_campaignrewrite 遇到参数不确定时对着看。3. 可复制配置agno MultiMCPTools FastAPI 流式接口原文的代码能跑但有两个可以优化的点一是 MCP 连接每次请求都新建再关闭开销大二是流式输出里工具调用日志和正文混在一起前端不好区分。下面这版把 MCP 做成常驻用 FastAPI 的 startup/shutdown 事件管理生命周期同时保留流式。先装依赖pip install agno fastapi uvicorn requests然后是服务端代码保存为agnoagent_mcp_stream1.pyfrom agno.agent import Agent from agno.models.openai.like import OpenAILike from agno.tools.mcp import MultiMCPTools from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() # Agent 实际推理用的模型通道同样走 TaoToken model OpenAILike( idqwen-plus-latest, base_urlhttps://taotoken.net/api, api_key你的TaoToken Key, ) multi_mcp_tools None # 全局变量存 MCP 连接 app.on_event(startup) async def startup_event(): global multi_mcp_tools multi_mcp_tools MultiMCPTools( urls[ https://mcp.api-inference.modelscope.net/9e8a6bb57a774b/sse, https://mcp.api-inference.modelscope.net/44e33ba901cc4f/sse, ], urls_transports[sse, sse], ) await multi_mcp_tools.connect() print(MultiMCPTools 已连接) app.on_event(shutdown) async def shutdown_event(): global multi_mcp_tools if multi_mcp_tools: await multi_mcp_tools.close() print(MultiMCPTools 已关闭) app.post(/stream-run) async def stream_run(input: dict): message input.get(text, ) async def generator(): agent Agent( modelmodel, tools[multi_mcp_tools], markdownTrue, ) stream_result await agent.arun(messagemessage, streamTrue, markdownTrue) async for chunk in stream_result: if chunk.content: yield chunk.content.encode(utf-8) return StreamingResponse(generator(), media_typetext/plain) if __name__ __main__: import uvicorn uvicorn.run(agnoagent_mcp_stream1:app, host0.0.0.0, port8002, reloadTrue)这里base_url填的是https://taotoken.net/api和 Codex 那侧保持一致不带/v1。id换成你在模型对话页面看到的可用模型名即可。MCP 地址是原文那两个你换成自己的 SSE 端点也行只要urls_transports对应写sse。启动服务python agnoagent_mcp_stream1.py看到MultiMCPTools 已连接就说明两个 MCP 都挂上了这一步不通的话先别急着发请求往下看第 5 节的排查。4. 验证请求流式读取工具日志和回答正文服务起来后用 requests 发一个流式请求。原文的客户端把工具日志和正文混着打印这里加一个正则把工具调用日志单独拎出来正文按行流式输出读起来清楚很多。import requests import re url http://localhost:8002/stream-run data {text: 推荐下重庆美食} pattern re.compile(r(\w)\((.*?)\) completed in ([\d\.])s\.) buffer print( 工具调用日志 ) body_started False with requests.post(url, jsondata, streamTrue) as response: if response.status_code 200: for chunk in response.iter_content(chunk_size1024): if not chunk: continue buffer chunk.decode(utf-8) # 1. 优先处理工具日志 while True: m pattern.search(buffer) if not m: break tool, params, duration m.groups() print(f工具: {tool}, 参数: {params}, 耗时: {duration}s) buffer buffer[m.end():] # 2. 正文开始第一次出现非空 buffer 时打印提示 if buffer.strip() and not body_started: print(\n 回答正文流式, flushTrue) body_started True # 3. 按行流式输出正文 while \n in buffer: line, buffer buffer.split(\n, 1) print(line, flushTrue) # 4. 最后一行可能没有换行符 if buffer.strip(): if not body_started: print(\n 回答正文流式, flushTrue) print(buffer.strip(), flushTrue) else: print(f请求失败状态码{response.status_code})跑起来后正常会先看到类似工具: xxx(...) completed in 1.23s.的日志说明 Agent 确实调用了 MCP接着是 回答正文流式后面逐行吐出模型回答。如果只看到正文没有工具日志说明 MCP 没被触发检查一下你的提问是否真的需要外部工具。想验证模型通道本身是否通可以到模型对话页面直接发一句话试试https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodels_ctautm_campaignrewrite 。这一步能返回就说明 Key 和 Base URL 没问题问题只可能在 agno 或 MCP 侧。5. 本篇常见错排查Codex 报 404 或找不到模型九成是 Base URL 多写了/v1。正确写法是https://taotoken.net/api不带/v1。改完重启 Codex 再试。agno 报鉴权失败检查OpenAILike里的api_key是不是你复制的 TaoToken Keybase_url是不是同样不带/v1。两个通道的 Key 可以共用一个但别把 Key 写错行。MCP 连接超时或 startup 卡住先单独测 MCP 地址能不能访问SSE 端点对网络环境有要求。如果某个地址不通先去掉它只留一个确认单个能连上再加回来。urls_transports的数量要和urls一一对应写错会直接抛异常。流式输出没有内容确认agent.arun的streamTrue传了并且chunk.content有值。有些模型返回的 chunk 里 content 为空、只有 role 信息加个if chunk.content过滤就行原文这版已经处理了。工具日志正则匹配不到不同版本的 agno 打印格式可能略有差异completed in前后如果有额外字符调整正则即可。匹配不到不影响正文输出只是日志没被单独拎出来。端口被占用8002被占就换一个客户端 URL 同步改。reloadTrue在开发时方便但会重复触发 startup调试 MCP 常驻时可以先关掉。6. 把 Codex 和 agno 两条通道都固定下来整套链路跑通后建议把两个通道的配置固定成环境变量别硬编码在代码里。Codex 那侧在配置里写死 Base URLhttps://taotoken.net/api和 Keyagno 这侧用os.environ读 Key 和模型名。这样换 Key 或换模型时只改一处。如果你后面要长期跑多智能体和 Agent 调试Coding Plan 会比按量更省心https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_plan_ctautm_campaignrewrite 。接入过程中遇到参数问题直接翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdoc_ctautm_campaignrewrite 比在群里问快。最后留一个我踩过的坑MCP 常驻之后multi_mcp_tools是全局单例多个请求并发时 Agent 共用同一个 tools 实例一般没问题但如果你在 Agent 里改了 tools 的状态记得加锁或者每个请求单独建 Agent。原文那版每次请求新建 Agent、复用全局 MCP是相对稳的写法可以直接用。