OpenAI Agents SDK

https://openai.github.io/openai-agents-python/zh/
Admin
OpenAI Agents SDK is a production-ready Python package for building agent-based AI applications with...
Tokens:228,530
Snippets:1,108
Trust Score:9.7
Update:19 hours ago
Show doc for...
Tokens:
Raw
### 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")
```
Tokens:
Raw