openai-agents-python MCP 集成全指南五種傳輸方式、審批策略與高級配置實戰【免費下載鏈接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows項目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonModel Context ProtocolMCP為應用程序向語言模型暴露工具與上下文提供了一套標準化協議。本文以 openai-agents-python 官方文檔docs/zh/mcp.md為骨架系統講解如何把 MCP 服務器接入智能體從托管式 MCP 工具、Streamable HTTP、SSE、stdio 到服務器管理器覆蓋每一種傳輸方式的配置方法、審批策略、工具篩選、分頁、緩存與追蹤等通用能力并深入到 server.py 與 tool.py 的源碼實現幫助你在真實項目中做出正確的集成選型并快速落地。MCP 是什么為什么要在智能體中使用它MCP 是一種開放協議用于標準化應用程序向 LLM 提供上下文的方式。官方將其類比為“AI 應用程序的 USB-C 端口”正如 USB-C 提供了一種將設備連接到各種外圍設備和配件的標準化方式MCP 也提供了一種將 AI 模型連接到不同數據源和工具的標準化方式。Agents Python SDK 支持多種 MCP 傳輸方式因此你可以復用現有的 MCP 服務器也可以自行構建服務器向智能體公開由文件系統、HTTP 或連接器支持的工具。??連接前務必進行信任驗證MCP 工具可以公開模型上下文中的數據并使用你提供的憑據執行操作。請僅連接你信任的服務器使用最小權限憑據將訪問令牌放在授權字段或標頭中而不是 URL 中并要求對敏感操作進行審批。如何選擇 MCP 集成方式將 MCP 服務器接入智能體之前需要先確定應在何處執行工具調用以及你可以訪問哪些傳輸方式。下表匯總了 Python SDK 支持的選項需求推薦選項讓 OpenAI 的 Responses API 代表模型調用可公開訪問的 MCP 服務器通過HostedMCPTool使用托管式 MCP 服務器工具連接到你在本地或遠程運行的 Streamable HTTP 服務器通過MCPServerStreamableHttp使用Streamable HTTP MCP 服務器與實現了基于 Server-Sent Events 的 HTTP 的服務器通信通過MCPServerSse使用基于 SSE 的 HTTP MCP 服務器啟動本地進程并通過 stdin/stdout 通信通過MCPServerStdio使用stdio MCP 服務器其中HostedMCPTool定義在 src/agents/tool.py三種本地 MCP 服務器類則全部實現于 src/agents/mcp/server.pyMCPServerStdio在 L1888 附近、MCPServerSse在 L2030 附近、MCPServerStreamableHttp在 L2200 附近并統一通過 src/agents/mcp/init.py 對外導出。MCP Python SDK v1 與 v2 的兼容性Agents SDK 通過依賴版本范圍mcp1.19.0,3同時支持mcpPython 軟件包的兩個主要版本。需要特別注意的是已安裝的mcp軟件包版本與同服務器協商的 MCP 協議版本相互獨立。SDK 會檢測已安裝軟件包的主版本并自動適配 stdio、SSE 和 Streamable HTTP 連接因此普通服務器配置不需要提供版本切換選項。安裝 MCP Python SDK v2 后Agents SDK 會圍繞配置的本地傳輸方式創建帶有modeauto的 v2mcp.Client。客戶端首先使用已安裝 MCP SDK 所支持的最新協議版本發送server/discover探測請求現代服務器會響應此探測請求客戶端隨后采用響應結果較舊的服務器不支持server/discover時客戶端會回退到舊版initialize握手并使用在該過程中協商的協議版本。因此安裝 MCP Python SDK v2并不會強制所有連接都使用最新的 MCP 協議版本。大多數應用程序應讓依賴解析器選擇兼容版本如果你的應用程序必須固定使用某個主版本請在openai-agents旁添加顯式約束# MCP Python SDK v1 pip install mcp1.19.0,2 # MCP Python SDK v2 pip install mcp2,3HTTP 傳輸自定義與包版本強相關HTTP 傳輸自定義必須使用已安裝 MCP 軟件包所擁有的 HTTP 棧二者不能混用自定義項MCP Python SDK v1MCP Python SDK v2params[auth]httpx.Authhttpx2.Authparams[httpx_client_factory]返回值httpx.AsyncClienthttpx2.AsyncClientMCPServerStreamableHttp的params[ignore_initialized_notification_failure] True支持不支持連接前會被拒絕應盡可能使用Authorization標頭——它在兩個軟件包版本中均可保持不變。當應用程序提供params[auth]或params[httpx_client_factory]時這些值必須使用已安裝mcp包主版本對應的 HTTP 類型。當應用程序設置MCPServerStreamableHttp的params[ignore_initialized_notification_failure] True時必須保留mcp2或在升級前禁用該選項。從源碼可以看到這一約束是硬校驗而非軟提示server.py 中create_streams()檢測到 MCP v2 且該參數為True時會直接拋出UserErrornot supported with MCP Python SDK v2。同時 MCPServerStreamableHttpParams 對auth、httpx_client_factory的注釋也分別標注了 v1/v2 對應的 HTTP 類型。這些本地mcp依賴要求不適用于HostedMCPTool因為遠程 MCP 連接由 OpenAI Responses API 管理本地無需安裝或協調mcp包。智能體級 MCP 配置Agent.mcp_config除了選擇傳輸方式外還可以通過設置Agent.mcp_config調整 MCP 工具的準備方式from agents import Agent agent Agent( nameAssistant, mcp_servers[server], mcp_config{ # Try to convert MCP tool schemas to strict JSON schema. convert_schemas_to_strict: True, # If None, MCP tool failures are raised as exceptions instead of # returning model-visible error text. failure_error_function: None, # Prefix local MCP tool names with their server name. include_server_in_tool_names: True, }, )注意事項convert_schemas_to_strict采用盡力而為的方式。如果某個架構無法轉換則使用原始架構。failure_error_function控制如何向模型呈現 MCP 工具調用失敗。未設置時SDK 使用默認的工具錯誤格式化程序。服務器級的failure_error_function會覆蓋該服務器的Agent.mcp_config[failure_error_function]。這一優先級在MCPServerStreamableHttp的構造函數注釋中有明確說明server.py顯式設為None時拋出錯誤而非轉換完全不設置時才回退到智能體級配置或 SDK 默認。include_server_in_tool_names需要主動啟用。啟用后每個本地 MCP 工具都會使用確定性的服務器前綴名稱向模型公開有助于避免多個 MCP 服務器發布同名工具時發生沖突。生成的名稱兼容 ASCII、不會超過FunctionTool實例的名稱長度限制也不會與同一智能體上本地FunctionTool實例的已配置名稱或已啟用的任務轉移發生沖突。SDK 仍會在原始服務器上調用具有原始名稱的 MCP 工具。各傳輸方式的通用決策模式選擇傳輸方式后大多數集成還需要作出相同的后續決策如何僅公開一部分工具工具篩選。服務器是否還提供可復用的提示詞提示詞。是否應緩存list_tools()緩存。MCP 活動如何顯示在追蹤中追蹤。對于本地 MCP 服務器MCPServerStdio、MCPServerSse、MCPServerStreamableHttp審批策略和每次調用的_meta載荷也是通用概念。下文 Streamable HTTP 一節給出了最完整的代碼示例同樣的模式也適用于其他本地傳輸方式。1. 托管式 MCP 服務器工具Hosted MCP Server Tools托管工具會將整個工具調用往返流程交由 OpenAI 基礎設施處理。你的代碼無需列出和調用工具HostedMCPTool會將服務器標簽以及可選的連接器元數據轉發給 Responses API。模型會列出遠程服務器的工具并調用它們而無需額外回調你的 Python 進程。目前托管工具適用于支持 Responses API 托管式 MCP 集成的 OpenAI 模型。從源碼看HostedMCPTool是一個簡單的 dataclasstool.py核心字段只有tool_config類型為Mcp與發送給 REST API 的 JSON 相對應和可選的on_approval_request回調其name屬性固定返回hosted_mcp——也就是說工具調用本身完全由模型端驅動。基礎托管式 MCP 工具將HostedMCPTool添加到智能體的tools列表即可創建托管工具。tool_config字典與發送給 REST API 的 JSON 相對應import asyncio from agents import Agent, HostedMCPTool, Runner async def main() - None: agent Agent( nameAssistant, instructionsUse the DeepWiki hosted MCP server to inspect openai/openai-agents-python., tools[ HostedMCPTool( tool_config{ type: mcp, server_label: deepwiki, server_url: https://mcp.deepwiki.com/mcp, require_approval: never, } ) ], ) result await Runner.run( agent, Which language is the repository openai/openai-agents-python written in?, ) print(result.final_output) asyncio.run(main())托管服務器會自動公開其工具無需將其添加到mcp_servers。如果希望托管工具搜索以延遲加載方式加載托管式 MCP 服務器請設置tool_config[defer_loading] True并將ToolSearchTool添加到智能體。僅 OpenAI Responses 模型支持此功能完整的工具搜索設置和限制可參閱 docs/zh/tools.md。托管式 MCP 結果的流式傳輸托管工具支持流式傳輸結果其方式與函數工具完全相同。使用Runner.run_streamed可在模型仍在工作時接收增量 MCP 輸出result Runner.run_streamed(agent, Summarise this repositorys top languages) async for event in result.stream_events(): if event.type run_item_stream_event: print(fReceived: {event.item}) print(result.final_output)可選審批流程如果服務器能夠執行敏感操作可以要求在每次執行工具前進行人工或程序化審批。在tool_config中配置require_approval其值可以是單一策略always、never也可以是將工具名稱映射到策略的字典。若要在 Python 中作出決定請提供on_approval_request回調from agents import MCPToolApprovalFunctionResult, MCPToolApprovalRequest SAFE_TOOLS {read_wiki_structure, read_wiki_contents, ask_question} def approve_tool(request: MCPToolApprovalRequest) - MCPToolApprovalFunctionResult: if request.data.name in SAFE_TOOLS: return {approve: True} return {approve: False, reason: Escalate to a human reviewer} agent Agent( nameAssistant, tools[ HostedMCPTool( tool_config{ type: mcp, server_label: deepwiki, server_url: https://mcp.deepwiki.com/mcp, require_approval: always, }, on_approval_requestapprove_tool, ) ], )回調可以是同步或異步的并且每當模型需要審批數據才能繼續運行時都會調用它。對應類型定義在 tool.pyMCPToolApprovalRequest攜帶ctx_wrapper運行上下文與data審批請求數據MCPToolApprovalFunctionResult則是一個 TypedDict必填approve: bool可選用reason說明拒絕原因。由連接器支持的托管服務器托管式 MCP 還支持 OpenAI 連接器。無需指定server_url只需提供connector_id和訪問令牌。Responses API 會處理身份驗證托管服務器則會公開連接器的工具import os HostedMCPTool( tool_config{ type: mcp, server_label: google_calendar, connector_id: connector_googlecalendar, authorization: os.environ[GOOGLE_CALENDAR_AUTHORIZATION], require_approval: never, } )完整可運行的托管工具代碼示例包括流式傳輸、審批和連接器位于倉庫的 examples/hosted_mcp 目錄。2. Streamable HTTP MCP 服務器如果希望自行管理網絡連接請使用MCPServerStreamableHttp。當你需要控制傳輸方式或者希望在自己的基礎設施中運行服務器并保持較低延遲時Streamable HTTP 服務器是理想選擇import asyncio import os from agents import Agent, Runner from agents.mcp import MCPServerStreamableHttp from agents.model_settings import ModelSettings async def main() - None: token os.environ[MCP_SERVER_TOKEN] async with MCPServerStreamableHttp( nameStreamable HTTP Python Server, params{ url: http://localhost:8000/mcp, headers: {Authorization: fBearer {token}}, timeout: 10, }, cache_tools_listTrue, max_retry_attempts3, ) as server: agent Agent( nameAssistant, instructionsUse the MCP tools to answer the questions., mcp_servers[server], model_settingsModelSettings(tool_choicerequired), ) result await Runner.run(agent, Add 7 and 22.) print(result.final_output) asyncio.run(main())構造函數完整參數說明對照 MCPServerStreamableHttp.init的簽名構造函數除params外還接受以下選項client_session_timeout_seconds控制 MCP ClientSession 的讀取超時默認 5 秒。可由datetime.timedelta表示且至少為一微秒的有限正值會設置有限超時None和0會禁用超時。構造服務器時會拒絕其他值。use_structured_content控制是否優先使用tool_result.structured_content而不是文本輸出。默認False向后兼容——大多數 MCP 服務器仍會把結構化內容包含在tool_result.content中默認啟用會導致重復內容只有當你確認服務器不會在tool_result.content中重復結構化內容時才設為True。max_retry_attempts與retry_backoff_seconds_base為list_tools()和call_tool()添加自動重試指數退避默認不重試retry_backoff_seconds_base默認 1.0 秒另有可選的retry_backoff_seconds_max封頂退避延遲。tool_filter允許你僅公開一部分工具見下文工具篩選。require_approval為本地 MCP 工具啟用人機協同審批策略。failure_error_function自定義模型可見的 MCP 工具失敗消息設置為None可改為拋出錯誤。tool_meta_resolver會在call_tool()之前注入每次調用的 MCP_meta載荷。message_handler可選處理 ClientSession 交付的會話消息。tool_input_guardrails/tool_output_guardrails可選對服務器上的每個工具在調用前/返回后應用守衛guardrail。params字典本身MCPServerStreamableHttpParams支持url必填、headers、timeoutHTTP 請求超時默認 5 秒、sse_read_timeoutSSE 連接超時默認 5 分鐘、terminate_on_close默認True、httpx_client_factory、auth、ignore_initialized_notification_failure。本地 MCP 服務器的審批策略MCPServerStdio、MCPServerSse和MCPServerStreamableHttp均接受require_approval支持以下形式類型定義見 server.py 中的RequireApprovalPolicy、RequireApprovalMapping、RequireApprovalObject與RequireApprovalSetting對所有工具使用always或never。True要求審批所有工具False不要求審批任何工具分別等同于always和never。按工具配置的映射例如{delete_file: always, read_file: never}。分組對象{always: {tool_names: [...]}, never: {tool_names: [...]}}。此外還支持傳入一個可調用對象LocalMCPApprovalCallable接收RunContextWrapper、請求工具的AgentBase和MCPTool返回是否放行的布爾值。async with MCPServerStreamableHttp( nameFilesystem MCP, params{url: http://localhost:8000/mcp}, require_approval{always: {tool_names: [delete_file]}}, ) as server: ...有關完整的暫停/恢復流程請參閱 docs/zh/human_in_the_loop.md 和 examples/mcp/get_all_mcp_tools_example/main.py。使用 tool_meta_resolver 注入每次調用元數據當 MCP 服務器要求在_meta中提供請求元數據例如租戶 ID 或追蹤上下文時請使用tool_meta_resolver。以下代碼示例假設你將dict作為context傳遞給Runner.run(...)from agents.mcp import MCPServerStreamableHttp, MCPToolMetaContext def resolve_meta(context: MCPToolMetaContext) - dict[str, str] | None: run_context_data context.run_context.context or {} tenant_id run_context_data.get(tenant_id) if tenant_id is None: return None return {tenant_id: str(tenant_id), source: agents-sdk} server MCPServerStreamableHttp( nameMetadata-aware MCP, params{url: http://localhost:8000/mcp}, tool_meta_resolverresolve_meta, )如果運行上下文是 Pydantic 模型、dataclass 或自定義類請改用屬性訪問方式讀取租戶 ID而不是字典式get。MCP 工具輸出文本、圖像及其他內容當 MCP 結果使用內容塊時SDK 的處理規則如下文本內容作為文本輸出轉發圖像內容映射為工具輸出中的圖像類型條目其他 MCP 內容塊類型包括音頻和資源塊SDK 轉發文本輸出其值為該內容塊的有效 JSON 序列化結果包含多個內容塊的響應會作為輸出項列表轉發如果use_structured_contentTrue選擇了非空且無錯誤的structuredContent載荷則該結構化載荷優先于這些內容塊結構化內容缺失或為空時回退到內容塊。3. 基于 SSE 的 HTTP MCP 服務器??注意MCP 項目已棄用Server-Sent Events 傳輸方式。對于新集成請優先使用 Streamable HTTP 或 stdio僅為舊版服務器保留 SSE。如果 MCP 服務器實現了基于 SSE 的 HTTP 傳輸方式請實例化MCPServerSse。除傳輸方式外其 API 與 Streamable HTTP 服務器完全相同from agents import Agent, Runner from agents.model_settings import ModelSettings from agents.mcp import MCPServerSse workspace_id demo-workspace async with MCPServerSse( nameSSE Python Server, params{ url: http://localhost:8000/sse, headers: {X-Workspace: workspace_id}, }, cache_tools_listTrue, ) as server: agent Agent( nameAssistant, mcp_servers[server], model_settingsModelSettings(tool_choicerequired), ) result await Runner.run(agent, Whats the weather in Tokyo?) print(result.final_output)4. stdio MCP 服務器對于以本地子進程方式運行的 MCP 服務器請使用MCPServerStdio。SDK 會啟動該進程、保持管道打開并在退出上下文管理器時自動關閉管道。此選項適合快速構建概念驗證或服務器僅公開命令行入口點的情況from pathlib import Path from agents import Agent, Runner from agents.mcp import MCPServerStdio current_dir Path(__file__).parent samples_dir current_dir / sample_files async with MCPServerStdio( nameFilesystem Server via npx, params{ command: npx, args: [-y, modelcontextprotocol/server-filesystem, str(samples_dir)], }, ) as server: agent Agent( nameAssistant, instructionsUse the files in the sample directory to answer questions., mcp_servers[server], ) result await Runner.run(agent, List the files available to you.) print(result.final_output)一個真實可運行的完整版本位于 examples/mcp/filesystem_example/main.py它在__main__中先通過shutil.which(npx)檢查npx是否安裝再用trace()包裹一次多輪對話列出文件、讀取書目、推薦歌曲并在上下文中把 stdio 服務器掛載到同一個智能體上反復復用。5. MCP 服務器管理器MCPServerManager如果有多個 MCP 服務器請使用MCPServerManager預先連接它們并向智能體公開其中成功連接的服務器子集from agents import Agent, Runner from agents.mcp import MCPServerManager, MCPServerStreamableHttp servers [ MCPServerStreamableHttp(namecalendar, params{url: http://localhost:8000/mcp}), MCPServerStreamableHttp(namedocs, params{url: http://localhost:8001/mcp}), ] async with MCPServerManager(servers) as manager: agent Agent( nameAssistant, instructionsUse MCP tools when they help., mcp_serversmanager.active_servers, ) result await Runner.run(agent, Which MCP tools are available?) print(result.final_output)主要行為當drop_failed_serversTrue默認值時active_servers僅包含成功連接的服務器。失敗信息記錄在failed_servers和errors中。設置strictTrue可在首次連接失敗時拋出異常。調用reconnect(failed_onlyTrue)可重試失敗的服務器調用reconnect(failed_onlyFalse)可重啟所有服務器。對connect_all()、reconnect()和cleanup_all()的調用會串行執行。如果某個生命周期操作已在運行另一個生命周期操作會等待其完成而不會并發連接或清理相同的服務器。設置connect_timeout_seconds、cleanup_timeout_seconds和connect_in_parallel可調整生命周期行為。兩個生命周期超時的默認值均為10 秒。它們接受有限正秒數或使用None禁用在構造和賦值時都會進行驗證零會被拒絕因為它會產生立即到期的截止時間。構造函數選項與重連行為的完整 API 參考見 docs/ref/mcp/manager.md可運行示例見 examples/mcp/manager_example含app.py、mcp_server.py與smoke_test.py。通用服務器能力以下各節適用于所有 MCP 服務器傳輸方式具體 API 范圍取決于服務器類。工具篩選每個 MCP 服務器都支持工具篩選器因此你可以僅公開智能體所需的函數。篩選可以在構造時進行也可以在每次運行時動態進行。靜態工具篩選使用create_static_tool_filter配置簡單的允許列表和阻止列表from pathlib import Path from agents.mcp import MCPServerStdio, create_static_tool_filter samples_dir Path(/path/to/files) filesystem_server MCPServerStdio( params{ command: npx, args: [-y, modelcontextprotocol/server-filesystem, str(samples_dir)], }, tool_filtercreate_static_tool_filter(allowed_tool_names[read_file, write_file]), )同時提供allowed_tool_names和blocked_tool_names時SDK 會先應用允許列表然后從剩余集合中移除所有被阻止的工具。動態工具篩選對于更復雜的邏輯請傳入一個可調用對象該對象接收ToolFilterContext。該可調用對象可以是同步或異步的并在應公開工具時返回Truefrom pathlib import Path from agents.mcp import MCPServerStdio, ToolFilterContext samples_dir Path(/path/to/files) async def context_aware_filter(context: ToolFilterContext, tool) - bool: if context.agent.name Code Reviewer and tool.name.startswith(danger_): return False return True async with MCPServerStdio( params{ command: npx, args: [-y, modelcontextprotocol/server-filesystem, str(samples_dir)], }, tool_filtercontext_aware_filter, ) as server: ...篩選器上下文會公開活動的run_context、請求工具的agent以及server_name。可運行示例見 examples/mcp/tool_filter_example/main.py。提示詞MCP 服務器還可以提供動態生成智能體指令的提示詞。支持提示詞的服務器會公開兩種方法list_prompts()枚舉可用的提示詞模板。get_prompt(name, arguments)獲取具體提示詞并可選擇提供參數。from agents import Agent prompt_result await server.get_prompt( generate_code_review_instructions, {focus: security vulnerabilities, language: python}, ) instructions prompt_result.messages[0].content.text agent Agent( nameCode Reviewer, instructionsinstructions, mcp_servers[server], )服務端與客戶端兩端完整的提示詞示例見 examples/mcp/prompt_serverserver.py提供提示詞main.py消費提示詞構造智能體。分頁內置的本地 MCP 服務器類在列出工具和提示詞時會自動跟隨nextCursorlist_tools()會先收集完整的工具列表再應用篩選器或填充緩存list_prompts()返回一個合并結果其中包含nextCursorNone如果后續頁面失敗或服務器重復使用游標該操作會拋出錯誤而不會公開或緩存部分結果。資源仍需顯式分頁將list_resources()或list_resource_templates()返回的nextCursor作為cursor參數傳回以獲取下一頁。緩存每次智能體運行都會在每個 MCP 服務器上調用list_tools()。遠程服務器可能帶來明顯的延遲因此所有 MCP 服務器類都公開了cache_tools_list選項。僅當你確信工具定義不會頻繁變化時才應將其設置為True。如需稍后強制獲取最新列表請在服務器實例上調用invalidate_tools_cache()。從源碼看server.py啟用緩存后工具列表只會從服務器獲取一次能大幅降低延遲避免每次運行都做一次往返反之每次list_tools()都會實時拉取。追蹤追蹤會自動捕獲 MCP 活動包括為列出工具而對 MCP 服務器發起的調用。工具調用中的 MCP 相關信息。上圖展示了 MCP 追蹤的實際效果你可以看到對 MCP 服務器的list_tools調用以及工具調用上攜帶的 MCP 相關元數據便于在分布式運行中定位工具列表延遲或調用失敗的具體環節。延伸閱讀examples/mcp —— 可運行的 stdio、SSE 和 Streamable HTTP 代碼示例含 filesystem、git、SSE、Streamable HTTP 遠程/自定義客戶端、提示詞、工具篩選與管理器等完整場景。examples/hosted_mcp —— 完整的托管式 MCP 演示包括審批on_approval.py和連接器connectors.py。docs/zh/human_in_the_loop.md —— 人機協同審批的完整暫停/恢復流程。docs/ref/mcp —— MCP 相關 API 參考server、manager、util 等。結語按場景選擇集成路線綜合以上內容實際項目中的選型建議可歸納為服務器可公開訪問且希望零本地依賴時選托管式HostedMCPTool自建基礎設施、追求低延遲與傳輸可控時選 Streamable HTTP舊版服務器只支持 SSE 時用MCPServerSse新項目避免本地子進程或命令行入口點用MCPServerStdio多服務器接入時用MCPServerManager統一生命周期。無論走哪條路線工具篩選、審批策略、緩存與追蹤這四個通用能力都值得在架構設計階段一并規劃這樣既能保證安全邊界也能讓 MCP 集成在長期運行中保持可觀測、可維護。【免費下載鏈接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows項目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考