openai-agents-python 中 MCP StreamableHTTP 自定义 HTTP 客户端：httpx_client_factory 实战指南

openai-agents-python 中 MCP StreamableHTTP 自定义 HTTP 客户端httpx_client_factory 实战指南【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python导读本文基于仓库 examples/mcp/streamablehttp_custom_client_example 中的示例系统讲解如何在 openai-agents-pythonAgents SDK中通过MCPServerStreamableHttp的httpx_client_factory参数为 MCP StreamableHTTP 连接注入自定义 HTTP 客户端行为——包括自定义 SSL 证书与校验、请求头、超时、代理与重试配置。读完本文你将掌握工厂函数的正确签名、SDK 底层调用链、MCP v1httpx与 MCP v2httpx2的兼容性差异并能直接落地到企业网络、安全认证、调试等真实场景中。背景为什么需要自定义 HTTP 客户端MCPModel Context Protocol的 Streamable HTTP 传输方式通过 HTTP 长连接与服务器通信。默认情况下Agents SDK 使用内置的_create_default_streamable_http_client创建客户端其行为以follow_redirectsFalse为基础见 src/agents/mcp/server.py对大多数场景够用但以下需求无法满足企业内网要求走 HTTP 代理内部 CA 签发的证书需要自定义 SSL 校验需要对每个请求附加认证头如 API Key需要更长的读超时或更精细的连接池配置开发阶段需要临时关闭 SSL 校验便于调试。httpx_client_factory正是为此提供的注入点你传入一个工厂函数SDK 在建立 MCP StreamableHTTP 连接时调用它来构造实际的 HTTP 客户端。注意本示例面向 MCP Python SDK v2 与httpx2并依赖仓库锁定的开发环境运行。Agents SDK 的客户端本身同时支持 MCP v1httpx与 MCP v2httpx2工厂函数必须返回与已安装 MCP SDK 版本匹配的客户端类型。环境准备与运行示例示例运行依赖uv仓库锁定的依赖管理工具安装方式见官方 uv 文档。进入示例目录后直接运行cd examples/mcp/streamablehttp_custom_client_example uv run main.py运行流程在 main.py 中自动完成检测uv是否已安装未安装则报错提示通过子进程执行uv run server.py在本地启动一个 Streamable HTTP 服务器真实场景中通常是远程服务器这里仅为演示等待 3 秒后运行 Agent调用 MCP 工具add计算7 22并打印结果结束时自动终止服务器子进程。服务器端 server.py 定义了一个Echo Server暴露两个工具add(a, b)和get_secret_word()并以streamable-http传输方式在http://host:port/mcp提供服务。端口与绑定地址控制示例会自动选择一个空闲的本地端口也可以通过环境变量显式控制STREAMABLE_HTTP_PORT指定服务器端口不设置时自动挑选空闲端口逻辑见 main.py 的_choose_portSTREAMABLE_HTTP_HOST指定绑定地址默认127.0.0.1。核心 APIMCPServerStreamableHttp 与 httpx_client_factoryMCPServerStreamableHttp是 Agents SDK 中基于 Streamable HTTP 传输的 MCP 服务器封装见 src/agents/mcp/server.py。它的params参数类型为MCPServerStreamableHttpParams见 src/agents/mcp/server.py各字段如下参数类型说明urlstr必填服务器地址headersdict[str, str]可选发送到服务器的请求头timeouttimedelta \| float可选HTTP 请求超时默认 5 秒sse_read_timeouttimedelta \| float可选SSE 连接读超时默认 5 分钟terminate_on_closebool可选关闭时是否终止连接httpx_client_factoryHttpClientFactory可选自定义 HTTP 客户端工厂MCP v1 返回httpx.AsyncClientMCP v2 返回httpx2.AsyncClientauthAny可选认证处理器MCP v1 用httpx.AuthMCP v2 用httpx2.Authignore_initialized_notification_failurebool可选是否忽略 initialized 通知失败其中HttpClientFactory是一个Protocol见 src/agents/mcp/util.py其核心约束是工厂必须使用与已安装 MCP SDK 匹配的 HTTP 栈——MCP v1 用httpxMCP v2 用httpx2。工厂函数签名与底层调用链自定义工厂并非无参函数SDK 在创建连接时会主动传入三个关键字参数。以 MCP v2 为例_streamablehttp_client_v2中调用工厂的方式如下见 src/agents/mcp/server.pyclient factory( headersheaders, timeoutMCP_HTTPX.Timeout(timeout_seconds, readsse_read_timeout_seconds), authauth, )因此自定义工厂应当接受headers、timeout、auth三个可选参数再叠加自己的定制项。完整调用链为MCPServerStreamableHttp.create_streams()读取params把httpx_client_factory与默认工厂合并见 src/agents/mcp/server.pyMCP v2 分支中_validated_v2_http_client_factory对工厂做包装校验见 src/agents/mcp/server.py校验auth必须是httpx2.Auth实例否则抛出UserError见_validate_v2_http_auth调用工厂后校验返回值必须是httpx2.AsyncClient实例否则抛出UserErrorMCP Python SDK v2 requires httpx_client_factory to return an httpx2.AsyncClient. Use an httpx2 factory or pin mcp2.连接建立后_configure_v2_session_id_hook见 src/agents/mcp/server.py向客户端的event_hooks[response]注册回调捕获initialize响应头中的mcp-session-id用于会话管理整个生命周期内你自定义的客户端实例被用于所有 MCP 请求。这意味着你的工厂返回值会被严格类型校验——这是 MCP v2 下最常见的报错来源务必返回httpx2.AsyncClient而不是httpx.AsyncClient。代码实战从最小示例到完整实现最小自定义客户端README 原始示例import httpx2 from agents.mcp import MCPServerStreamableHttp def create_custom_http_client() - httpx2.AsyncClient: return httpx2.AsyncClient( verifyFalse, # Disable SSL verification for testing timeouthttpx2.Timeout(60.0, read120.0), headers{X-Custom-Client: my-app}, ) async with MCPServerStreamableHttp( nameCustom Client Server, params{ url: http://localhost:port/mcp, httpx_client_factory: create_custom_http_client, }, ) as server: # Use the server...完整示例仓库 main.py 的增强版main.py 给出了更贴合 SDK 调用链的写法——工厂接收 SDK 传入的headers、timeout、auth并在未提供时填入默认值def create_custom_http_client( headers: dict[str, str] | None None, timeout: httpx2.Timeout | None None, auth: httpx2.Auth | None None, ) - httpx2.AsyncClient: if headers is None: headers { X-Custom-Client: agents-mcp-example, User-Agent: OpenAI-Agents-MCP/1.0, } if timeout is None: timeout httpx2.Timeout(60.0, read120.0) if auth is None: auth None return httpx2.AsyncClient( # Disable SSL verification for testing (not recommended for production) verifyFalse, # Set custom timeout timeouthttpx2.Timeout(60.0, read120.0), # Add custom headers that will be sent with every request headersheaders, )随后通过MCPServerStreamableHttp挂载到 Agent 上运行见 main.pyasync with MCPServerStreamableHttp( nameStreamable HTTP with Custom Client, params{ url: STREAMABLE_HTTP_URL, httpx_client_factory: create_custom_http_client, }, ) as server: agent Agent( nameAssistant, instructionsUse the tools to answer the questions., mcp_servers[server], model_settingsModelSettings(tool_choicerequired), ) result await Runner.run(starting_agentagent, inputAdd these numbers: 7 and 22.) print(result.final_output)示例还通过gen_trace_id()与trace(workflow_nameCustom HTTP Client Example)开启追踪输出可查看的 trace 链接便于观察 MCP 工具调用过程。五大自定义能力详解示例 README 强调的五个能力全部落在httpx2.AsyncClient的构造参数上自定义 SSL 配置通过verify参数。可传入证书文件路径或 CA 包来信任私有 CA也可verifyFalse临时关闭校验仅限测试环境生产不建议自定义请求头通过headers参数附加到所有请求适合 API 认证、自定义 User-Agent 等自定义超时通过timeout参数。示例中httpx2.Timeout(60.0, read120.0)表示连接/写超时 60 秒、读超时 120 秒适配慢速网络或大响应代理配置通过proxy/mounts等 httpx2 参数配置 HTTP 代理示例中以注释形式预留自定义重试逻辑通过 httpx 的传输层配置如transport、重试传输实现请求级重试。MCP v1 与 v2 的兼容性要点Agents SDK 客户端同时支持两代 MCP SDK这是使用httpx_client_factory时必须注意的差异点MCP v1MCP v2HTTP 栈httpxhttpx2工厂返回类型httpx.AsyncClienthttpx2.AsyncClient认证类型httpx.Authhttpx2.Auth不匹配时的行为—抛UserError并提示 Use an httpx2 factory or pin mcp2.如果你在 MCP v2 环境下误用了基于httpx的工厂SDK 会在连接建立前通过 src/agents/mcp/server.py 的类型校验直接抛错而不是在运行时产生难以排查的网络异常。另外MCPServerSseHTTP SSE 传输见 src/agents/mcp/server.py的MCPServerSseParams同样支持httpx_client_factory与auth见 src/agents/mcp/server.py上述自定义模式可平滑迁移到 SSE 场景。适用场景与收益典型使用场景企业网络为受代理保护的网络环境配置代理与连接池SSL/TLS 需求使用内部 CA 证书或自定义证书完成双向安全连接自定义认证附加请求头实现 API Key、Bearer Token 等认证方式也可配合auth参数使用httpx2.Auth网络优化按业务响应特点定制超时与连接参数调试排障开发环境临时关闭 SSL 校验、注入标识头以追踪请求链路。核心收益灵活性HTTP 客户端行为完全对齐你的网络要求安全性支持自定义 SSL 证书与认证方式性能通过超时与连接配置优化调用质量兼容性适配企业代理与各类网络限制。参考资源示例主程序examples/mcp/streamablehttp_custom_client_example/main.py示例 MCP 服务器examples/mcp/streamablehttp_custom_client_example/server.py基础 Streamable HTTP 示例examples/mcp/streamablehttp_example/README.mdSDK 实现src/agents/mcp/server.py默认工厂 L306、v2 校验 L340、Streamable HTTP 参数 L2159、服务器类 L2200工厂协议定义src/agents/mcp/util.pyMCP 相关测试tests/mcp含test_streamable_http_client_factory.py等客户端与连接测试【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考