Agno 动态 MCP 请求头把 Run、用户与会话身份透传给 MCP 服务器【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno在多租户、多会话的 Agent 平台上MCP 工具服务器往往需要知道“这次工具调用来自哪个用户、哪次 Run、哪个租户”才能做鉴权、审计和数据隔离。Agno 通过MCPTools的header_provider机制允许客户端在每次工具调用时基于RunContext动态生成 HTTP 请求头把身份上下文透传到远端 MCP 服务器。读完本篇你将掌握这套动态头机制的完整用法从 FastMCP 服务端如何读取身份头到 AgentOS 应用中如何声明 header provider、由 AgentOS 托管 MCP 连接生命周期并结合源码理解每个 Run 独立的会话缓存与头部合并策略。演示结构与文件说明示例位于 cookbook/91_tools/mcp/dynamic_headers/由一对文件构成分别扮演 MCP 链路两端文件说明server.py本地 FastMCP 服务器接收并打印log它收到的身份请求头client.pyAgentOS 应用其MCPTools的 header provider 会转发 run、user、session、tenant 和 agent 身份整体数据流是curl POST /agents/dynamic-header-agent/runs │ ▼ AgentOS (client.py) FastMCP 服务器 (server.py) Agent 触发 greet 工具调用 ───► 读取 X-User-ID / X-Tenant-ID / header_provider 依据 X-Agent-Name 头并打印 RunContext 生成动态请求头关键分工是AgentOS 应用拥有MCPTools的连接生命周期——工具客户端在应用启动时建立连接在关停时干净地关闭这一点在后文“AgentOS 托管的连接生命周期”一节结合源码展开。服务端FastMCP 中读取身份头server.py 用 FastMCP 定义了一个greet工具核心是演示如何从 HTTP 请求对象中取出客户端注入的身份头from fastmcp import FastMCP from fastmcp.server import Context from fastmcp.server.dependencies import get_http_request mcp FastMCP(My Server) mcp.tool async def greet(name: str, ctx: Context) - str: Greet a user with personalized information from headers. # Get the HTTP request object request get_http_request() # Access headers (lowercase!) user_id request.headers.get(x-user-id, unknown) tenant_id request.headers.get(x-tenant-id, unknown) agent_name request.headers.get(x-agent-name, unknown) print( * 60) print(fHeaders - Agent: {agent_name}, User: {user_id}, Tenant: {tenant_id}) print( * 60) return fHello, {name}! (User: {user_id}, Tenant: {tenant_id}) if __name__ __main__: mcp.run(transportstreamable-http, port8000)服务端有四个要点值得注意服务器以 Streamable HTTP 传输在 8000 端口运行mcp.run(transportstreamable-http, port8000)这是动态请求头机制唯一能生效的传输形态之一原因见后文源码分析。通过get_http_request()拿到当前请求对象。这是 FastMCP 在 HTTP 传输下暴露的请求上下文工具函数据此能访问原始 HTTP 头。HTTP 头名必须小写读取。源码注释专门标注了Access headers (lowercase!)——HTTP 协议层面头名不区分大小写但 Starlette/FastAPI 的request.headers按小写键访问若写成X-User-ID会取到None。头缺失时有降级默认值unknown使服务端在身份信息缺位时仍能工作只是无法识别调用方。客户端用 RunContext 构建动态请求头client.py 是整个示例的主体包含三部分header provider 函数、MCPTools声明、Agent 与 AgentOS 装配。header_provider把 Run 身份映射为请求头from agno.run import RunContext from agno.tools.mcp import MCPTools from typing import TYPE_CHECKING, Optional if TYPE_CHECKING: from agno.agent import Agent as AgentType from agno.team import Team as TeamType def header_provider( run_context: Optional[RunContext] None, agent: Optional[AgentType] None, team: Optional[TeamType] None, ) - dict[str, str]: Build headers for discovery or for a contextual agent or team run. entity_name ( agent.name if agent is not None else team.name if team is not None else unnamed-agno-entity ) tenant_id ( run_context.metadata.get(tenant_id, no-tenant) if run_context is not None and run_context.metadata else no-tenant ) return { X-User-ID: run_context.user_id if run_context is not None and run_context.user_id else anonymous, X-Session-ID: run_context.session_id if run_context is not None else unknown, X-Run-ID: run_context.run_id if run_context is not None else unknown, X-Tenant-ID: str(tenant_id), X-Agent-Name: entity_name, }该函数把 Agno 的运行时上下文映射成五个身份头注意每个字段都有“无上下文”时的兜底值——因为 provider 不仅在 Run 期间被调用也会在连接发现discovery/handshake阶段被调用那时可能还没有RunContext请求头取值来源缺省兜底X-User-IDrun_context.user_idanonymousX-Session-IDrun_context.session_idunknownX-Run-IDrun_context.run_idunknownX-Tenant-IDrun_context.metadata[tenant_id]Run 元数据no-tenantX-Agent-Nameagent.name或 Team 场景下的team.nameunnamed-agno-entityRunContext本身定义在 libs/agno/agno/run/base.py是一个 dataclass除上表用到的run_id、session_id、user_id外还携带metadata、session_state、dependencies等字段。示例中的租户 ID 正是从metadata中取的——这意味着通过 AgentOS 创建 Run 时租户信息可以作为 Run 元数据传入无需侵入 Agent 代码。签名中的agent和team两个参数使同一个 provider 能同时服务于 Agent 与 Teamprovider 会按 Run 实际所属实体回填其中之一。MCPTools 声明与 AgentOS 装配mcp_tools MCPTools( urlhttp://localhost:8000/mcp, transportstreamable-http, header_providerheader_provider, ) dynamic_headers_agent Agent( iddynamic-header-agent, nameDynamic Header Agent, modelOpenAIResponses(idgpt-5.5), tools[mcp_tools], ) agent_os AgentOS( descriptionAgentOS-managed MCP client with dynamic request headers, agents[dynamic_headers_agent], ) app agent_os.get_app() if __name__ __main__: agent_os.serve(appapp)MCPTools通过urltransportstreamable-http指向本地 FastMCP 服务器的/mcp端点并以header_provider启用动态头。AgentOS把 Agent 注册为dynamic-header-agentid即 URL 路径段get_app()产出 FastAPI 应用agent_os.serve(appapp)默认在 7777 端口启动 AgentOS 服务。运行示例前置条件模型调用所需的OPENAI_API_KEY环境变量仓库的 demo 虚拟环境包含fastmcp依赖启动步骤终端一启动 MCP 服务器监听 8000 端口.venvs/demo/bin/python cookbook/91_tools/mcp/dynamic_headers/server.py终端二启动 AgentOS 客户端应用监听 7777 端口.venvs/demo/bin/python cookbook/91_tools/mcp/dynamic_headers/client.py创建一次 Runcurl -X POST http://localhost:7777/agents/dynamic-header-agent/runs \ -F messageUse the greet tool to greet Ada. \ -F user_idada \ -F session_iddiscord-demo \ -F streamfalse预期结果模型调用greet工具后MCP 服务器终端一的日志会打印转发过来的身份头形如Headers - Agent: Dynamic Header Agent, User: ada, Tenant: no-tenant。tenant_id若通过 Run 元数据传入则会被转发本示例未传故回退为no-tenant。源码解析header_provider 在 MCPTools 内部如何工作以上示例对应的实现位于 libs/agno/agno/tools/mcp/mcp.py。结合源码看动态头机制有几个关键设计1. 仅支持 HTTP 系传输且与静态 headers 分工明确header_provider参数声明在构造函数签名中mcp.py 第 266 行其文档说明给出了headers与header_provider的分工headers: 适用于连接握手阶段的静态 HTTP 头……优先用于连接期鉴权 tokenheader_provider 用于每次 Run 的动态值。初始化时会做传输校验mcp.py 第 388-396 行self.header_provider None if header_provider is not None: if self.transport not in [sse, streamable-http]: raise ValueError( fheader_provider is not supported with {self.transport} transport. Use sse or streamable-http transport instead. ) self.header_provider header_provider即stdio传输会直接抛错——这也解释了为什么示例两端都使用streamable-http。2. 基于函数签名的自适应调用Agno 并不要求 provider 固定形如(run_context, agent, team)的参数签名。_call_header_providermcp.py 第 456-514 行用inspect.signature检查你的 provider 实际接受哪些参数只传它声明了的声明了run_context/agent/team中的哪些就注入哪些 kwarg带**kwargs的函数会收到全部三项上下文无识别参数但有位置参数的按旧版约定把run_context作为第一个位置实参传入向后兼容调用异常时不会中断 Run而是记录 warning 并返回空字典。这意味着示例中的header_provider可以按需删减参数例如只写def header_provider(run_context)而不破坏集成。单元测试 libs/agno/tests/unit/tools/test_mcp.py 中对header_provider的覆盖即验证了这些行为。3. 头部合并顺序_merge_http_headersmcp.py 第 516-531 行按固定优先级合并三层来源server_params自带的 headers → 构造时传入的静态headers→header_provider的输出。也就是说provider 的动态值优先级最高静态值适合放连接级凭据动态值适合放每次 Run 变化的身份。4. 按 Run 隔离的会话缓存动态头的本质诉求是“不同 Run 用不同头”而 HTTP 头在会话建立时已固定。get_session_for_runmcp.py 第 598 行起的解决方式是未设置header_provider或没有run_context时直接复用默认长连接会话设置了 provider 且有run_context时以run_context.run_id为键为每次 Run 创建独立会话并用asyncio.Lock串行化同run_id的会话创建避免并行工具调用各自建连会话带 TTL后台会清理过期 Run 的会话_cleanup_stale_sessionsmcp.py 第 533-547 行防止内存泄漏另有一条refresh_connectionTrue且同时配置 provider 时的临时会话路径should_use_temporary_run_sessionmcp.py 第 549-556 行每次工具调用即时建连、用完即关规避 anyio cancel scope 跨任务复用的问题。因此对调用方而言同一 Run 内的多次工具调用共享一个带该 Run 身份头的会话跨 Run 则自动切换——这正是示例能按 Run 区分用户身份而不需要手动管理连接的原因。5. AgentOS 托管的连接生命周期在 AgentOS 场景下MCPTools的connect()发现工具与关闭由应用生命周期驱动应用启动时 AgentOS 对注册的 MCP 工具建立连接、完成工具发现示例中即发现greet应用关停时关闭 MCP 会话。TEST_LOG.md 记录了 2026-07-24 的验证结果/health与/config返回 200、/config列出了dynamic-header-agentAgentOS 在启动时连接MCPTools并发现greet工具在关停时关闭了 MCP 会话header_provider在给定具体RunContext时产生了预期的 user、session、run、tenant、agent 五个头。该验证为构造冒烟模式CONSTRUCTION_SMOKE未实际触发 OpenAI 模型调用故最终一次带真实OPENAI_API_KEY的 MCP 工具调用未在该测试中执行。小结header_provider是MCPTools中用于注入每次 Run 动态身份头的钩子仅在sse/streamable-http传输下可用与静态headers互补后者用于连接期鉴权前者用于 Run 级身份。服务端FastMCP通过get_http_request()读取头注意键名小写与缺失兜底。每次 Run 的会话以run_id隔离并带 TTL 缓存跨 Run 自动重建AgentOS 负责连接的整体生命周期。复用该机制时把tenant_id等租户信息放进 Run 元数据、按RunContext字段派生头即可把 Agno 平台的多租户身份无缝带到任意远端 MCP 服务。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考