fromagentsimportAgentagent=Agent(name="Assistant",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,},)

importasynciofromagentsimportAgent,HostedMCPTool,Runnerasyncdefmain()->None:agent=Agent(name="Assistant",tools=[HostedMCPTool(tool_config={"type":"mcp","server_label":"gitmcp","server_url":"https://gitmcp.io/openai/codex","require_approval":"never",})],)result=awaitRunner.run(agent,"Which language is this repository written in?")print(result.final_output)asyncio.run(main())

result=Runner.run_streamed(agent,"Summarise this repository's top languages")asyncforeventinresult.stream_events():ifevent.type=="run_item_stream_event":print(f"Received:{event.item}")print(result.final_output)

fromagentsimportMCPToolApprovalFunctionResult,MCPToolApprovalRequestSAFE_TOOLS={"read_project_metadata"}defapprove_tool(request:MCPToolApprovalRequest)->MCPToolApprovalFunctionResult:ifrequest.data.nameinSAFE_TOOLS:return{"approve":True}return{"approve":False,"reason":"Escalate to a human reviewer"}agent=Agent(name="Assistant",tools=[HostedMCPTool(tool_config={"type":"mcp","server_label":"gitmcp","server_url":"https://gitmcp.io/openai/codex","require_approval":"always",},on_approval_request=approve_tool,)],)

importosHostedMCPTool(tool_config={"type":"mcp","server_label":"google_calendar","connector_id":"connector_googlecalendar","authorization":os.environ["GOOGLE_CALENDAR_AUTHORIZATION"],"require_approval":"never",})

importasyncioimportosfromagentsimportAgent,Runnerfromagents.mcpimportMCPServerStreamableHttpfromagents.model_settingsimportModelSettingsasyncdefmain()->None:token=os.environ["MCP_SERVER_TOKEN"]asyncwithMCPServerStreamableHttp(name="Streamable HTTP Python Server",params={"url":"http://localhost:8000/mcp","headers":{"Authorization":f"Bearer{token}"},"timeout":10,},cache_tools_list=True,max_retry_attempts=3,)asserver:agent=Agent(name="Assistant",instructions="Use the MCP tools to answer the questions.",mcp_servers=[server],model_settings=ModelSettings(tool_choice="required"),)result=awaitRunner.run(agent,"Add 7 and 22.")print(result.final_output)asyncio.run(main())

asyncwithMCPServerStreamableHttp(name="Filesystem MCP",params={"url":"http://localhost:8000/mcp"},require_approval={"always":{"tool_names":["delete_file"]}},)asserver:...

fromagents.mcpimportMCPServerStreamableHttp,MCPToolMetaContextdefresolve_meta(context:MCPToolMetaContext)->dict[str,str]|None:run_context_data=context.run_context.contextor{}tenant_id=run_context_data.get("tenant_id")iftenant_idisNone:returnNonereturn{"tenant_id":str(tenant_id),"source":"agents-sdk"}server=MCPServerStreamableHttp(name="Metadata-aware MCP",params={"url":"http://localhost:8000/mcp"},tool_meta_resolver=resolve_meta,)

Warning

MCP 项目已弃用 Server-Sent Events 传输方式。新集成请优先使用 Streamable HTTP 或 stdio，仅在遗留服务场景保留 SSE。

fromagentsimportAgent,Runnerfromagents.model_settingsimportModelSettingsfromagents.mcpimportMCPServerSseworkspace_id="demo-workspace"asyncwithMCPServerSse(name="SSE Python Server",params={"url":"http://localhost:8000/sse","headers":{"X-Workspace":workspace_id},},cache_tools_list=True,)asserver:agent=Agent(name="Assistant",mcp_servers=[server],model_settings=ModelSettings(tool_choice="required"),)result=awaitRunner.run(agent,"What's the weather in Tokyo?")print(result.final_output)

frompathlibimportPathfromagentsimportAgent,Runnerfromagents.mcpimportMCPServerStdiocurrent_dir=Path(__file__).parentsamples_dir=current_dir/"sample_files"asyncwithMCPServerStdio(name="Filesystem Server via npx",params={"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem",str(samples_dir)],},)asserver:agent=Agent(name="Assistant",instructions="Use the files in the sample directory to answer questions.",mcp_servers=[server],)result=awaitRunner.run(agent,"List the files available to you.")print(result.final_output)

fromagentsimportAgent,Runnerfromagents.mcpimportMCPServerManager,MCPServerStreamableHttpservers=[MCPServerStreamableHttp(name="calendar",params={"url":"http://localhost:8000/mcp"}),MCPServerStreamableHttp(name="docs",params={"url":"http://localhost:8001/mcp"}),]asyncwithMCPServerManager(servers)asmanager:agent=Agent(name="Assistant",instructions="Use MCP tools when they help.",mcp_servers=manager.active_servers,)result=awaitRunner.run(agent,"Which MCP tools are available?")print(result.final_output)

frompathlibimportPathfromagents.mcpimportMCPServerStdio,create_static_tool_filtersamples_dir=Path("/path/to/files")filesystem_server=MCPServerStdio(params={"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem",str(samples_dir)],},tool_filter=create_static_tool_filter(allowed_tool_names=["read_file","write_file"]),)

frompathlibimportPathfromagents.mcpimportMCPServerStdio,ToolFilterContextsamples_dir=Path("/path/to/files")asyncdefcontext_aware_filter(context:ToolFilterContext,tool)->bool:ifcontext.agent.name=="Code Reviewer"andtool.name.startswith("danger_"):returnFalsereturnTrueasyncwithMCPServerStdio(params={"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem",str(samples_dir)],},tool_filter=context_aware_filter,)asserver:...

fromagentsimportAgentprompt_result=awaitserver.get_prompt("generate_code_review_instructions",{"focus":"security vulnerabilities","language":"python"},)instructions=prompt_result.messages[0].content.textagent=Agent(name="Code Reviewer",instructions=instructions,mcp_servers=[server],)