### Quick start with SQLiteSession and Runner Source: https://openai.github.io/openai-agents-python/zh/sessions/ Shows how to create an agent and a SQLiteSession, then run the agent multiple times with the same session to maintain context. The agent automatically remembers previous turns, and the synchronous runner also works. ```python from agents import Agent, Runner, SQLiteSession # Create agent agent = Agent( name="Assistant", instructions="Reply very concisely.", ) # Create a session instance with a session ID session = SQLiteSession("conversation_123") # First turn result = await Runner.run( agent, "What city is the Golden Gate Bridge in?", session=session ) print(result.final_output) # "San Francisco" # Second turn - agent automatically remembers previous context result = await Runner.run( agent, "What state is it in?", session=session ) print(result.final_output) # "California" # Also works with synchronous runner result = Runner.run_sync( agent, "What's the population?", session=session ) print(result.final_output) # "Approximately 39 million" ``` -------------------------------- ### Basic HostedMCPTool setup Source: https://openai.github.io/openai-agents-python/zh/mcp/ Creates an agent with a hosted MCP tool that uses the DeepWiki server. The tool_config dictionary corresponds to the JSON sent to the REST API. The hosted server automatically exposes its tools; no need to add them to mcp_servers. ```python import asyncio from agents import Agent, HostedMCPTool, Runner async def main() -> None: agent = Agent( name="Assistant", instructions="Use 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()) ``` -------------------------------- ### Configure RealtimeRunner with model settings for gpt-realtime-2.1 Source: https://openai.github.io/openai-agents-python/zh/realtime/guide/ Sets up a RealtimeRunner with nested audio configuration, including input format, transcription model, turn detection, and output voice. New realtime agents should start from gpt-realtime-2.1. ```python runner = RealtimeRunner( starting_agent=agent, config={ "model_settings": { "model_name": "gpt-realtime-2.1", "audio": { "input": { "format": "pcm16", "transcription": {"model": "gpt-4o-mini-transcribe"}, "turn_detection": {"type": "semantic_vad", "interrupt_response": True}, }, "output": {"format": "pcm16", "voice": "ash"}, }, "tool_choice": "auto", } }, ) ``` -------------------------------- ### Complete session memory example with SQLiteSession Source: https://openai.github.io/openai-agents-python/zh/sessions/ Use this example to see how SQLiteSession persists conversation history across multiple Runner.run calls. The agent remembers previous messages automatically, as shown by the third turn referencing the state from the second turn. Requires the agents package and asyncio. ```python import asyncio from agents import Agent, Runner, SQLiteSession async def main(): # Create an agent agent = Agent( name="Assistant", instructions="Reply very concisely.", ) # Create a session instance that will persist across runs session = SQLiteSession("conversation_123", "conversation_history.db") print("=== Sessions Example ===") print("The agent will remember previous messages automatically.\n") # First turn print("First turn:") print("User: What city is the Golden Gate Bridge in?") result = await Runner.run( agent, "What city is the Golden Gate Bridge in?", session=session ) print(f"Assistant: {result.final_output}") print() # Second turn - the agent will remember the previous conversation print("Second turn:") print("User: What state is it in?") result = await Runner.run( agent, "What state is it in?", session=session ) print(f"Assistant: {result.final_output}") print() # Third turn - continuing the conversation print("Third turn:") print("User: What's the population of that state?") result = await Runner.run( agent, "What's the population of that state?", session=session ) print(f"Assistant: {result.final_output}") print() print("=== Conversation Complete ===") print("Notice how the agent remembered the context from previous turns!") print("Sessions automatically handles conversation history.") if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Runner.run_streamed() Source: https://openai.github.io/openai-agents-python/zh/streaming/ Starts a streaming run of an agent. Returns a RunResultStreaming object that can be used to consume stream events. ```APIDOC ## Runner.run_streamed() ### Description Starts a streaming run of an agent. Returns a RunResultStreaming object that can be used to consume stream events. ### Method run_streamed ### Endpoint Runner.run_streamed(agent, input) ### Parameters - **agent** (Agent) - Required - The agent to run. - **input** (str) - Required - The input to the agent. ### Returns - **RunResultStreaming** - The streaming result object. ### Usage Example ```python result = Runner.run_streamed(agent, input="Please tell me 5 jokes.") ``` ``` -------------------------------- ### 启用 LiteLLM 序列化器兼容性补丁的环境变量 Source: https://openai.github.io/openai-agents-python/zh/models/ 设置环境变量以启用 SDK 的兼容性补丁,抑制 LiteLLM 响应序列化警告。该补丁默认禁用,仅在值为 '1' 或 'true' 时启用。 ```bash export OPENAI_AGENTS_ENABLE_LITELLM_SERIALIZER_PATCH=true ``` -------------------------------- ### Runner.run with SandboxRunConfig Source: https://openai.github.io/openai-agents-python/zh/sandbox/guide/ Example of running an agent with a shared sandbox session and a model-facing working directory using the Runner.run API. ```APIDOC ## Runner.run ### Description Runs an agent with a specified run configuration, including sandbox settings. This example shows how to use a shared sandbox session and set a model-facing working directory. ### Method run ### Endpoint Runner.run(agent, input, run_config) ### Parameters - **agent** (Agent) - Required - The agent to run. - **input** (str) - Required - The user input to the agent. - **run_config** (RunConfig) - Required - Configuration for the run, including sandbox settings. ### Request Example ```python from agents import Runner from agents.run import RunConfig from agents.sandbox import SandboxRunConfig result = await Runner.run( agent, "Work only on task A.", run_config=RunConfig( sandbox=SandboxRunConfig( session=shared_sandbox, cwd="tasks/task-a", ), ), ) ``` ### Response Returns a RunResult object containing the final output and other run details. ``` -------------------------------- ### SQLAlchemySession with existing AsyncEngine Source: https://openai.github.io/openai-agents-python/zh/sessions/ Example 2: Pass an existing SQLAlchemy AsyncEngine to SQLAlchemySession. Shows creating an engine and disposing it after use. ```python import asyncio from agents import Agent, Runner from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession from sqlalchemy.ext.asyncio import create_async_engine async def main(): # In your application, you would use your existing engine engine = create_async_engine("sqlite+aiosqlite:///conversations.db") agent = Agent("Assistant") session = SQLAlchemySession( "user-456", engine=engine, create_tables=True, # Auto-create tables for the demo ) result = await Runner.run(agent, "Hello", session=session) print(result.final_output) await engine.dispose() if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### 安装 Docker 支持的沙箱 Source: https://openai.github.io/openai-agents-python/zh/sandbox_agents/ 安装支持 Docker 沙箱的 SDK 扩展。 ```bash pip install "openai-agents[docker]" ``` -------------------------------- ### Hello world 示例:创建并运行智能体 Source: https://openai.github.io/openai-agents-python/zh/ 创建一个简单的智能体并运行,输出一首关于递归的俳句。运行前需设置 OPENAI_API_KEY 环境变量。 ```python from agents import Agent, Runner agent = Agent(name="Assistant", instructions="You are a helpful assistant") result = Runner.run_sync(agent, "Write a haiku about recursion in programming.") print(result.final_output) # Code within the code, # Functions calling themselves, # Infinite loop's dance. ``` -------------------------------- ### Access Usage in on_agent_end Hook Source: https://openai.github.io/openai-agents-python/zh/usage/ Example of a custom RunHooks class that prints usage information (requests and total tokens) when an agent ends. ```APIDOC ## RunHooks.on_agent_end ### Description Access usage data from the context object within a RunHooks lifecycle hook to log or record usage at key moments. ### Method Hook method (async) ### Signature `async def on_agent_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None` ### Parameters - **context** (RunContextWrapper) - Required - The run context wrapper containing the `usage` attribute. - **agent** (Agent) - Required - The agent that ended. - **output** (Any) - Required - The output of the agent. ### Usage Example ```python class MyHooks(RunHooks): async def on_agent_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None: u = context.usage print(f"{agent.name} → {u.requests} requests, {u.total_tokens} total tokens") ``` ### Notes - The `context.usage` object provides `requests` and `total_tokens` attributes. - This hook is called at the end of an agent's execution. ### Related API References - `Usage` - usage tracking data structure - `RequestUsage` - per-request usage details - `RunContextWrapper` - access usage from run context - `RunHooks` - lifecycle hooks for usage tracking ``` -------------------------------- ### SQLAlchemySession with from_url and in-memory SQLite Source: https://openai.github.io/openai-agents-python/zh/sessions/ Example 1: Simplest way to use SQLAlchemySession with an in-memory SQLite database via from_url. create_tables=True auto-creates tables for the demo. ```python import asyncio from agents import Agent, Runner from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession async def main(): agent = Agent("Assistant") session = SQLAlchemySession.from_url( "user-123", url="sqlite+aiosqlite:///:memory:", create_tables=True, # Auto-create tables for the demo ) result = await Runner.run(agent, "Hello", session=session) if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### 通过 RunConfig 设置默认模型 Source: https://openai.github.io/openai-agents-python/zh/models/ 为一次运行设置默认模型,未设置模型的智能体将使用该模型。 ```python from agents import Agent, RunConfig, Runner agent = Agent( name="Assistant", instructions="You're a helpful agent.", ) result = await Runner.run( agent, "Hello", run_config=RunConfig(model="gpt-5.6-sol"), ) ``` -------------------------------- ### 工具工作流测试 Source: https://openai.github.io/openai-agents-python/zh/testing/ 编写调用工具的模型响应脚本,再编写生成最终答案的响应脚本。真实的 SDK 工具管线会在这些模型调用之间运行,涵盖工具输入验证、执行、结果转换、钩子、安全防护措施和下一轮模型调用。直接调用 Python 函数会绕过这些 SDK 行为。 ```python import pytest from agents import Agent, RunConfig, Runner from agents.decorators import tool from agents.testing import ScriptedModel, assistant_message, function_call @tool def get_weather(city: str) -> str: """Return the weather for a city.""" return f"{city}: sunny" @pytest.mark.asyncio async def test_tool_workflow() -> None: model = ScriptedModel( [ [function_call("get_weather", {"city": "Tokyo"}, call_id="call_1")], [assistant_message("It is sunny in Tokyo.")], ] ) agent = Agent(name="Weather assistant", model=model, tools=[get_weather]) result = await Runner.run( agent, "What is the weather in Tokyo?", run_config=RunConfig(tracing_disabled=True), ) assert result.final_output == "It is sunny in Tokyo." assert len(model.calls) == 2 assert model.last_call is not None assert any( item.get("type") == "function_call_output" for item in model.last_call.input ) model.assert_complete() ``` -------------------------------- ### 配置运行器 Source: https://openai.github.io/openai-agents-python/zh/realtime/quickstart/ 配置 RealtimeRunner,使用嵌套的 audio.input/audio.output 设置,并指定模型 gpt-realtime-2.1。 ```python runner = RealtimeRunner( starting_agent=agent, config={ "model_settings": { "model_name": "gpt-realtime-2.1", "audio": { "input": { "format": "pcm16", "transcription": {"model": "gpt-4o-mini-transcribe"}, "turn_detection": { "type": "semantic_vad", "interrupt_response": True, }, }, "output": { "format": "pcm16", "voice": "ash", }, }, } }, ) ``` -------------------------------- ### Tool Guardrail Implementation Source: https://openai.github.io/openai-agents-python/zh/guardrails/ Defines tool input and output guardrails that block secrets (strings containing 'sk-') from being passed to or returned from a tool. The tool is decorated with these guardrails, and the example shows how to run the agent and print the final output. ```python import json from agents import ( Agent, Runner, ToolGuardrailFunctionOutput, ) from agents.decorators import tool, tool_input_guardrail, tool_output_guardrail @tool_input_guardrail def block_secrets(data): args = json.loads(data.context.tool_arguments or "{}") if "sk-" in json.dumps(args): return ToolGuardrailFunctionOutput.reject_content( "Remove secrets before calling this tool." ) return ToolGuardrailFunctionOutput.allow() @tool_output_guardrail def redact_output(data): text = str(data.output or "") if "sk-" in text: return ToolGuardrailFunctionOutput.reject_content("Output contained sensitive data.") return ToolGuardrailFunctionOutput.allow() @tool( tool_input_guardrails=[block_secrets], tool_output_guardrails=[redact_output], ) def classify_text(text: str) -> str: """Classify text for internal routing.""" return f"length:{len(text)}" agent = Agent(name="Classifier", tools=[classify_text]) result = Runner.run_sync(agent, "hello world") print(result.final_output) ``` -------------------------------- ### 使用 SandboxAgent 和 run_as 配置用户权限 Source: https://openai.github.io/openai-agents-python/zh/sandbox/guide/ 展示如何创建沙箱用户、设置清单条目权限,并通过 run_as 让智能体以该用户身份运行。 ```python from agents import Runner from agents.run import RunConfig from agents.sandbox import FileMode, Manifest, Permissions, SandboxAgent, SandboxRunConfig, User from agents.sandbox.entries import Dir, LocalDir from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient analyst = User(name="analyst") agent = SandboxAgent( name="Dataroom analyst", instructions="Review the files in `dataroom/` and write findings to `output/`.", default_manifest=Manifest( # Declare the sandbox user so manifest entries can grant access to it. users=[analyst], entries={ "dataroom": LocalDir( src="./dataroom", # Let the analyst traverse and read the mounted dataroom, but not edit it. group=analyst, permissions=Permissions( owner=FileMode.READ | FileMode.EXEC, group=FileMode.READ | FileMode.EXEC, other=FileMode.NONE, ), ), "output": Dir( # Give the analyst a writable scratch/output directory for artifacts. group=analyst, permissions=Permissions( owner=FileMode.ALL, group=FileMode.ALL, other=FileMode.NONE, ), ), }, ), # Run model-facing sandbox actions as this user, so those permissions apply. run_as=analyst, ) result = await Runner.run( agent, "Summarize the contracts and call out renewal dates.", run_config=RunConfig( sandbox=SandboxRunConfig(client=UnixLocalSandboxClient()), ), ) ``` -------------------------------- ### 配置实时任务转移 Source: https://openai.github.io/openai-agents-python/zh/realtime/guide/ 创建两个 RealtimeAgent,并通过 realtime_handoff() 将 billing_agent 作为 main_agent 的转移目标。可自定义转移描述。 ```python from agents.realtime import RealtimeAgent, realtime_handoff billing_agent = RealtimeAgent( name="Billing Support", instructions="You specialize in billing issues.", ) main_agent = RealtimeAgent( name="Customer Service", instructions="Triage the request and hand off when needed.", handoffs=[ realtime_handoff( billing_agent, tool_description_override="Transfer to billing support", ) ], ) ``` -------------------------------- ### Group multiple runs in a trace with trace() Source: https://openai.github.io/openai-agents-python/zh/tracing/ Wrap multiple Runner.run() calls in a with trace() block to make them part of the same overall trace instead of separate traces. The example shows two runs: one to generate a joke and another to rate it, printing both outputs. ```python from agents import Agent, Runner, trace async def main(): agent = Agent(name="Joke generator", instructions="Tell funny jokes.") with trace("Joke workflow"): # (1)! first_result = await Runner.run(agent, "Tell me a joke") second_result = await Runner.run(agent, f"Rate this joke: {first_result.final_output}") print(f"Joke: {first_result.final_output}") print(f"Rating: {second_result.final_output}") ``` -------------------------------- ### Configure LocalSnapshotSpec with UnixLocalSandboxClient Source: https://openai.github.io/openai-agents-python/zh/sandbox/guide/ Sets up a RunConfig with a UnixLocalSandboxClient and a LocalSnapshotSpec pointing to /tmp/my-sandbox-snapshots. Used when the runner creates a new sandbox session and needs to restore/persist workspace content. ```python from pathlib import Path from agents.run import RunConfig from agents.sandbox import LocalSnapshotSpec, SandboxRunConfig from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient run_config = RunConfig( sandbox=SandboxRunConfig( client=UnixLocalSandboxClient(), snapshot=LocalSnapshotSpec(base_path=Path("/tmp/my-sandbox-snapshots")), ) ) ``` -------------------------------- ### Using MCPServerSse for HTTP SSE MCP server Source: https://openai.github.io/openai-agents-python/zh/mcp/ Instantiates an MCPServerSse to connect to an MCP server over SSE. The SSE transport is deprecated; prefer streamable HTTP or stdio for new integrations. The example sets a workspace header and forces tool usage with tool_choice='required'. ```python from agents import Agent, Runner from agents.model_settings import ModelSettings from agents.mcp import MCPServerSse workspace_id = "demo-workspace" async with MCPServerSse( name="SSE Python Server", params={ "url": "http://localhost:8000/sse", "headers": {"X-Workspace": workspace_id}, }, cache_tools_list=True, ) as server: agent = Agent( name="Assistant", mcp_servers=[server], model_settings=ModelSettings(tool_choice="required"), ) result = await Runner.run(agent, "What's the weather in Tokyo?") print(result.final_output) ``` -------------------------------- ### Developer-owned lifecycle with explicit methods Source: https://openai.github.io/openai-agents-python/zh/sandbox/guide/ Creates a sandbox with a LocalSnapshotSpec and manually manages its lifecycle: start, run, stop to persist a checkpoint, and finally aclose for full cleanup. stop() only persists workspace content; aclose() also runs stop hooks and closes resources. ```python sandbox = await client.create( manifest=agent.default_manifest, snapshot=LocalSnapshotSpec(base_path=Path("/tmp/my-sandbox-snapshots")), ) try: await sandbox.start() await Runner.run( agent, "Analyze the files.", run_config=RunConfig(sandbox=SandboxRunConfig(session=sandbox)), ) # Persist a checkpoint of the live workspace before doing more work. # `aclose()` also calls `stop()`, so this is only needed for an explicit mid-lifecycle save. await sandbox.stop() finally: await sandbox.aclose() ``` -------------------------------- ### 从 Unix-local 切换到 Docker 的 RunConfig 配置 Source: https://openai.github.io/openai-agents-python/zh/sandbox/clients/ 保持智能体定义不变,仅更改运行配置以使用 DockerSandboxClient,并指定镜像。 ```python from docker import from_env as docker_from_env from agents.run import RunConfig from agents.sandbox import SandboxRunConfig from agents.sandbox.sandboxes.docker import DockerSandboxClient, DockerSandboxClientOptions run_config = RunConfig( sandbox=SandboxRunConfig( client=DockerSandboxClient(docker_from_env()), options=DockerSandboxClientOptions(image="python:3.14-slim"), ), ) ``` -------------------------------- ### RunConfig.tool_execution Source: https://openai.github.io/openai-agents-python/zh/running_agents/ 配置本地工具调用的 SDK 端执行行为,例如限制同时运行的本地函数工具调用数量。 ```APIDOC ## RunConfig.tool_execution ### Description 配置本地工具调用的 SDK 端执行行为,例如限制同时运行的本地函数工具调用数量。 ### Method 属性配置 ### Parameters - **tool_execution** (ToolExecutionConfig) - 可选 - 配置本地工具调用的执行行为。 ### Response 无直接返回值,作为 RunConfig 的配置属性使用。 ``` -------------------------------- ### Using MCPServerStdio for local subprocess MCP server Source: https://openai.github.io/openai-agents-python/zh/mcp/ Instantiates an MCPServerStdio to run an MCP server as a local subprocess. The SDK starts the process, keeps the pipe open, and closes it when the context manager exits. Useful for quick proofs of concept or when the server only has a CLI entry point. ```python 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( name="Filesystem Server via npx", params={ "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)], }, ) as server: agent = Agent( name="Assistant", instructions="Use 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) ``` -------------------------------- ### 设置 API 密钥环境变量 Source: https://openai.github.io/openai-agents-python/zh/realtime/quickstart/ 在环境中设置 OPENAI_API_KEY。 ```bash export OPENAI_API_KEY="your-api-key-here" ``` -------------------------------- ### Input Guardrail Implementation Source: https://openai.github.io/openai-agents-python/zh/guardrails/ Defines an input guardrail that checks if the user is asking for math homework help. The guardrail function runs a separate agent to classify the input and triggers a tripwire if it detects math homework. The main agent is configured with this guardrail, and the example shows how to catch the InputGuardrailTripwireTriggered exception. ```python from pydantic import BaseModel from agents import ( Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, RunContextWrapper, Runner, TResponseInputItem, ) from agents.decorators import input_guardrail class MathHomeworkOutput(BaseModel): is_math_homework: bool reasoning: str guardrail_agent = Agent( # (1)! name="Guardrail check", instructions="Check if the user is asking you to do their math homework.", output_type=MathHomeworkOutput, ) @input_guardrail async def math_guardrail( # (2)! ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem] ) -> GuardrailFunctionOutput: result = await Runner.run(guardrail_agent, input, context=ctx.context) return GuardrailFunctionOutput( output_info=result.final_output, # (3)! tripwire_triggered=result.final_output.is_math_homework, ) agent = Agent( # (4)! name="Customer support agent", instructions="You are a customer support agent. You help customers with their questions.", input_guardrails=[math_guardrail], ) async def main(): # This should trip the guardrail try: await Runner.run(agent, "Hello, can you help me solve for x: 2x + 3 = 11?") print("Guardrail didn't trip - this is unexpected") except InputGuardrailTripwireTriggered: print("Math homework guardrail tripped") ```