Book

OpenAI Agents SDK

Compiled from 43 nodes · est. 72 min read

2.1Orientation

Updated 2026-08-28

The OpenAI Agents SDK is a provider-agnostic Python framework supporting OpenAI's Responses and Chat Completions APIs plus 100+ other LLMs; its public entry points (Agent, Runner, Handoff, tools, tracing, sessions) are defined in src/agents/__init__.py. Configuration functions in src/agents/__init__.py delegate to an internal _config module to centralize LLM client selection, API choice (Responses or Chat Completions), tracing credentials, and harness ID defaults. The Responses API is OpenAI's stateful API that stores conversation context server-side; the Chat Completions API is the stateless interface where the client manages full conversation context.

The OpenAI Agents SDK is provider-agnostic, supporting the OpenAI Responses and Chat Completions APIs as well as 100+ other LLMs.[1] The JavaScript/TypeScript equivalent of this SDK is maintained at openai/openai-agents-js on GitHub.[1]

The openai-agents wheel is built from src/agents using Hatchling as the build backend.[2] The agents package's public API surface is defined in src/agents/__init__.py, which re-exports all major symbols including Agent, Runner, RunConfig, RunState, Handoff, guardrails, tools, tracing, sessions, MCP utilities, and retry/error-handler types.[3] SQLiteSession is lazily imported in src/agents/__init__.py via __getattr__ — it is only loaded when first accessed, not at module import time, to avoid the sqlite3 dependency cost.[3] All calls to set_default_openai_* and enable_verbose_stdout_logging in src/agents/__init__.py delegate to the internal _config module, keeping global state management centralized there.[3]

set_default_openai_key(key, use_for_tracing=True) in src/agents/__init__.py sets the OpenAI API key for both LLM requests and tracing; passing use_for_tracing=False requires a separate set_tracing_export_api_key() call or the OPENAI_API_KEY env var for traces.[3] set_default_openai_client(client, use_for_tracing=True) in src/agents/__init__.py replaces the default AsyncOpenAI instance for both LLM calls and trace uploads; set use_for_tracing=False to decouple the trace upload key from the client.[3] set_default_openai_api(api) in src/agents/__init__.py switches between "responses" (the default) and "chat_completions" for all OpenAI LLM requests.[3] set_default_openai_responses_transport(transport) in src/agents/__init__.py controls whether the Responses API uses HTTP (default) or WebSocket transport.[3] set_default_openai_harness(harness_id) in src/agents/__init__.py sets the default agent harness ID; passing None clears the override and restores the OPENAI_AGENT_HARNESS_ID environment variable fallback.[3]

Sources

2.2Installation and requirements

Updated 2026-08-28

The openai-agents SDK requires Python 3.10+ and a modern OpenAI client (v3+); core dependencies include pydantic, mcp, and websockets; additional features like voice, Redis sessions, sandboxes, and orchestration platforms are available via optional extras. Optional extras enable specialized capabilities: voice adds audio I/O, redis/encrypt/sqlalchemy provide session backends, docker/e2b/modal/temporal support different sandbox and workflow environments.

The current published version of openai-agents is 0.22.0, and the package is licensed under the MIT License.[1]

The openai-agents SDK requires Python 3.10 or newer (requires-python = ">=3.10") and supports CPython 3.10 through 3.14.[1]

The openai-agents package depends on openai>=3.0.0,<4, meaning the OpenAI Python v3/HTTPX2 client is the minimum supported version; the openai<3 compatibility shim was removed.[1] pydantic>=2.12.2,<3 is a required core dependency of openai-agents.[1] mcp>=1.19.0,<3 is a required core dependency of openai-agents for Python 3.10+.[1] websockets>=15.0,<17 is a required core dependency of openai-agents.[1]

Voice support requires the voice extra (pip install 'openai-agents[voice]'), which adds numpy>=2.2.0,<3 and websockets>=15.0,<17.[1] The any-llm optional extra requires Python 3.11+ (python_version >= '3.11'), unlike most other extras that support Python 3.10+; it pulls in any-llm-sdk>=1.11.0,<2.[1]

Redis session support requires the redis extra (pip install 'openai-agents[redis]'), which adds redis>=7.[1] The encrypt optional extra adds cryptography>=45.0,<46 for encrypted session storage.[1] The sqlalchemy optional extra adds SQLAlchemy>=2.0 and asyncpg>=0.29.0 for SQLAlchemy-backed session storage.[1]

The docker optional extra (pip install 'openai-agents[docker]') adds docker>=6.1 and enables DockerSandboxClient for sandbox workloads on Windows or any Docker-capable host.[1] The e2b optional extra pins exact versions: e2b==2.31.0 and e2b-code-interpreter==2.8.1.[1] The modal optional extra pins an exact version: modal==1.4.3.[1] The temporal optional extra pins exact versions: temporalio==1.26.0 and textual>=8.2.3,<8.3.[1]

Sources

pyproject.toml

2.3Quickstart and how-to

Updated 2026-08-28

The OpenAI Agents SDK lets you build and run agents that call language models and custom tools; a minimal agent needs only a name and instructions, and can be executed synchronously with Runner.run_sync(). Function tools are registered via the @tool decorator and passed to an Agent's tools= parameter, letting agents autonomously call your code during execution.

Install the Agents SDK with pip install openai-agents (or an equivalent package-manager command such as uv add openai-agents).[1] Before running any agent, set your OpenAI API key in the OPENAI_API_KEY environment variable (e.g. export OPENAI_API_KEY=sk-...).[1]

A minimal agent is constructed by passing at least name and instructions to Agent; a specific model may also be set at construction time.[1] Running that agent synchronously requires only Agent and Runner from the agents package — call Runner.run_sync(agent, "<prompt>") and read result.final_output.[2]

Function tools are defined with the @tool decorator (from agents.decorators) and passed in the tools= list of an Agent.

from agents.decorators import tool

@tool
def history_fun_fact() -> str:
    """Return a short history fact."""
    return "Sharks are older than trees."

agent = Agent(
    name="History Tutor",
    tools=[history_fun_fact],
)

[1]

Sources

2.4Examples and developer workflow

Updated 2026-08-28

The OpenAI Agents SDK uses uv for workspace management, ruff for linting and formatting, mypy and pyright for dual static type checking, and pytest with auto-async mode for testing, configured to enforce strict typing while permitting flexible package resolution. Custom pytest markers (allow_call_model_methods, requires_native_macos_sandbox, review_optional, serial) gate integration tests, sandbox-dependent tests, slow tests, and serialization constraints respectively. A uv workspace groups multiple related Python packages in one repository under a single lock file, enabling local inter-package dependency resolution without publishing to a registry.

The openai-agents project uses uv as its package and workspace manager and ruff as its linter and formatter, with documentation built via MkDocs and mkdocs-material.[1] The uv workspace configuration treats agents as a local workspace member, so uv add openai-agents resolves to the workspace package during development.[1] uv dependency resolution is configured with exclude-newer = "7 days" globally, with openai = false as a package-level exception so the openai package always resolves to the latest version regardless of the recency cutoff.[1]

Ruff lint rules enabled include E/W (pycodestyle), F (pyflakes), I (isort), B (bugbear), C4 (comprehensions), ASYNC, DTZ005 (timezone-naive datetime.now()), G004 (f-string in logging), RUF006 (unowned asyncio tasks), RUF012 (mutable class attributes without ClassVar), UP (pyupgrade), and others.[1] mypy is configured in strict mode (strict = true) but with disallow_incomplete_defs, disallow_untyped_defs, and disallow_untyped_calls all set to false.[1] pyright==1.1.408 is pinned in the dev dependency group alongside mypy for dual static analysis coverage.[1]

pytest is configured to discover tests in the tests/ directory; asyncio_mode = "auto" means all async tests run automatically without explicit marks.[1] Four custom pytest markers are registered: allow_call_model_methods (permits live model calls), requires_native_macos_sandbox (macOS sandbox-exec process required), review_optional (slow subsystem-specific test), and serial (requires exclusive post-xdist execution).[1]

Sources

pyproject.toml

3.1Handoffs

Updated 2026-08-28

A handoff transfers control from one agent to another as a callable tool — for example, routing from triage to billing specialists; the Handoff dataclass declares targets via the routing agent's handoffs= list and manages context, filtering, and dynamic enabling. Handoff mechanics include JSON schema exposure to the model, optional input filtering before delegation, conditional history nesting, and weak references to prevent circular dependencies.

A handoff is a directed transfer of control from one agent to another, exposing it as a tool the LLM can call — for example, a triage agent routing requests to billing or account-management specialists. The two multi-agent orchestration patterns are handoffs (specialist takes over the conversation) and agents-as-tools (orchestrator stays in control and calls specialists as tools).[1][2] An agent's outgoing handoff targets are declared via the handoffs= list on the routing agent; the Runner automatically executes those handoffs and subsequent tool calls.[2] handoff_description on a specialist Agent gives the routing agent context about when to delegate to that specialist.[2]

The Handoff dataclass in src/agents/handoffs/__init__.py stores a weak reference (_agent_ref) to the target agent when constructed via handoff(), preventing circular strong references.[1] Handoff.is_enabled accepts either a bool or a callable (RunContextWrapper, AgentBase) -> bool | Awaitable[bool], allowing dynamic enable/disable decisions based on runtime context or state.[1] Handoff.strict_json_schema defaults to True; the SDK strongly recommends keeping it True because it increases the likelihood of correct JSON input from the model.[1] Handoff.input_json_schema is the JSON schema exposed to the model as the handoff tool's parameters; it only describes the structured payload passed to on_invoke_handoff and does not replace the next agent's main input.[1]

Handoff.default_tool_name() derives the tool name from the agent name using transform_string_function_style(f"transfer_to_{agent.name}"), so an agent named "Billing Support" becomes transfer_to_billing_support.[1] Handoff.default_tool_description() combines the agent name with its handoff_description field to form the tool description shown to the model.[1]

HandoffInputFilter is a type alias for Callable[[HandoffInputData], MaybeAwaitable[HandoffInputData]] — the filter can be sync or async.[1] Handoff.input_filter (type HandoffInputFilter) receives the full conversation history including the trigger item and handoff tool output; the next agent receives input_items when set, otherwise new_items, enabling filtering without losing session history.[1] In streaming mode, Handoff.input_filter changes are not streamed — items generated before the handoff will already have been streamed. Server-managed conversations (conversation_id, previous_response_id, or auto_previous_response_id) do not support handoff input filters.[1] HandoffHistoryMapper is a type alias for Callable[[list[TResponseInputItem]], list[TResponseInputItem]], used to map a previous transcript to the nested summary payload.[1] Handoff.nest_handoff_history overrides the run-level nest_handoff_history setting for a single handoff; server-managed conversations automatically disable nested handoff history with a warning.[1]

The handoff() helper in src/agents/handoffs/__init__.py always returns the specific agent captured at call time; on_handoff is for side effects or bookkeeping, not dynamic destination selection.[1] _invoke_handoff_with_redaction in src/agents/handoffs/__init__.py intercepts ModelBehaviorError exceptions that carry redacted data, nulls out the context and input, and re-raises via _raise_data_redacted_error to prevent sensitive data from leaking in tracebacks.[1] HandoffInputData.clone() creates a copy with specified fields overridden and preserves internal _nested_history_owned_items from the original, ensuring history ownership is not lost on copy.[1]

Sources

3.2Guardrails

Updated 2026-08-28

Guardrails are validation checkpoints that intercept an agent's input or output; InputGuardrail runs before or alongside agent execution and can block the run, while OutputGuardrail validates the final result and raises a distinct exception if checks fail. Both guardrail types accept sync or async functions that inspect data and return a GuardrailFunctionOutput signaling whether a tripwire was triggered, with optional diagnostic metadata attached.

Guardrails are implemented in src/agents/guardrail.py as two types: InputGuardrail, which checks incoming input, and OutputGuardrail, which checks the final agent output.[1]

InputGuardrail accepts a run_in_parallel flag (default True): when True the guardrail runs concurrently with the agent; when False it runs before the agent starts.[1] When GuardrailFunctionOutput.tripwire_triggered is True for an InputGuardrail, the agent's execution immediately stops and an InputGuardrailTripwireTriggered exception is raised.[1] When GuardrailFunctionOutput.tripwire_triggered is True for an OutputGuardrail, an OutputGuardrailTripwireTriggered exception is raised — distinct from the input-guardrail exception.[1]

InputGuardrail.guardrail_function in src/agents/guardrail.py receives (RunContextWrapper[TContext], Agent[Any], str | list[TResponseInputItem]) and must return a GuardrailFunctionOutput or an awaitable of one.[1] OutputGuardrail.guardrail_function in src/agents/guardrail.py receives (RunContextWrapper[TContext], Agent[Any], Any) — where the third argument is the final agent output — and must return a GuardrailFunctionOutput or an awaitable of one.[1] GuardrailFunctionOutput.output_info is an arbitrary Any value the guardrail can populate with granular diagnostic information about the checks it performed.[1]

The @input_guardrail decorator in src/agents/guardrail.py can be applied directly to a function or called with keyword arguments name and run_in_parallel; both sync and async functions are accepted.[1] The @output_guardrail decorator in src/agents/guardrail.py can be applied directly to a function or called with a name keyword argument; both sync and async functions are accepted.[1]

InputGuardrail.get_name() and OutputGuardrail.get_name() in src/agents/guardrail.py return self.name when set, otherwise fall back to self.guardrail_function.__name__.[1] InputGuardrail.run() and OutputGuardrail.run() in src/agents/guardrail.py raise UserError if guardrail_function is not callable.[1]

Sources

src/agents/guardrail.py

3.4Result and items

Updated 2026-08-28

The result and items modules define the SDK's type aliases and data carriers for agent responses and conversation history: TResponse* types wrap OpenAI models or dicts, InputItem tracks occurrences by UUID for exactly-once semantics, and RunItemBase maintains weak/strong agent references to enable replay and post-execution inspection. Conversion between output and input items (via to_input_item() and helpers like _output_item_to_input_item) strips fields like created_by to produce replayable conversation history, while result reconciliation ensures nested agent invocations don't create duplicate ownership of history items.

src/agents/items.py defines TResponse, TResponseInputItem, TResponseOutputItem, and TResponseStreamEvent as type aliases for the corresponding OpenAI SDK types.[1] ToolSearchCallRawItem and ToolSearchOutputRawItem in src/agents/items.py are union type aliases that accept either the typed SDK model or a plain dict, supporting partial dict snapshots.[1] ToInputListMode in src/agents/result.py is a Literal type with two valid values: 'preserve_all' (keeps full converted history from new_items) and 'normalized' (returns canonical continuation input, falling back to full history when handoff filtering did not rewrite model history).[2]

InputItem in src/agents/items.py carries an input_id field — a UUID hex string generated at construction — described as "a durable occurrence identifier used for exactly-once conversation tracking".[1] RunItemBase in src/agents/items.py stores a weak reference to the agent alongside the strong reference, so callers can call release_agent() to drop the strong reference while still allowing the agent to be resolved via the weak reference.[1] RunItemBase.agent is accessed via a custom __getattribute__ that lazily resolves the weak reference when the strong reference has been released, so repr and dataclass.asdict continue to work after release_agent() is called.[1] RunItemBase.to_input_item() in src/agents/items.py converts any run item back into a TResponseInputItem suitable for replay to the model by delegating to _output_item_to_input_item.[1]

For shell_call_output items, _output_item_to_input_item in src/agents/items.py performs a two-level strip: it removes created_by from each nested content chunk inside the output list, requiring fresh chunk copies to avoid mutating the caller's mapping.[1] ToolSearchCallItem.to_input_item() and ToolSearchOutputItem.to_input_item() in src/agents/items.py both delegate to _tool_search_item_to_input_item, which pops the created_by field before returning the replayable input item.[1] coerce_tool_search_call_raw_item in src/agents/items.py prefers the typed ResponseToolSearchCall SDK model but tolerates partial dict snapshots by falling back to a raw dict when Pydantic validation fails, and raises AgentsException if the dict's type field is not "tool_search_call".[1]

result.last_agent.name on a RunResult identifies which agent produced the final answer in a multi-agent run.[3] _reconciled_result_owned_item_refs in src/agents/result.py filters _nested_history_owned_session_item_refs from a RunResultBase to retain only the references that match exact positions in the caller-supplied public_input, preventing double-ownership when input is rewritten.[2] _copy_pending_nested_agent_tool_states in src/agents/result.py binds detached nested approval checkpoints from the source result scope into the new outer checkpoint's scope, enabling HITL approval resume across nested agent-tool invocations.[2] src/agents/result.py imports _await_data_redacted_error_boundary, _detach_data_redacted_error_traceback, _is_error_data_redacted, and _should_drain_stream_events_before_raising from src/agents/exceptions.py, making the result layer a consumer of the error-redaction and stream-drain machinery — details of which live on the Exceptions and redaction page.[2]

Sources

3.5Exceptions and redaction

Updated 2026-08-28

The SDK isolates sensitive data in exceptions using a redaction boundary that wraps awaitable factories and detects cancellation signals marked with _RedactedExceptionCancellationError, enabling rejection of outputs before their traceback payloads can leak. Redaction implements defensive patterns — reading exception state via BaseException.__reduce__, comparing dict keys with identity-level string operations, and keeping the boundary itself synchronous — to avoid triggering attacker-controlled descriptors or leaving partially-sanitized state in async suspension.

src/agents/exceptions.py imports BaseExceptionGroup from the exceptiongroup backport package on Python < 3.11, and uses the built-in builtins.BaseExceptionGroup on Python 3.11+.[1] _RedactedExceptionCancellationError in src/agents/exceptions.py is a private exception class that inherits from both asyncio.CancelledError and Exception, making it catchable as a plain Exception while still acting as a cancellation signal.[1] _data_redacted_sync_cancellation_source in src/agents/run.py detects a _RedactedExceptionCancellationError wrapped inside another CancelledError — a Python 3.10 pattern — and returns the inner marked error, enabling safe error-boundary handling.[2]

_await_data_redacted_error_boundary in src/agents/exceptions.py accepts an awaitable_factory callable rather than a pre-built awaitable, so that the factory is invoked inside the try block, preventing accidental capture of payload data before the boundary is established.[1] OutputGuardrailBlockedMessageFormatter in src/agents/run_config.py is intentionally synchronous: awaiting application code at the redaction boundary can leave rejected output reachable through cancellation traceback locals or partially sanitized state, so async support requires a full redesign of the redaction boundary rather than simply awaiting the formatter result.[3]

_base_exception_instance_dict in src/agents/exceptions.py reads built-in exception instance state via BaseException.__reduce__ specifically to avoid invoking subclass attribute descriptors that could be attacker-controlled.[1] _exact_string_state_entry in src/agents/exceptions.py iterates the raw exception state dict using identity-level string comparison — type(candidate) is str and str.__eq__ — to avoid triggering any custom __hash__ or __eq__ on attacker-supplied keys.[1]

Sources

4.1RunState and resume

Updated 2026-08-28

RunState is a serializable snapshot of an in-progress agent run stored in src/agents/run_state.py that enables pause (HITL), approval flows, resume, and replay across run boundaries. RunState enforces strict schema versioning: only versions in SUPPORTED_SCHEMA_VERSIONS load; every shipped version must have a summary in SCHEMA_VERSION_SUMMARIES, and minimum versions guard specific features like programmatic tool calling (1.13) and hosted MCP approvals (1.14). HITL (Human-in-the-Loop) is a pattern that pauses agent execution at a defined checkpoint so a human can review or approve state before the run continues.

src/agents/run_state.py implements RunState, a serializable snapshot of an in-progress agent run that supports interruption (HITL pause), approval flows, resume, and cross-run replay.[1]

SUPPORTED_SCHEMA_VERSIONS in run_state.py is a frozenset of all schema version strings that can be read back; attempting to load a snapshot with a version outside this set causes a fail-fast error.[1] Every schema version ever shipped in a release must have a non-empty entry in SCHEMA_VERSION_SUMMARIES; an assertion fires at import time if any version is missing a summary.[1] RunState schema policy mandates that unreleased intermediate schema versions may be renumbered or squashed before release when their snapshots are intentionally unsupported.[1] Schema version 1.13 (_PROGRAMMATIC_TOOL_CALLING_MIN_SCHEMA_VERSION) is the minimum version required to deserialize programmatic tool calling and nested handoff history ownership.[1] Schema version 1.14 (_HOSTED_MCP_APPROVALS_MIN_SCHEMA_VERSION) is the minimum version required to deserialize scoped hosted MCP approvals and restored requests by server label.[1]

_PendingSessionWrite is a TypedDict in run_state.py representing one canonical resumed-output append awaiting acknowledgement, containing fields session_id, items, before, and persisted_count.[1] RunStateValidationError is a type alias for UserError | ValueError; validation failures in RunState deserialization surface as one of these two exception types.[1]

Sources

src/agents/run_state.py

4.2Sessions and memory

Updated 2026-08-28

The SDK supports three strategies for managing conversation history: full manual control via result.to_input_list(), SDK-managed persistence through Session implementations, or server-side state via OpenAI response IDs—allowing callers to choose based on their persistence needs. Session is a structural Protocol, not a class hierarchy; third-party implementations only need the session_id attribute and get_items(), pop_item(), add_items(), clear_session() methods, and may optionally implement run_compaction() for responses compaction support.

For multi-turn conversations, callers choose between result.to_input_list() (full manual history control, provider-agnostic), session=... (SDK-managed persistence), or previous_response_id/conversation_id (OpenAI server-side state).[1]

Session in src/agents/memory/session.py is a @runtime_checkable Protocol, meaning third-party implementations only need to satisfy the structural interface — they are NOT required to inherit from any base class.[2] Every Session implementation must expose a session_id: str attribute and an optional session_settings: SessionSettings | None (defaults to None).[2] Session.get_items(limit) retrieves conversation history; when limit is specified it returns the latest N items in chronological order; when None it returns all items.[2] Session.pop_item() removes and returns the most recent item from the session, or None if the session is empty.[2] SessionABC in src/agents/memory/session.py is an abstract base class intended for internal use and as a base for concrete implementations; third-party libraries should implement the Session Protocol instead.[2]

OpenAIResponsesCompactionAwareSession is a @runtime_checkable Protocol that extends Session with a run_compaction(args) method for sessions that support responses compaction.[2] OpenAIResponsesCompactionArgs.compaction_mode has three values: "auto" (use input when the last response was not stored), "previous_response_id" (use server-managed response history), and "input" (send locally stored session items as input).[2] When OpenAIResponsesCompactionArgs.store is False, compaction should avoid "previous_response_id" mode unless explicitly requested.[2] is_openai_responses_compaction_aware_session() in src/agents/memory/session.py checks for compaction support by looking for a callable run_compaction attribute on the session; it returns False for None or any session where attribute access raises.[2]

Custom Session implementations can opt into receiving the RunContextWrapper by adding a wrapper keyword parameter to all four history methods (get_items, add_items, pop_item, clear_session); the public Session Protocol does not include wrapper so existing structural implementations remain type-compatible.[2] _session_accepts_wrapper() in src/agents/memory/session.py requires ALL four history methods to accept wrapper — if any one method lacks it, wrapper is not passed to any method.[2] _get_session_wrapper() in src/agents/memory/session.py returns None (suppressing context propagation) whenever wrapper is None OR the session does not have a complete context-aware contract (all four methods accept wrapper).[2] _call_session_method() in src/agents/memory/session.py transparently handles both sync and async session method implementations — it awaits the result only when inspect.isawaitable returns True.[2]

Sources

4.3Sandbox

Updated 2026-08-28

A SandboxAgent is a specialized agent for tasks that require isolated file-system access; it extends Agent with sandbox-specific fields (default_manifest, capabilities, run_as, base_instructions) while keeping transport details transport-agnostic via RunConfig. The sandbox manifests, environment resolution, and entry types form a declarative system for configuring workspace mounts, file permissions, environment variables, and user identity — all validated against exposure policies and deserialized polymorphically at runtime.

The recommended pattern is to use a plain Agent plus Runner for prompt/tool/conversation tasks, and switch to SandboxAgent only when the agent must inspect or modify real files in an isolated workspace.[1]

SandboxAgent, defined in src/agents/sandbox/sandbox_agent.py, is a subclass of Agent[TContext] that adds sandbox-specific fields (default_manifest, base_instructions, capabilities, run_as) to the base Agent interface.[2] Runtime transport details — sandbox client, client options, and live session — are NOT stored on SandboxAgent itself; they are provided at run time through RunConfig(sandbox=...), keeping the agent declaration transport-agnostic.[2]

SandboxAgent.default_manifest holds the default sandbox Manifest applied when Runner creates a new session; it defaults to None.[2] SandboxAgent.capabilities is a Sequence[Capability] (defaulting to Capabilities.default()) whose entries can mutate the manifest, add instructions, and expose tools.[2] SandboxAgent.run_as sets the user identity presented to model-facing sandbox tools such as shell, file reads, and patches; it accepts a User object, a plain string, or None.[2] SandboxAgent.base_instructions overrides the SDK sandbox base prompt and accepts a string, an async or sync callable (RunContextWrapper, Agent) -> str | None, or None; most callers should use instructions instead.[2]

SandboxAgent.__post_init__ coerces a raw dict passed as default_manifest into a typed Manifest via _coerce_manifest, so callers may supply either type.[2] SandboxAgent.__post_init__ coerces a raw dict passed as run_as into a User Pydantic model via coerce_pydantic_config, so callers may pass a plain dict.[2] SandboxAgent.__post_init__ raises TypeError if base_instructions is not a string, callable, or None, giving an explicit error message that names the invalid type.[2] SandboxAgent.__post_init__ raises TypeError if run_as is not a string, User, or None, giving an explicit error message that names the invalid type.[2] SandboxAgent._sandbox_concurrency_guard is an internal field excluded from __init__ and repr, used to guard concurrent sandbox access; it is not part of the public API.[2]

Manifest in src/agents/sandbox/manifest.py is a Pydantic BaseModel with a fixed version: Literal[1] = 1, a root defaulting to /workspace, and fields for entries, environment, users, groups, extra_path_grants, and remote_mount_command_allowlist.[3] Manifest.remote_mount_command_allowlist defaults to a fixed list of safe read/write commands: ls, find, stat, cat, less, head, tail, du, grep, rg, wc, sort, cut, cp, tee, echo, mkdir, rm.[3] Manifest entries are deserialized from raw mappings using BaseEntry.parse(entry) inside the _parse_entries field validator, supporting polymorphic entry types.[3] JSON serialization of Manifest entries converts Path keys to their POSIX string representation via .as_posix(), ensuring cross-platform compatibility in serialized output.[3]

Manifest actively rejects any input containing mount-credential-exposure policy keys (e.g. mount_credential_exposure_policy, inContainerMountCredentialExposureAllowedPaths) via a model_validator, enforcing that this policy can only be set on a trusted Manifest instance in code, not via untrusted input.[3]

EnvValue is an abstract Pydantic BaseModel in src/agents/sandbox/manifest.py serving as an extension point for environment variable sources; concrete subtypes register themselves via __pydantic_init_subclass__ under a string type discriminator.[3] EnvValue.parse() accepts an existing EnvValue instance unchanged, a plain {"value": str} mapping (coerced to StrEnvValue), or a typed mapping dispatched to the registered subclass; it raises ValueError for unknown type strings.[3] StrEnvValue is the built-in EnvValue subtype registered under type="str" in src/agents/sandbox/manifest.py; its resolve() simply returns self.value as a string.[3] EnvEntry in src/agents/sandbox/manifest.py wraps an EnvValue with optional description and ephemeral (default False) metadata fields.[3] Environment.resolve() in src/agents/sandbox/manifest.py concurrently resolves all env values using gather_with_cancel rather than a bare asyncio.gather, so that if one lookup fails the remaining sibling coroutines are cancelled rather than left running.[3]

UnixLocalSandboxClient is supported only on macOS and Linux; on Windows, DockerSandboxClient (with the openai-agents[docker] extra) or a hosted sandbox client must be used instead.[4] SandboxConcurrencyLimits in src/agents/run_config.py defaults to 4 concurrent manifest entries (manifest_entries=4) and 4 concurrent file copies per local_dir entry (local_dir_files=4); either can be set to None to remove the limit.[5] SandboxArchiveLimits in src/agents/run_config.py defaults to a 1 GiB input size limit, a 4 GiB extracted-bytes limit, and a 100,000-member limit; all three can be set to None to disable that specific limit.[5]

Example: minimal sandbox agent run using SandboxAgent, UnixLocalSandboxClient, and Runner.run_sync with a SandboxRunConfig.

from agents import Runner
from agents.run import RunConfig
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
from agents.sandbox.entries import GitRepo
from agents.sandbox.sandboxes import UnixLocalSandboxClient

agent = SandboxAgent(
    name="Workspace Assistant",
    instructions="Inspect the sandbox workspace before answering.",
    default_manifest=Manifest(entries={"repo": GitRepo(repo="openai/openai-agents-python", ref="main")}),
)

result = Runner.run_sync(
    agent,
    "Inspect the repo README and summarize what this project does.",
    run_config=RunConfig(sandbox=SandboxRunConfig(client=UnixLocalSandboxClient())),
)
print(result.final_output)

[4]

Sources

5.1Tracing

Updated 2026-08-28

Agent run traces are viewable at https://platform.openai.com/traces in the OpenAI Dashboard after a run completes.[1] set_tracing_disabled(disabled) in src/agents/tracing/__init__.py globally enables or disables tracing by calling get_trace_provider().set_disabled(disabled).[2] trace_include_sensitive_data defaults to True but can be overridden by setting the OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA environment variable to 0, false, no, or off.[3]

add_trace_processor(span_processor) in src/agents/tracing/__init__.py registers an additional TracingProcessor with the global trace provider; it calls get_trace_provider().register_processor(span_processor) and is additive (does not replace existing processors).[2] set_trace_processors(processors) in src/agents/tracing/__init__.py replaces the entire current list of trace processors on the global provider; use add_trace_processor instead to append without clearing existing ones.[2] flush_traces() in src/agents/tracing/__init__.py forces immediate export of buffered traces; the default BatchTraceProcessor exports periodically in the background, so flush_traces() is needed when a worker or request handler needs traces visible immediately after a unit of work finishes.[2] set_tracing_export_api_key(api_key) in src/agents/tracing/__init__.py sets the OpenAI API key used by the default backend exporter, calling default_exporter().set_api_key(api_key).[2] BatchTraceProcessor in src/agents/tracing/processors.py collects spans in memory and exports them to a SpanExporter (such as BackendSpanExporter) in periodic background batches rather than one at a time, reducing export overhead.

The Span context-manager protocol in src/agents/tracing/spans.py calls start(mark_as_current=True) in __enter__ and finish(reset_current=True) in __exit__, providing automatic span lifecycle management; using context managers is the documented approach for reliable start/finish.[4] Span.start(mark_as_current=True) in src/agents/tracing/spans.py pushes the span as the context-var current span; finish(reset_current=True) pops it. The mark_as_current and reset_current flags are separate parameters, both defaulting to False.[4] SpanImpl.start() in src/agents/tracing/spans.py calls TracingProcessor.on_span_start(self) and records an ISO timestamp; calling it a second time only logs a warning and returns without effect.[4] SpanImpl.finish() in src/agents/tracing/spans.py calls TracingProcessor.on_span_end(self) and records an ISO timestamp; calling it a second time only logs a warning and returns without effect.[4] Span.__exit__ in src/agents/tracing/spans.py detects GeneratorExit and delegates to _finish_on_generator_exit instead of the normal finish(reset_current=True) path, to handle abandoned async generators safely.[4] _finish_on_generator_exit in src/agents/tracing/spans.py silently swallows ValueError from Scope.reset_current_span because an abandoned async generator may be finalized from a different task whose context never set the token — raising would add a crash on top of an already-unwinding generator.[4]

The Span abstract base class in src/agents/tracing/spans.py exposes started_at and ended_at as ISO-format timestamp strings (or None if not yet started/finished), and tracing_api_key for export authentication.[4] SpanImpl in src/agents/tracing/spans.py accepts an optional trace_metadata dict stored and exposed via the inherited trace_metadata property, enabling trace-level metadata to propagate to individual spans.[4] When span_id is not provided to SpanImpl.__init__ in src/agents/tracing/spans.py, a new span ID is auto-generated via util.gen_span_id().[4] Both NoOpSpan and SpanImpl in src/agents/tracing/spans.py manage the current-span context via Scope.set_current_span (on start) and Scope.reset_current_span (on finish), using a stored _prev_span_token.[4] SpanError in src/agents/tracing/spans.py is a TypedDict with a message string and an optional data dict for attaching arbitrary error context to a span. The canonical pattern for recording errors is to call span.set_error({"message": str(e), "data": {...}}) inside an except block, then re-raise.[4] NoOpSpan in src/agents/tracing/spans.py is a no-op Span implementation used when tracing is disabled; export() returns None, error is always None, and trace_id/span_id both return the literal string "no-op".[4]

BackendSpanExporter in src/agents/tracing/processors.py posts traces to https://api.openai.com/v1/traces/ingest by default (the endpoint constructor parameter), and adds the OpenAI-Beta: traces=v1 header to every export request.[5] BackendSpanExporter resolves the API key at first access from os.environ["OPENAI_API_KEY"] if not set in the constructor (cached via @cached_property); calling set_api_key() clears the cache so the new value takes effect.[5] BackendSpanExporter.set_api_key() in src/agents/tracing/processors.py manually deletes the api_key entry from __dict__ to invalidate the @cached_property before storing the new key.[5] BackendSpanExporter in src/agents/tracing/processors.py reads the OpenAI organization from os.environ["OPENAI_ORG_ID"] and the project from os.environ["OPENAI_PROJECT_ID"] when those fields are not supplied to the constructor.[5] BackendSpanExporter skips exporting an entire group and logs a warning when no API key is resolvable for that group.[5] BackendSpanExporter._export_with_deadline in src/agents/tracing/processors.py groups items by their tracing_api_key and issues a separate HTTP POST per group, allowing multi-tenant exports in a single call.[5] BackendSpanExporter in src/agents/tracing/processors.py keeps a persistent httpx2.Client with a 60-second read timeout and a 5-second connect timeout, enabling connection pooling across export calls.[5] BackendSpanExporter in src/agents/tracing/processors.py defaults to max_retries=3, base_delay=1.0 second, and max_delay=30.0 seconds for exponential backoff on failed exports.[5] BackendSpanExporter in src/agents/tracing/processors.py treats HTTP 4xx responses as non-retryable and logs an error, while 5xx or unexpected codes are retried up to max_retries.[5] BackendSpanExporter in src/agents/tracing/processors.py applies exponential backoff with 10% jitter between retries: sleep_time = delay + random.uniform(0, 0.1 * delay), doubling delay each attempt up to max_delay.[5] BackendSpanExporter._sleep_before_retry in src/agents/tracing/processors.py interrupts its sleep early and returns False if the shutdown event fires, abandoning the remaining retries cleanly on shutdown.[5] BackendSpanExporter sanitizes payloads before sending to the OpenAI ingest endpoint: it truncates input/output fields to 100,000 bytes (with a ... [truncated] suffix) and drops the usage key from non-generation span types.[5] Sanitization in BackendSpanExporter (src/agents/tracing/processors.py) is only applied when the configured endpoint matches https://api.openai.com/v1/traces/ingest; custom endpoints receive the raw payload.[5]

ConsoleSpanExporter in src/agents/tracing/processors.py prints trace and span data to stdout, but redacts all content to a short notice string when either _debug.DONT_LOG_MODEL_DATA or _debug.DONT_LOG_TOOL_DATA is set.[5]

Sources

5.2Testing

Updated 2026-08-28

ScriptedModel is a deterministic, provider-neutral Model for testing agents without live calls; it records each model call's full boundary and lets you script responses, errors, or streams with factory methods and optional dynamic responders. Tests use ScriptedModel to enqueue or extend scripted steps, inspect recorded calls via snapshots, validate all interactions were exercised with assert_complete(), and diagnose misconfigurations through structured error types.

ScriptedModel in src/agents/testing/model.py is a deterministic, provider-neutral Model implementation for testing agent workflows without live provider calls, introduced in v0.21.0.[1] ScriptedModel.__init__ accepts an iterable of ModelScriptItem steps (each a ModelStep, ModelStepSpec dict, ModelResponse, output-item sequence, or Exception), an emit_traces bool, and a default_usage that applies when a step does not supply its own usage.[1]

ModelCall in src/agents/testing/model.py is a frozen dataclass capturing the full provider-neutral boundary of one model call: system_instructions, input, model_settings, tools, output_schema, handoffs, tracing, previous_response_id, conversation_id, prompt, and streamed.[1] ScriptedModel.calls returns detached deep-copy snapshots of every recorded model call, so mutations after the fact do not corrupt the record.[1] ScriptedModel.first_call and ScriptedModel.last_call return detached snapshots of the first and most recent recorded calls, or None if no calls have been made yet.[1] ScriptedModel.remaining_steps returns the count of scripted model steps not yet consumed.[1]

ScriptedModel.enqueue() appends a single step and ScriptedModel.extend() appends multiple steps, allowing incremental test setup after construction.[1] ScriptedModel.assert_complete() raises UnconsumedModelSteps if any configured steps remain after a test, letting tests enforce that all scripted interactions were exercised.[1]

ModelStep in src/agents/testing/model.py has a default response_id of "resp-789", so tests that do not set a response ID will see that sentinel value.[1] ModelStep.raise_error() is a factory that creates a step which raises a given exception, optionally with provider retry guidance via retry_advice.[1] ModelStep.respond() is a factory that creates a step whose result is dynamically derived from the recorded ModelCall via a ModelResponder callable.[1] ModelStep.stream() is a factory for streaming tests that accepts either a sequence of TResponseStreamEvent objects or a ModelStreamFactory callable, plus optional output, usage, and response_id.[1] ModelResponder is a type alias for a callable that takes a ModelCall and returns either a ModelStepResult or an Awaitable[ModelStepResult], so responders may be async.[1] ModelStreamFactory is a type alias for a callable that takes a ModelCall and returns an AsyncIterator[TResponseStreamEvent], used to supply dynamic streaming event sequences.[1] ModelStepResult in src/agents/testing/model.py is the resolved outcome of a scripted step — either a model response or a raised exception — at the provider-neutral boundary; it is the return type required by ModelResponder and produced internally by ModelStreamFactory.

ScriptedModel.get_retry_advice() returns retry advice only when the request.error object is the exact same exception instance that was configured on the step — identity-checked via id() — preventing accidental advice leakage.[1]

UnexpectedModelCall carries the ModelCall object and its index (call_index) for diagnosis when a test drives more model calls than were scripted.[1] InvalidModelStep carries a reason field (a ModelStepReason literal) and the input_index of the offending step, aiding diagnosis of misconfigured scripts.[1] ModelStepReason in src/agents/testing/model.py is a Literal type alias enumerating the reasons a step can be rejected: "invalid_input", "unsupported_field", "invalid_error", "invalid_responder", "invalid_stream_events", "conflicting_outcomes", "invalid_retry_advice".[1]

Use ScriptedModel with enqueue to script model responses, then run an agent with duplicate input items via Runner.run

model = ScriptedModel()
model.enqueue([get_text_message("done")])
agent = Agent(name="test", model=model)

input_items = [get_text_input_item("repeat"), get_text_input_item("repeat")]

await Runner.run(agent, input=input_items)

[2]

Sources

5.3Upgrading

Updated 2026-08-28

This page covers testing and development utilities for the OpenAI Agents SDK: helper functions for interactive mode fallbacks and citation extraction, scripted model testing with ScriptedModel.extend, demo loop execution patterns, and logging/test suite configuration.

Use input_with_fallback to return a preset string instead of prompting when EXAMPLES_INTERACTIVE_MODE=auto is set

from examples.auto_mode import input_with_fallback

user_input = input_with_fallback("Enter your question: ", "What is the weather today?")

[1]

Use confirm_with_fallback to auto-approve confirmations when EXAMPLES_INTERACTIVE_MODE=auto is set

from examples.auto_mode import confirm_with_fallback

approved = confirm_with_fallback("Proceed with action? (y/n): ", default=True)

[1]

Extract deduplicated URLCitation objects (title + URL) from a sequence of run output items using extract_url_citations

from examples.web_search_utils import extract_url_citations

citations = extract_url_citations(result.new_items)
for citation in citations:
    print(citation.title, citation.url)

[2]

Run all auto-mode examples via Make, optionally filtering by substring or enabling extra categories

# Run all examples
make examples-run

# Run only examples whose path contains "basic"
make examples-run EXAMPLES_ARGS="--filter basic"

# Include server and audio examples
make examples-run EXAMPLES_ARGS="--include-server --include-audio"

[3]

Run a multi-turn non-streaming demo loop with ScriptedModel.extend providing per-turn responses, feeding simulated user input via monkeypatched builtins.input

model = ScriptedModel()
model.extend([[get_text_message("hello")], [get_text_message("good")]])
agent = Agent(name="test", model=model)

inputs = iter(["Hi", "How are you?", "quit"])
monkeypatch.setattr("builtins.input", lambda _=" > ": next(inputs))

await run_demo_loop(agent, stream=False)

[4]

Run a streaming demo loop that exercises tool calls, tool outputs, and agent handoffs using ScriptedModel.extend with multi-step scripted responses

model = ScriptedModel()
agent = Agent(
    name="test",
    model=model,
    tools=[get_function_tool("foo", "tool_result")],
    handoffs=[target_agent],
)
model.extend(
    [
        [get_function_tool_call("foo", "{}")],
        [get_handoff_tool_call(target_agent)],
        [get_text_message("all done")],
    ]
)
inputs = iter(["Hello", "exit"])
monkeypatch.setattr("builtins.input", lambda _=" > ": next(inputs))
await run_demo_loop(agent, stream=True)

[4]

ScriptedModel is a test double that replaces a real OpenAI model with a pre-programmed sequence of responses, enabling deterministic unit tests without live API calls or network access.

Control model-data and tool-data logging with OPENAI_AGENTS_DONT_LOG_MODEL_DATA / OPENAI_AGENTS_DONT_LOG_TOOL_DATA env vars; the loaders default to True when the variable is absent

# Unset → True (logging suppressed by default)
_load_dont_log_model_data()   # True

# Explicit "0" or "false" → False (logging enabled)
# Explicit "1" or "true"  → True (logging suppressed)
_load_dont_log_tool_data()    # True when OPENAI_AGENTS_DONT_LOG_TOOL_DATA unset

[5]

Run the full test suite (xdist parallel + serial) with make tests; fix or create inline snapshots with dedicated make targets

make tests          # shard-safe parallel run then serial tests
make snapshots-fix  # fix broken inline-snapshot assertions
make snapshots-create  # create new inline-snapshot assertions

[6]

Sources

6Agents

Updated 2026-08-28

Agents are built on a shared AgentBase class that manages tools (both FunctionTool and MCP tools), validates tool names to prevent collisions, and handles model-specific defaults and handoff snapshots. Agent execution can be paused at specific tools via StopAtTools, directed to a final output via ToolsToFinalOutputResult, or streamed as AgentToolStreamEvents that capture tool calls and their nested agent runs. MCP (Model Context Protocol) is a standard interface for connecting agents to external tool servers; agents discover and invoke tools exposed by mcp_servers instances at runtime, separately from locally defined FunctionTools.

AgentBase in src/agents/agent.py is the shared base class for both Agent and RealtimeAgent, providing name, handoff_description, tools, mcp_servers, and mcp_config fields, as well as MCP tool retrieval and tool-enable logic.[1]

AgentBase.get_all_tools in src/agents/agent.py evaluates each FunctionTool's is_enabled attribute — which may be a bool or a callable returning MaybeAwaitable[bool] — concurrently via gather_with_cancel, then appends the enabled tools to MCP tools and calls prune_orphaned_tool_search_tools.[1] _validate_codex_tool_name_collisions in src/agents/agent.py raises UserError if any Codex tool (a FunctionTool with _is_codex_tool=True) shares its name with another tool in the list; the error message names the duplicate tools and instructs the caller to provide a unique codex_tool(name=...) per instance.[1]

AgentBase._use_mcp_handoff_snapshot in src/agents/agent.py is a context manager that stores a snapshot of enabled handoffs in the _mcp_handoff_snapshot ContextVar, so that reserved-name generation uses a single consistent set of handoffs for the duration of the context.[1]

_initial_model_settings_for_model in src/agents/agent.py returns model-specific defaults via get_default_model_settings(model_str) when a string model name is provided, per-model defaults when None, and an empty ModelSettings() when a Model object is passed directly.[1]

ToolsToFinalOutputResult in src/agents/agent.py carries two fields: is_final_output (bool) and final_output (Any, defaults to None). When is_final_output is False the LLM runs again with the tool output; when True, final_output must match the agent's output_type.[1] StopAtTools (TypedDict) in src/agents/agent.py contains a single key stop_at_tool_names: list[str]; any tool whose name appears in that list halts further agent execution.[1] AgentToolStreamEvent (TypedDict) in src/agents/agent.py carries three fields: event (the StreamEvent from the nested run), agent (the nested Agent), and tool_call (the originating ResponseFunctionToolCall, or None).[1]

Sources

src/agents/agent.py

6.1Agent types

Updated 2026-08-28

The OpenAI Agents SDK provides four agent types—Agent, SandboxAgent, RealtimeAgent, and VoicePipeline—to cover text, isolated execution, WebSocket-based voice/multimodal, and voice workflow use cases. Text agents run synchronously; realtime and voice agents handle streaming audio/multimodal events asynchronously, differing in connection model (RealtimeAgent uses RealtimeRunner, VoicePipeline processes AudioInput buffers).

The openai-agents-python SDK supports four primary agent types: the text Agent, SandboxAgent (isolated workspace), RealtimeAgent (WebSocket voice/multimodal), and VoicePipeline for voice workflows.[1]

Example: minimal realtime agent loop using RealtimeAgent, RealtimeRunner, and async event iteration.

import asyncio
from agents.realtime import RealtimeAgent, RealtimeRunner

async def main() -> None:
    agent = RealtimeAgent(name="Assistant", instructions="You are a helpful voice assistant. Keep responses short.")
    runner = RealtimeRunner(starting_agent=agent)
    session = await runner.run()

    async with session:
        await session.send_message("Say hello in one short sentence.")
        async for event in session:
            if event.type == "audio":
                pass
            elif event.type == "history_added":
                print(event.item)
            elif event.type == "agent_end":
                break

[1]

Example: minimal voice pipeline run using VoicePipeline, SingleAgentVoiceWorkflow, and AudioInput.

from agents import Agent
from agents.voice import AudioInput, SingleAgentVoiceWorkflow, VoicePipeline

async def main() -> None:
    agent = Agent(name="Assistant", instructions="You are a helpful voice assistant.")
    pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent))
    audio_input = AudioInput(buffer=np.zeros(24000 * 3, dtype=np.int16))
    result = await pipeline.run(audio_input)
    async for event in result.stream():
        if event.type == "voice_stream_event_audio":
            pass

[1]

Sources

README.md

7.1Tool catalog

Updated 2026-08-28

The tool catalog organizes five tool categories—hosted OpenAI tools (web search, file search, code interpreter), local/runtime tools (ComputerTool, ShellTool), FunctionTool instances, agents as tools, and experimental tools—each with distinct availability, configuration, and output requirements. Tool outputs are validated through Pydantic models (ToolOutputText, ToolOutputImage, ToolOutputFileContent), and tools track their runtime origin (FUNCTION, MCP, AGENT_AS_TOOL) via serializable ToolOrigin metadata.

docs/tools.md catalogs five tool categories: hosted OpenAI tools (web search, file search, code interpreter, hosted MCP, image generation), local/runtime tools (ComputerTool, ApplyPatchTool, ShellTool), FunctionTool instances, agents as tools, and an experimental Codex tool.[1] Hosted tools (WebSearchTool, FileSearchTool, CodeInterpreterTool, HostedMCPTool, ImageGenerationTool, ToolSearchTool, ProgrammaticToolCallingTool) are available only when using OpenAIResponsesModel.[1]

WebSearchTool supports filters, user_location, and search_context_size options.[1] FileSearchTool supports filters, ranking_options, include_search_results, vector_store_ids, and max_num_results; max_num_results accepts integers 1–50, and None or zero uses the provider default.[1]

Canonical usage of WebSearchTool and FileSearchTool with an Agent:

from agents import Agent, FileSearchTool, Runner, WebSearchTool

agent = Agent(
    name="Assistant",
    tools=[
        WebSearchTool(),
        FileSearchTool(
            max_num_results=3,
            vector_store_ids=["VECTOR_STORE_ID"],
        ),
    ],
)

async def main():
    result = await Runner.run(agent, "Which coffee shop should I go to, taking into account my preferences and the weather today in SF?")
    print(result.final_output)

[1]

src/agents/tool.py defines three tool function signatures: ToolFunctionWithoutContext (no context), ToolFunctionWithContext (first arg is RunContextWrapper), and ToolFunctionWithToolContext (first arg is ToolContext), all unified as ToolFunction.[2] ToolCaller in src/agents/tool.py is a Literal type with values "direct" and "programmatic", distinguishing calls made directly by the model from those made via Programmatic Tool Calling — see Programmatic tool calling.[2]

ValidToolOutputPydanticModels in src/agents/tool.py is the union ToolOutputText | ToolOutputImage | ToolOutputFileContent, and a pre-built TypeAdapter named ValidToolOutputPydanticModelsTypeAdapter handles efficient Pydantic validation of tool outputs.[2] ToolOutputImage in src/agents/tool.py requires at least one of image_url or file_id; providing neither raises a ValueError via a Pydantic model_validator.[2] ToolOutputImage.detail accepts "low", "high", or "auto" to control vision detail level, and is optional (defaults to None).[2] ToolOutputFileContent in src/agents/tool.py requires at least one of file_data (base64), file_url, or file_id; providing none raises a ValueError via a Pydantic model_validator.[2]

ToolOriginType in src/agents/tool.py is a str Enum with three values — FUNCTION, MCP, and AGENT_AS_TOOL — indicating the runtime source of a function-tool-backed run item.[2] ToolOrigin in src/agents/tool.py is a frozen dataclass carrying serializable metadata (type, optional MCP server name, agent name, agent tool name) about where a function-tool-backed item originated, and round-trips via to_json_dict() / from_json_dict().[2] ToolOrigin.from_json_dict() returns None (rather than raising) when the input is not a Mapping, lacks a "type" key, or contains an unrecognized ToolOriginType value.[2]

DEFAULT_APPROVAL_REJECTION_MESSAGE in src/agents/tool.py is "Tool execution was not approved." — the default message returned to the model when a tool call is rejected by the approval flow.[2] ToolTimeoutBehavior in src/agents/tool.py is a Literal type with two values: "error_as_result" (return the error as the tool result) or "raise_exception" (propagate the ToolTimeoutError).[2]

FunctionToolCustomDataContext, CustomToolCustomDataContext, ComputerToolCustomDataContext, and ApplyPatchToolCustomDataContext in src/agents/tool.py are frozen dataclasses passed to custom data extractor callbacks, each carrying the invocation context, the tool object, the model-visible output, and the raw replay item.[2]

Sources

7.2Function schema

Updated 2026-08-29

FuncSchema in src/agents/function_schema.py captures the JSON schema and Pydantic model for a Python function's parameters, and is the data structure used to present a Python function to an LLM as a tool.[1] FuncSchema.takes_context marks whether the function's first argument is a RunContextWrapper; to_call_args skips that first parameter when building the positional args list.[1] FuncSchema.return_annotation stores the resolved return annotation including Annotated metadata, defaulting to inspect.Signature.empty when no annotation is present.[1] FuncSchema in src/agents/function_schema.py raises an error at schema-construction time when a variadic tool argument (e.g., *args) is annotated with a fixed-length tuple type (e.g., tuple[int, str]), preventing schema misrepresentation and runtime failures from malformed JSON Schema.

FuncSchema.strict_json_schema defaults to True; the SDK strongly recommends keeping it True because strict mode increases the likelihood of the LLM producing correctly-structured JSON input for the tool.[1]

FuncSchema.to_call_args reads parameter values from object.__getattribute__(data, "__dict__") before falling back to getattr, so that Pydantic properties like model_extra and model_fields_set cannot shadow tool parameters of the same name.[1] FuncSchema.to_call_args raises ModelBehaviorError when a **kwargs payload contains a key that also names a POSITIONAL_OR_KEYWORD or KEYWORD_ONLY parameter, because such a key would either replace a validated value or cause a Python "got multiple values for argument" error.[1] Positional-only parameters and *args are deliberately not reserved when checking **kwargs collisions in FuncSchema._raise_on_var_keyword_collisions: for def f(a, /, **kw), the call f(1, a=2) is legal and routes a=2 into kw.[1]

_detect_docstring_style in src/agents/function_schema.py heuristically scores a docstring for sphinx, numpy, and google styles using regex patterns; in a tie, priority is sphinx > numpy > google. When no patterns match, it defaults to "google".[1] The _GOOGLE_SECTION_HEADER_RE regex in src/agents/function_schema.py matches the aliases args, arguments, params, parameters (case-insensitive) as full-line Google parameter section headers, anchored at column 0.[1] _ensure_blank_line_before_google_sections in src/agents/function_schema.py inserts a blank line before a Google-style Args: (or alias) section header when one is missing, working around a griffe parser bug that silently drops parameter descriptions in that case. The original string object is returned unchanged when no insertion is needed.[1]

Sources

src/agents/function_schema.py

7.3Programmatic tool calling

Updated 2026-08-28

Programmatic Tool Calling allows an OpenAI Responses model to generate and execute JavaScript that orchestrates tool calls with loops, branching, and intermediate calculations, returning a final result without pausing after each invocation. The generated program runs isolated in a V8 sandbox with access only to explicitly allowed tools; tool authors use allowed_callers to control whether tools are invocable by the model directly, the program, or both. Programmatic Tool Calling reduces latency and token overhead by completing multi-step tool workflows in a single model turn, eliminating intermediate request-response cycles where tool results would otherwise be sent back to the model to generate a new response.

Programmatic Tool Calling lets a supported OpenAI Responses model generate JavaScript that calls eligible tools, combines their outputs, and returns one result to the model — useful for bounded workflows that benefit from loops, branching, parallel calls, or intermediate calculations without a model round trip after every tool call.[1] The generated JavaScript program runs in a fresh hosted V8 environment with no Node.js APIs, filesystem or network access, or persistent process; it can interact only with tools explicitly allowed.[1]

ProgrammaticToolCallingTool() and tool_choice="programmatic_tool_calling" are available only with supported OpenAI Responses models; Chat Completions models and non-Responses backends reject both.[1] An agent must include at most one ProgrammaticToolCallingTool() instance and must also expose at least one programmatically callable tool, a ToolSearchTool() backed by a namespace, deferred function, or deferred hosted MCP server, or an opaque prompt-managed tool surface; a bare ToolSearchTool() without a searchable surface is rejected.[1]

The allowed_callers field on a tool controls invocation mode: omitting it allows direct model calls only; ["programmatic"] restricts the tool to program-only access; ["direct", "programmatic"] allows both.[1] SDK tool types that support allowed_callers are FunctionTool, CustomTool, ShellTool, ApplyPatchTool, HostedMCPTool, and CodeInterpreterTool; for HostedMCPTool and CodeInterpreterTool, the field is set inside tool_config.[1] For @function_tool(allowed_callers=[...]), a structured return annotation (Pydantic model, TypedDict, or dataclass) automatically becomes a strict object output schema validated before the value is returned to the program; output_type and output_json_schema are mutually exclusive alternatives when no usable annotation exists.[1]

For schema-backed program-owned tool calls, the default failure formatter is disabled because its free-form text does not satisfy the output schema; a handler exception propagates unless a custom failure_error_function returning schema-conforming JSON is provided.[1] Program-owned SDK tools still run through the normal Runner lifecycle — tool input/output guardrails, hooks, timeouts, concurrency limits, approvals, sessions, and RunState pause/resume all apply (see RunState and resume).[1] When ProgrammaticToolCallingTool() is present, the SDK applies a stricter replay-safety boundary and disables provider-managed retries and WebSocket pre-event retries, even before a program executes.[1]

Canonical ProgrammaticToolCallingTool usage with allowed_callers and a Pydantic output type:

@tool(allowed_callers=["programmatic"])
def get_inventory(sku: str) -> InventoryOutput:
    return InventoryOutput(sku=sku, available_units=42)

agent = Agent(
    name="Inventory planner",
    model="gpt-5.6",
    model_settings=ModelSettings(tool_choice="programmatic_tool_calling"),
    tools=[get_inventory, ProgrammaticToolCallingTool()],
)
result = Runner.run_sync(agent, "Check inventory for desk-lamp and summarize it.")

[1]

Sources

docs/tools.md

7.4Hosted tool search

Updated 2026-08-28

Hosted tool search lets OpenAI Responses models defer large tool surfaces until runtime, so the model loads only the subset it needs for the current turn, reducing tool-schema tokens without exposing every tool up front.[1] Hosted tool search is available only with OpenAI Responses models and requires openai>=2.25.0.[1]

Searchable deferred surfaces include @function_tool(defer_loading=True), tool_namespace(name=..., description=..., tools=[...]), and HostedMCPTool(tool_config={..., "defer_loading": True}).[1] Namespaces can mix immediate and deferred tools: tools without defer_loading=True remain callable immediately, while deferred tools in the same namespace are loaded through tool search.[1]

Deferred-loading function tools must be paired with exactly one ToolSearchTool() instance on the agent.[1] Named tool_choice cannot target bare namespace names or deferred-only tools; use auto, required, or a real top-level callable tool name instead.[1] ToolSearchTool(execution="client") is for manual Responses orchestration; if the model emits a client-executed tool_search_call, the standard Runner raises instead of executing it.[1] Tool search activity appears in RunResult.new_items and in RunItemStreamEvent with dedicated item and event types — see Result and items and Streaming for those type details.[1] ToolSearchTool() is the runtime mechanism the model calls to discover and load deferred tools; without it, the agent cannot resolve any deferred surface.

Canonical hosted tool search setup using tool_namespace and ToolSearchTool:

from agents import Agent, Runner, ToolSearchTool, tool_namespace
from agents.decorators import tool

@tool(defer_loading=True)
def get_customer_profile(customer_id: Annotated[str, "The customer ID to look up."]) -> str:
    """Fetch a CRM customer profile."""
    return f"profile for {customer_id}"

crm_tools = tool_namespace(
    name="crm",
    description="CRM tools for customer lookups.",
    tools=[get_customer_profile, list_open_orders],
)

agent = Agent(
    name="Operations assistant",
    model="gpt-5.6-sol",
    instructions="Load the crm namespace before using CRM tools.",
    tools=[*crm_tools, ToolSearchTool()],
)

[1]

Sources

docs/tools.md

8.1Runner API

Updated 2026-08-28

The Runner API in src/agents/run.py provides three execution methods—run() (async), run_sync() (sync), and run_streamed() (async streaming)—that invoke agents in a loop: calling the LLM, executing tools, handling handoffs, and returning a final result when no more actions are needed. The API supports custom error handlers, resuming from paused runs, optional WebSocket transport for the Responses API, and multi-turn session reuse with responses_websocket_session().

The public exports of src/agents/run.py include AgentRunner, Runner, RunConfig, RunOptions, RunState, RunContextWrapper, ModelInputData, CallModelData, CallModelInputFilter, OutputGuardrailBlockedMessageArgs, OutputGuardrailBlockedMessageFormatter, ToolNameCollisionPolicy, ReasoningItemIdPolicy, ToolExecutionConfig, ToolErrorFormatter, ToolErrorFormatterArgs, ToolNotFoundBehavior, DEFAULT_MAX_TURNS, set_default_agent_runner, and get_default_agent_runner.[1] The module-level DEFAULT_AGENT_RUNNER in src/agents/run.py is initialized to None at definition time and is set to an AgentRunner() instance by set_default_agent_runner.[1]

Runner exposes three execution methods: Runner.run() (async, returns RunResult), Runner.run_sync() (a sync wrapper over run()), and Runner.run_streamed() (async, returns RunResultStreaming, calls the LLM in streaming mode).[2]

Canonical agent execution: call await Runner.run(agent, input) which returns a RunResult; access the final answer via result.final_output.

async def main():
    result = await Runner.run(agent, "When did the Roman Empire fall?")
    print(result.final_output)

[3]

Runner.run in src/agents/run.py accepts input as a plain string (treated as a user message), a list of TResponseInputItem objects in OpenAI Responses API format, or a RunState object for resuming a paused or cancelled run.[1][2] Runner.run accepts error_handlers (a RunErrorHandlers instance keyed by error kind) to intercept and handle specific run errors without letting them propagate as exceptions.[1]

Set a custom default AgentRunner with set_default_agent_runner so that Runner.run, Runner.run_streamed, and Runner.run_sync all delegate to it

runner = mock.Mock(spec=AgentRunner)
set_default_agent_runner(runner)

agent = Agent(name="test", model=ScriptedModel())
await Runner.run(agent, input="test")
Runner.run_streamed(agent, input="test")
Runner.run_sync(agent, input="test")

[4]

The Runner agent loop: (1) calls the LLM; (2a) if output is final, returns the result; (2b) if a handoff is requested, updates the current agent and loops; (2c) if tool calls are produced, runs them, appends results, and loops; (3) raises MaxTurnsExceeded if max_turns is exceeded — pass max_turns=None to disable the turn limit.[2] Output is classified as "final" only when the LLM produces text output of the desired type and there are no tool calls.[2] During a handoff, the Runner updates its internal current-agent pointer and continues the agent loop with the new agent, so the caller receives a single unified result rather than separate per-agent results.

When the OpenAI Responses WebSocket transport is enabled via set_default_openai_responses_transport("websocket"), the normal Runner APIs continue to work unchanged; this transport uses the Responses API over WebSocket and is distinct from the Realtime API.[2] responses_websocket_session() is recommended when a shared WebSocket-capable provider and RunConfig are needed across multiple runs, including nested agent-as-tool calls that inherit the same run_config.[2] The WebSocket service processes one response at a time per connection and limits each connection to 60 minutes; responses_websocket_session() reuses the connection but does not remove those constraints.[2] After a WebSocket reconnect, store=False and ZDR flows cannot recover an uncached previous_response_id; start a new chain with full input context or rebuild from locally managed session state.[2] Exiting the responses_websocket_session() context while a WebSocket request is still in flight may force-close the shared connection; finish consuming streamed results before the context exits.[2] For long reasoning turns that hit WebSocket keepalive timeouts, increase ping_timeout or set ping_timeout=None to disable heartbeat timeouts; use HTTP/SSE transport when reliability matters more than WebSocket latency.[2]

Multi-turn WebSocket session reuse using responses_websocket_session() with previous_response_id:

async with responses_websocket_session(
    responses_websocket_options={"ping_interval": 20.0, "ping_timeout": 60.0},
) as ws:
    first = ws.run_streamed(agent, "Say hello in one short sentence.")
    async for _event in first.stream_events():
        pass

    second = ws.run_streamed(
        agent,
        "Now say goodbye.",
        previous_response_id=first.last_response_id,
    )
    async for _event in second.stream_events():
        pass

[2]

Sources

8.2RunConfig

Updated 2026-08-28

RunConfig is a dataclass in src/agents/run_config.py that defines run-wide settings: model selection, turn limits, input filtering, tool-naming policies, and tracing — applied uniformly across all agents in a single execution. Its hooks (call_model_input_filter, handoff_history_mapper, session_input_callback) let you customize how model inputs are prepared, history is compacted across agent handoffs, and new user messages merge into session state.

RunConfig is defined in src/agents/run_config.py and controls run-wide behavior — model selection, turn limits, tracing, tool policies, and input filtering — that applies across every agent invoked in a single run.[1] DEFAULT_MAX_TURNS is set to 10 in src/agents/run_config.py, making 10 AI invocations the default turn limit for a run.[2]

ModelInputData in src/agents/run_config.py is a dataclass holding input: list[TResponseInputItem] and instructions: str | None — the data container sent to the model each turn.[2] CallModelData in src/agents/run_config.py wraps ModelInputData, the active Agent, and the current context; it is passed to RunConfig.call_model_input_filter before every model call.[2] ReasoningItemIdPolicy in src/agents/run_config.py is a Literal["preserve", "omit"] type alias controlling whether reasoning item IDs are kept or stripped before model calls.[2] ToolNotFoundBehavior in src/agents/run_config.py is a Literal["raise_error", "return_error_to_model"] type alias controlling what happens when a model calls a tool that does not exist.[2] ToolNameCollisionPolicy in src/agents/run_config.py is a Literal["warn", "error"] type alias controlling whether duplicate tool names in a turn produce a warning or raise an error.[2]

RunConfig.call_model_input_filter is a hook to edit the fully prepared model input (instructions and input items) immediately before the model call — for example, to trim history or inject a system prompt.[1] RunConfig.reasoning_item_id_policy controls whether reasoning item IDs are preserved or omitted when the runner converts prior outputs into next-turn model input.[1] RunConfig.tool_not_found_behavior configures how the runner handles model-emitted function tool calls whose name does not match any available tool; the default raises ModelBehaviorError, but it can be set to return a model-visible error output instead.[1]

RunConfig.nest_handoff_history is an opt-in beta (default False) that compacts summarizable history into ordered assistant summary segments while preserving lossless message items; individual handoffs can override it via Handoff.nest_handoff_history.[1] RunConfig.handoff_history_mapper is an optional callable that receives the normalized transcript whenever nest_handoff_history is enabled and must return the exact list of input items to forward to the next agent, replacing the built-in ordered summary segments.[1] A handoff occurs when one agent delegates control to another agent within the same run; RunConfig's handoff history fields govern what context each receiving agent sees.

RunConfig.session_input_callback customizes how new user input is merged with session history before each Runner run when using Sessions; the callback can be sync or async, and the rewritten version of new-turn items is what gets persisted for that turn.[1][3]

RunConfig.workflow_name is recommended to be set on every run; RunConfig.trace_id sets the trace ID; RunConfig.group_id is an optional field that links traces across multiple runs.[1] RunConfig.trace_include_sensitive_data configures whether traces will include potentially sensitive data such as LLM and tool call inputs/outputs.[1]

Sources

8.3Run loop internals

Updated 2026-08-28

The run loop — the core of agent execution — is implemented as a delegation chain: src/agents/run.py orchestrates single turns, guardrails, and streaming by calling internal helpers from run_loop.py, session_persistence.py, and blocked_output.py, which in turn use internal data structures from run_steps.py to track step results and tool categories. ProcessedResponse in run_steps.py categorizes tool calls and approvals into separate lists (handoffs, functions, computer actions, shell calls, patches, MCP requests, interruptions, not-found tools, custom tools) and exports methods to query whether execution or user approval is needed.

src/agents/run.py delegates run-loop mechanics — single-turn execution, guardrail checks, tool execution, and streaming start — to helpers in src/agents/run_internal/run_loop.py via imports such as run_single_turn, run_input_guardrails, run_output_guardrails, and start_streaming.[1] run_loop.py is an internal orchestration module; all symbols it defines are explicitly marked as not part of the public SDK surface.[2] src/agents/run.py delegates session and memory persistence to helpers in src/agents/run_internal/session_persistence.py, including prepare_input_with_session, save_result_to_session, resumed_turn_items, and persist_session_items_for_guardrail_trip.[1] src/agents/run_internal/blocked_output.py manages output-guardrail-blocked scenarios — retaining items, sanitizing guardrail results, and replacing blocked content with a data-free placeholder — and is imported by src/agents/run.py.[1] src/agents/run_internal/run_steps.py defines the internal step and result data structures consumed by the run-loop orchestration; none of these types are part of the public SDK surface.[3]

ProcessedResponse in run_steps.py tracks separate lists for every category of tool call produced by a model turn: handoffs, functions, computer_actions, local_shell_calls, shell_calls, apply_patch_calls, mcp_approval_requests, interruptions, function_tools_not_found, and custom_tool_calls.[3] ProcessedResponse.has_tools_or_approvals_to_run() returns True when any of handoffs, functions, computer_actions, custom_tool_calls, local_shell_calls, shell_calls, apply_patch_calls, mcp_approval_requests, or function_tools_not_found are non-empty; hosted tools are excluded because they have already executed server-side.[3] ProcessedResponse.has_interruptions() returns True when self.interruptions is non-empty, indicating there are tool calls awaiting user approval (human-in-the-loop).[3] ToolRunFunctionNotFound in run_steps.py is a dataclass recording a tool call where the named function was not found; it appears in ProcessedResponse.function_tools_not_found and is counted by has_tools_or_approvals_to_run().[3]

SingleStepResult.generated_items in run_steps.py returns pre_step_items concatenated with session_step_items when set, falling back to new_step_items; session_step_items contains the full unfiltered item list for complete session observability.[3] SingleStepResult.processed_response in run_steps.py preserves the ProcessedResponse from the current step and is explicitly needed for resuming runs from interruptions.[3] SingleStepResult.nested_history_owned_items is None when the step did not rewrite handoff history; a non-None list signals that a handoff replaced history, requiring reconciliation of prior ownership against the new input.[3] NextStepInterruption in run_steps.py carries response_accepted (whether the server already accepted the response whose local processing is incomplete) and llm_end_hooks_started (whether response-end hooks fired before the interruption was persisted).[3]

QueueCompleteSentinel in run_steps.py is a sentinel class used to signal that streaming of run-loop results is finished; a singleton instance QUEUE_COMPLETE_SENTINEL is provided.[3] NOT_FINAL_OUTPUT in run_steps.py is a pre-built ToolsToFinalOutputResult(is_final_output=False, final_output=None) constant used to signal that tool execution did not produce a terminal output.[3] run_loop.py re-exports OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT from blocked_output under the alias _OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT.[2]

_prepare_turn_input_items in run_loop.py converts the caller's raw input to a new input list, converts generated RunItems to input items (respecting the reasoning_item_id_policy), then merges the two via prepare_model_input_items.[2] _should_persist_stream_items in run_loop.py skips session persistence when no session is configured or a server-managed conversation tracker is present, and also skips when an input guardrail tripwire was triggered during streaming.[2] cleanup_models_after_run in run_loop.py iterates every model resolved during a run and calls model._cleanup_on_run_end(tool_use_tracker) on each; failures are logged as warnings rather than raised.[2] _ensure_stream_event_item_occurrence_key in run_loop.py lazily assigns a uuid4().hex key to a RunItem if one is not already present, ensuring each streamed item has a stable occurrence identity.[2] _stream_event_item_occurrence_key in run_loop.py retrieves a per-RunItem UUID stored under the attribute _agents_stream_event_item_occurrence_key, returning None if absent or not a non-empty string.[2]

run_loop._retained_items_for_blocked_output is called to rebuild the allowed subset of tool call/output items when an output guardrail blocks a function batch (exercised in tests/test_agent_runner.py).[4] run_loop._retained_items_for_blocked_output guards against hash-collision and equality-impostor attacks on dict-key discriminators: it does not invoke __eq__ on non-string keys, and if EqualityImpostor.__eq__ returns True for discriminator fields the entire batch is discarded rather than accepted.[4] When the caller is a valid CallerDirect, the caller field is preserved in the retained ToolCallItem's raw_item as {"type": "direct"}.[4]

tests/test_agent_runner.py imports and exercises a wide range of internal helpers — including run_loop._retained_items_for_blocked_output, drop_orphan_function_calls, normalize_input_items_for_api, persist_session_items_for_guardrail_trip, save_result_to_session, and execute_approved_tools — confirming these are testable internal contracts.[4]

Sources

8.4Tool execution pipeline

Updated 2026-08-28

The tool execution pipeline in tool_execution.py manages concurrent function-tool invocations with failure arbitration, cancellation propagation, and phase-aware task state tracking across batches of tool calls. Task failures are ranked by exception severity and submission order, enabling the pipeline to surface the most significant early failure when multiple tools fail concurrently.

tool_execution.py is the execution-time module for the run pipeline, hosting tool execution helpers, approval plumbing, and payload coercion; action classes are defined separately in tool_actions.py.[1] Exported functions from tool_execution.py include execute_function_tool_calls, execute_custom_tool_calls, execute_local_shell_calls, execute_shell_calls, execute_apply_patch_calls, execute_computer_actions, execute_approved_tools, and approval/resolution helpers.[1]

_FunctionToolFailure is a frozen dataclass that pairs a BaseException with an order integer and a source literal ("direct", "cancelled_teardown", or "post_invoke") for arbitrating which failure wins when multiple function-tool tasks fail concurrently.[1] _get_function_tool_failure_priority assigns priority 0 to asyncio.CancelledError, 1 to Exception, and 2 to any other BaseException (e.g., SystemExit), so fatal exceptions always win arbitration.[1] _select_function_tool_failure arbitrates concurrent function-tool failures by priority first, then by the tool call's order (lower order wins ties), ensuring the most significant earlier-submitted failure surfaces.[1]

_FunctionToolTaskState tracks per-task mutable execution state including the ToolRunFunction, an order index, the live asyncio.Task, and a boolean in_post_invoke_phase flag used to distinguish execution phases during failure arbitration.[1] When a failure is detected, tool_execution.py calls _cancel_function_tool_tasks to cancel all sibling tasks in a batch, enabling structured cooperative cancellation across concurrently executing function tools.[1] The constant _FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS = 0.25 defines the time budget given to sibling tasks to drain after cancellation propagation.[1] The constant _FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS = 0.1 defines the wait window given after a function tool's primary invocation completes before proceeding to the next pipeline phase.[1]

_consume_function_tool_task_result reports background task exceptions to the event loop's exception handler via call_exception_handler; cancelled tasks and tasks whose message_for_exception returns None are silently dropped.[1] Background cleanup tasks that raise asyncio.CancelledError are silently ignored; those raising an Exception emit a warning message; those raising a fatal BaseException emit a fatal-level message — controlled by _background_cleanup_task_exception_message.[1] Detached parent-cancelled tasks that raise an Exception are silently dropped (return None); those raising a fatal BaseException emit a fatal-level message — controlled by _parent_cancelled_task_exception_message.[1]

_ToolOutputGuardrailExecutionResult wraps a tool output together with an is_rejection flag, distinguishing a genuine tool result from output synthesized by a rejecting guardrail.[1]

Sources

src/agents/run_internal/tool_execution.py

8.5Streaming

Updated 2026-08-28

Streaming an agent run via Runner.run_streamed() yields a RunResultStreaming object whose stream_events() async iterator emits StreamEvent objects — a union of raw LLM tokens, high-level run items (messages, tool calls, handoffs), and agent updates. The streaming surface supports pausing for tool approval, cancellation with graceful turn cleanup, and resuming from intermediate states via RunState conversion, making it possible to weave user input and approvals into a live agent execution. A RunState is a serializable snapshot of an in-progress agent run, capturing conversation history, pending tool results, and staged inputs; it allows the run to be persisted, transmitted, or resumed across process boundaries or after an interruption.

To stream an agent run, call Runner.run_streamed(), which returns a RunResultStreaming object; calling result.stream_events() on it yields an async stream of StreamEvent objects.[1] The StreamEvent type alias in src/agents/stream_events.py is the union RawResponsesStreamEvent | RunItemStreamEvent | AgentUpdatedStreamEvent, representing all possible streaming events from an agent.[2]

RawResponsesStreamEvent objects wrap raw LLM events from the OpenAI Responses API; each object's data field holds an event with a type such as response.created or response.output_text.delta, making them suitable for token-by-token streaming to the user.[1] RunItemStreamEvent provides higher-level streaming events that fire when a complete item is generated — a full message, tool result, or similar — in contrast to RawResponsesStreamEvent which fires per-token. AgentUpdatedStreamEvent fires when the current agent changes due to a handoff.[1] RunItemStreamEvent in src/agents/stream_events.py wraps a RunItem and uses the type discriminator "run_item_stream_event". Its name field uses a fixed set of literals: message_output_created, handoff_requested, handoff_occured, tool_called, tool_search_called, tool_search_output_created, tool_output, reasoning_item_created, mcp_approval_requested, mcp_approval_response, and mcp_list_tools. The spelling handoff_occured is intentionally preserved for backward compatibility and cannot be changed without a breaking change.[2][1] A handoff call is emitted only as handoff_requested, not additionally as tool_called; ordinary function tool calls in the same turn still emit tool_called.[1] When hosted tool search is used, tool_search_called is emitted when the model issues a tool-search request and tool_search_output_created is emitted when the Responses API returns the loaded subset.[1]

With Programmatic Tool Calling, tool_called is emitted for the generated program and for ordinary program-owned child tool calls; tool_output is emitted for child tool outputs and the program_output. Program-owned hosted MCP mcp_approval_request and mcp_list_tools items are exceptions: they emit as mcp_approval_requested and mcp_list_tools, wrapping MCPApprovalRequestItem and MCPListToolsItem respectively.[1] Program-owned child tool calls carry a caller field whose type is program and whose caller_id identifies the parent program; inspecting the raw item's type is needed to distinguish remaining item kinds.[1] Computer-tool raw streaming events differ between preview (computer_call with a single action) and gpt-5.5+ GA (computer_call with batched actions[]). The higher-level RunItemStreamEvent surface does not distinguish these shapes — both surface as tool_called, with screenshot results as tool_output wrapping a computer_call_output item.[1]

A streaming run is not complete until the result.stream_events() async iterator finishes; post-processing such as session persistence, approval bookkeeping, and history compaction may continue after the last visible token. When the loop exits, result.is_complete reflects the final run state.[1] To stop a streaming run immediately, call result.cancel(); to let the current turn finish cleanly before stopping, call result.cancel(mode="after_turn").[1] If cancel(mode="after_turn") stops a run after a tool turn and the caller is manually continuing from result.to_input_list(mode="normalized"), they should rerun result.last_agent with that normalized input to continue the unfinished turn rather than appending a fresh user turn.[1] When new user input arrives before a cancelled run resumes, the caller should convert the drained result with result.to_state(), call state.add_input(...), and resume from the state; the runner admits the staged input immediately before the next model call.[1]

When a streaming run pauses for tool approval, result.stream_events() finishes and pending approvals are exposed in RunResultStreaming.interruptions. The caller should convert the result to a RunState via result.to_state(), approve or reject each interruption, and resume with Runner.run_streamed(...).[1] A streamed run that stopped for tool approval must not be treated as a new turn; the caller should finish draining the stream, inspect result.interruptions, and resume from result.to_state() — not start a fresh user turn.[1]

Example: streaming token-by-token output from an agent by filtering for raw_response_event events whose data is a ResponseTextDeltaEvent.[1] Example: consuming RunItemStreamEvent to print agent updates, tool calls, tool outputs, and message outputs while ignoring raw token events.[1] Example: handling streaming with tool approval interruptions — drain the stream, check result.interruptions, convert to RunState, approve, and re-run.[1]

Sources

8.6Run context

Updated 2026-08-28

RunContextWrapper in src/agents/run_context.py wraps user-supplied context and accumulates token usage, approval state, and tool invocation metadata across an agent run—it never reaches the LLM itself, serving only to thread dependencies and data to tools and callbacks. Run contexts support resumable checkpoints by deep-copying usage and approval state, while sharing approval decisions across related runs via _share_tool_state_with() to maintain consistent tool authorization.

RunContextWrapper in src/agents/run_context.py wraps the context object passed to Runner.run() and accumulates per-run usage, approval state, and tool invocation tracking. Contexts are not passed to the LLM — they exist solely to carry dependencies and data to tool functions, callbacks, and hooks.[1] TContext in src/agents/run_context.py is a TypeVar with a default of Any, making RunContextWrapper generic over the user-supplied context type.[1]

RunContextWrapper.usage (a Usage instance) accumulates token usage over the lifetime of an agent run. For streamed responses, its value is stale until the last chunk of the stream is processed.[1] RunContextWrapper.tool_input holds structured input for the current agent tool run when available, and is None otherwise.[1]

_ApprovalRecord in src/agents/run_context.py tracks per-tool approval and rejection state. approved and rejected are either booleans (permanent allow/deny) or lists of call IDs when approval is scoped to specific tool calls.[1] _ToolInvocationRecord in src/agents/run_context.py tracks the canonical identity and lifecycle of a single provider tool call ID, including its invocation_type, approval_scope, fingerprint, and boolean flags executed and completed.[1] RunContextWrapper._resolve_approval_key() derives a single canonical approval key for a ToolApprovalItem, preferring the key from get_function_tool_approval_keys and falling back to tool_qualified_name or the bare tool name.[1] RunContextWrapper._resolve_call_id() resolves the provider call ID from a ToolApprovalItem. For hosted MCP approval requests it uses the request_id; for mcp_approval_request items it checks provider_data.id; for regular function calls it reads call_id from the raw item.[1]

RunContextWrapper._copy_for_run_state() in src/agents/run_context.py deep-copies usage, _approvals, and _tool_invocations so that resumable checkpoints do not bleed token counts or approval state into each other.[1] RunContextWrapper._copy_for_run_state() also calls set_agent_tool_state_scope with a fresh UUID hex to give the copied context an independent tool-state scope.[1] RunContextWrapper._share_tool_state_with() wires _approvals, _tool_invocations, _allow_legacy_approval_binding_reconstruction, and _restored_unbound_approval_call_ids from the source context directly into the target, so both wrappers share the same mutable approval objects.[1]

Runtime annotations for ToolApprovalItem and TResponseInputItem in src/agents/run_context.py are resolved to Any at runtime (instead of importing items.py) to avoid circular imports. The TYPE_CHECKING guard keeps them typed for static analysis only.[1]

Sources

src/agents/run_context.py

9.1Model interface

Updated 2026-08-28

The Model and ModelProvider abstract base classes define how agents fetch responses and manage resources: Model.get_response() is the core async method that accepts system instructions, tools, output schemas, and conversation state, while ModelProvider locates models by name and both define optional hooks for cleanup and retry guidance. ModelTracing controls whether and what response data the SDK records: DISABLED suppresses tracing entirely, ENABLED includes full payloads, and ENABLED_WITHOUT_DATA records structure only—so callers use include_data() to conditionally serialize inputs and outputs.

The Model abstract base class in src/agents/models/interface.py defines the contract every model implementation must fulfill, including how responses are fetched, resources released, and retry guidance surfaced.[1] ModelProvider in src/agents/models/interface.py is an abstract base class responsible for looking up Model instances by name via its abstract get_model(model_name) method.[1]

Model.get_response() is an abstract async method that accepts system_instructions, input (string or list of response input items), model_settings, tools, output_schema, handoffs, tracing, previous_response_id, conversation_id, and prompt, and returns a ModelResponse.[1]

Model implementations in src/agents/models/interface.py must assign a non-empty, unique call ID to every tool invocation; a call ID must not be reused for a changed tool identity or payload, and an exact completed replay may be omitted by the runtime without re-executing the invocation.[1] A missing or reused tool call ID in a Model implementation breaks runtime deduplication: the SDK cannot detect replayed invocations, allowing the same tool to execute multiple times with unintended side effects.

Model.close() in src/agents/models/interface.py is a no-op by default; models that maintain persistent connections should override it to release those resources.[1] ModelProvider.aclose() in src/agents/models/interface.py is a no-op by default; providers that cache persistent models or network connections should override it.[1]

Model.get_retry_advice() in src/agents/models/interface.py returns None by default; model implementations may override it to provide provider-specific hints such as replay safety, retry-after delays, or explicit server retry guidance.[1]

ModelTracing is an enum in src/agents/models/interface.py with three values: DISABLED (tracing off entirely), ENABLED (tracing on, all data included), and ENABLED_WITHOUT_DATA (tracing on, but inputs/outputs excluded).[1] ModelTracing.include_data() returns True only when tracing is ENABLED (not ENABLED_WITHOUT_DATA), so callers use it to gate serialization of prompt and response payloads.[1]

Sources

src/agents/models/interface.py

9.2OpenAIProvider

Updated 2026-08-28

OpenAIProvider is the concrete model provider for OpenAI APIs; it manages API credentials, websocket configuration, HTTP client pooling, and feature validation for Chat Completions and Responses models. OpenAIProvider caches websocket model instances per event loop to maintain persistent connections when the same provider instance handles multiple concurrent calls, and auto-prunes closed event loops to release resources.

OpenAIProvider, implemented in src/agents/models/openai_provider.py, is the concrete model provider for OpenAI APIs and resolves base_url from the OPENAI_BASE_URL environment variable and websocket_base_url from OPENAI_WEBSOCKET_BASE_URL when those options are not explicitly supplied at construction time.[1] DEFAULT_MODEL in src/agents/models/openai_provider.py is kept as "gpt-4o" only for backward compatibility; the recommended approach is to call get_default_model(), which reflects the current default.[1]

OpenAIProvider raises a UserError if openai_client is provided together with any of api_key, base_url, websocket_base_url, organization, or project.[1] OpenAIProvider accepts an agent_registration parameter (type OpenAIAgentRegistrationConfig | dict | None), resolves it via resolve_openai_agent_registration_config, and exposes the result through the agent_registration property.[1]

OpenAIProvider shares a single httpx2.AsyncClient across all requests via shared_http_client() to avoid per-request connection-pool teardown and the associated latency and resource cost.[1] Websocket model wrappers are cached per event loop using a WeakKeyDictionary keyed by asyncio.AbstractEventLoop, so that websocket transport can maintain a persistent connection when callers pass model names as strings through a shared provider instance.[1] OpenAIProvider._prune_closed_ws_loop_caches() drops websocket model cache entries for event loops that are already closed, and forcibly drops the underlying websocket connection synchronously for each cached OpenAIResponsesWSModel.[1]

OpenAIProvider accepts a strict_feature_validation flag (default False); when True, Chat Completions models raise a UserError if callers pass Responses-only features such as previous_response_id, conversation_id, prompt, or non-text-only tool outputs.[1] OpenAIProvider accepts a buffer_streamed_tool_calls flag (default False); when True, Chat Completions models buffer all streamed function tool-call deltas and emit them only after the provider stream finishes, for compatibility with providers whose streamed tool-call chunk semantics are unreliable.[1]

Sources

src/agents/models/openai_provider.py

9.3OpenAI Responses model

Updated 2026-08-28

OpenAI Responses model, implemented in openai_responses.py, handles request headers, tool parameter validation, namespace grouping, and response streaming for the OpenAI Agents SDK's async Responses API integration. _ResponseStreamWithRequestId wraps WebSocket event streams to attach request IDs and track terminal events, while supporting fallback usage computation and configurable message-size limits for memory-constrained deployments.

openai_responses.py sets a User-Agent header of the form Agents/Python <version> on all Responses API requests via the _HEADERS constant.[1] Per-async-task header injection is supported through _HEADERS_OVERRIDE, a ContextVar[dict[str, str] | None] in openai_responses.py that defaults to None and allows per-context-variable overrides of request headers on Responses API calls.[1]

_require_responses_tool_param in openai_responses.py validates that a tool param payload is a Mapping with a string type key, raising TypeError with a descriptive message if either check fails.[1] _NamespaceToolParam is an internal TypedDict in openai_responses.py representing a namespace-grouped tool parameter with type="namespace", a name, a description, and a list of FunctionToolParam entries.[1] Namespace grouping of function tools is resolved by the _tool_identity module — openai_responses.py imports get_explicit_function_tool_namespace and get_function_tool_namespace_description from .._tool_identity before assembling entries into _NamespaceToolParam.[1] _coerce_response_includables in openai_responses.py deliberately accepts arbitrary strings for ModelSettings.response_include so callers can pass through new server-supported flags before the local SDK updates its enum union.[1]

_ResponseStreamWithRequestId in openai_responses.py wraps an async SDK event stream, retaining the originating request ID and back-propagating it onto each response object in every yielded event via _attach_request_id.[1] _ResponseStreamWithRequestId recognizes four terminal event types — response.completed, response.failed, response.incomplete, and response.error — and sets an internal _yielded_terminal_event flag when one is encountered.[1] For response.completed events whose response carries no usage data, _ResponseStreamWithRequestId.__anext__ calls _mark_transport_request_without_usage to ensure the request counter is preserved.[1] _construct_response_stream_event_from_payload in openai_responses.py parses WebSocket event payloads using the OpenAI SDK's internal construct_type function; if that internal is unavailable, it raises RuntimeError advising an SDK upgrade or a switch back to HTTP transport.[1] OpenAIResponsesWebSocketOptions exposes a max_size field controlling the maximum byte size of an incoming WebSocket message; setting it to None disables the limit, while an explicit value bounds memory usage for long-lived agent processes in memory-constrained containers.[1]

_usage_from_response in openai_responses.py falls back to Usage(requests=_requests_for_response_without_usage(response)) when response.usage is None, preserving the request count even for responses that omit usage data.[1] _json_dumps_default in openai_responses.py handles custom JSON serialization for Pydantic models (via model_dump(mode='json', exclude_none=True)), dataclasses (via asdict), and Enum values; it raises TypeError for all other unrecognized types.[1]

Sources

src/agents/models/openai_responses.py

9.4OpenAI Chat Completions model

Updated 2026-08-28

OpenAIChatCompletionsModel wraps OpenAI's AsyncOpenAI client to implement the Model interface, supporting chat completions with configurable feature validation that either rejects or silently ignores unsupported fields like reusable prompts and certain reasoning modes. The model logs responses at DEBUG level (when enabled), wraps calls in tracing spans with error capture, and uses shielded background cleanup to safely close streams without abandoning in-progress close operations.

OpenAIChatCompletionsModel in src/agents/models/openai_chatcompletions.py implements the Model interface and wraps an AsyncOpenAI client to provide chat-completions-based model responses.[1]

OpenAIChatCompletionsModel accepts a strict_feature_validation flag at construction time: when True, unsupported features raise UserError; when False (the default), they are warned once and silently ignored.[1] The prompt (reusable prompt) parameter is not supported by OpenAIChatCompletionsModel; reusable prompts require the Responses API — see OpenAI Responses model.[1] Of the ModelSettings.reasoning fields, OpenAIChatCompletionsModel supports only reasoning.effort; the reasoning.mode and reasoning.context fields are unsupported and are either rejected with UserError (strict mode) or silently dropped.[1] For official OpenAI clients, OpenAIChatCompletionsModel validates that user-message content parts use only the supported types (input_text, input_image, input_audio, input_file), raising UserError on any other type.[1]

OpenAIChatCompletionsModel.get_response raises ModelBehaviorError when the provider returns a ChatCompletion with no choices, and includes the provider error payload in the error message if one is present.[1] When _debug.DONT_LOG_MODEL_DATA is falsy, OpenAIChatCompletionsModel.get_response logs the full model response message as pretty-printed JSON at DEBUG level; when truthy, it logs only a redacted "Received model response" string.[1] OpenAIChatCompletionsModel.get_response wraps every call in a generation_span tracing context and a model_span_errors error-capture context from src/agents/tracing/.[1]

OpenAIChatCompletionsModel._close_stream_allowing_background_completion shields the provider stream's aclose() call so that cancellation during close does not abandon a half-finished close operation; the task is detached to finish in the background, avoiding a second close (which is not guaranteed to be safe or idempotent).[1] OpenAIChatCompletionsModel.get_retry_advice delegates to get_openai_retry_advice in src/agents/models/_openai_retry.py to determine whether and how to retry a failed request.[1]

Sources

src/agents/models/openai_chatcompletions.py

9.5Model settings

Updated 2026-08-28

ModelSettings is a Pydantic dataclass in the SDK holding all optional LLM call parameters—temperature, tool choice, timeouts, retry policies, and provider-specific options like prompt caching—though not all models or providers support every field. A subset of these settings (including temperature, timeouts, retry policies, and caching options) are traced via _TRACEABLE_MODEL_SETTING_FIELDS, while resolve() lets you overlay runtime overrides on base settings and merge nested retry and extra_args policies.

ModelSettings in src/agents/model_settings.py is a Pydantic dataclass holding all optional LLM call parameters; not all models or providers support every field.[1] The _TRACEABLE_MODEL_SETTING_FIELDS tuple in src/agents/model_settings.py lists the subset of ModelSettings fields recorded in traces, including temperature, top_p, frequency_penalty, presence_penalty, tool_choice, parallel_tool_calls, truncation, max_tokens, reasoning, verbosity, metadata, store, prompt_cache_retention, include_usage, response_include, top_logprobs, retry, context_management, prompt_cache_options, and timeout.[1]

ModelSettings.parallel_tool_calls defaults to None, deferring to the provider default (typically enabled for most OpenAI models). Set to False to restrict the model to at most one tool call per turn.[1] ModelSettings.store defaults to None; for the Responses API it is automatically enabled when unset, while for the Chat Completions API it is enabled for OpenAI and omitted for other providers so their own default applies.[1] ModelSettings.preserve_raw_usage enables capturing a JSON-compatible snapshot of the raw provider usage payload before SDK normalization, stored as ModelResponse.raw_usage. This setting does not request usage from the provider; use include_usage separately when a streaming provider requires it.[1] ModelSettings.timeout is enforced via asyncio cancellation and bounds the complete model attempt (including transport waits), but does not cover the full run, tool calls, or retry backoff periods.[1] ModelSettings.context_management accepts a list of ContextManagement entries for OpenAI Responses API requests; for example, [{"type": "compaction", "compact_threshold": 200000}] enables server-side compaction when the context token count crosses the threshold.[1] ModelSettings.prompt_cache_options configures OpenAI prompt caching; use {"mode": "explicit", "ttl": "30m"} with content-part cache breakpoints to control which prompt prefixes are eligible for caching.[1] ModelSettings.top_logprobs automatically adds "message.output_text.logprobs" to the response include list when set.[1] ModelSettings.extra_args passes arbitrary keyword arguments directly to the underlying model provider's API call; not all models support all parameters.[1]

ToolChoice in src/agents/model_settings.py is a type alias supporting "auto", "required", "none", an arbitrary string, an MCPToolChoice dataclass, or None.[1] MCPToolChoice in src/agents/model_settings.py is a plain dataclass with server_label and name fields used to target a specific MCP tool by server identity and tool name.[1]

The Omit type in src/agents/model_settings.py wraps OpenAI's _Omit sentinel and is Pydantic-validated: None is coerced to _Omit() in both JSON and Python modes, and it serializes back to None.[1]

ModelSettings.resolve() produces a new ModelSettings by overlaying non-None values from an override — which may be a ModelSettings instance or a plain dict — on top of the current instance.[1] When both the base and override have extra_args, ModelSettings.resolve() merges those dictionaries rather than replacing them, combining keys from both sides.[1] When both the base and override contain retry, ModelSettings.resolve() merges the two ModelRetrySettings objects field-by-field rather than replacing wholesale, preserving base retry policy components under a partial override.

Compose retry policies using retry_policies.any() to combine provider-suggested, retry-after, network-error, and HTTP-status policies

apply_policies = retry_policies.any(
    retry_policies.provider_suggested(),
    retry_policies.retry_after(),
    retry_policies.network_error(),
    retry_policies.http_status([408, 409, 429, 500, 502, 503, 504]),
)

[2]

Configure ModelRetrySettings with composite retry policies (provider-suggested, retry-after, network error, HTTP status) and attach them via RunConfig and Agent.model_settings

retry = ModelRetrySettings(
    max_retries=4,
    backoff={
        "initial_delay": 0.5,
        "max_delay": 5.0,
        "multiplier": 2.0,
        "jitter": True,
    },
    policy=policy,
)
run_config = RunConfig(model_settings=ModelSettings(retry=retry))
agent = Agent(
    name="Assistant",
    instructions="You are a concise assistant. Answer in 3 short bullet points at most.",
    model_settings=ModelSettings(retry=retry),
)
result = await Runner.run(
    agent,
    "Explain exponential backoff for API retries in plain English.",
    run_config=run_config,
)

[2]

Sources

10.1MCP server internals

Updated 2026-08-28

The MCP server internals in the OpenAI Agents SDK handle tool discovery, schema caching, approval policies, and safe credential handling across transport and exception layers. Tool snapshots are deep-copied to prevent cache mutation; approval policies flexibly support literals, per-tool mappings, and async callbacks; error chains and logs are scrubbed of credentials to prevent leakage.

AgentBase.get_mcp_tools calls MCPUtil.get_all_function_tools, passing the agent's MCP servers, schema conversion settings, run context, and the agent itself as arguments.[1]

In src/agents/mcp/server.py, _snapshot_tools() returns deep-copied MCPTool objects — via tool.model_copy(deep=True) — so callers cannot mutate cached tool schemas.[2]

RequireApprovalSetting in src/agents/mcp/server.py accepts a policy literal "always"/"never", a structured RequireApprovalObject, a per-tool dict mapping tool names to policies, a LocalMCPApprovalCallable, a plain bool, or None.[2] LocalMCPApprovalCallable in src/agents/mcp/server.py is a Callable[[RunContextWrapper[Any], AgentBase, MCPTool], MaybeAwaitable[bool]] — it can be synchronous or asynchronous.[2]

_safe_transport_cause() in src/agents/mcp/server.py returns None — suppressing the error cause chain — for any HTTPX transport error whose URLs are not credential-safe, or that already carries a __cause__, __context__, or __notes__, preventing credential leakage via exception chaining.[2] _credential_safe_exception_group() in src/agents/mcp/server.py recursively replaces a BaseExceptionGroup with fixed-data nodes, substituting Exception leaves with RuntimeError(_SAFE_EXCEPTION_MESSAGE), so control-flow semantics are preserved while all credential-bearing error details are scrubbed.[2] _log_transport_warning() in src/agents/mcp/server.py suppresses the transport exception object from the log entirely when the URL is not credential-safe, emitting only the bare message string to avoid accidental credential exposure in logs.[2] When _debug.DONT_LOG_TOOL_DATA is set in src/agents/mcp/server.py, transport warnings are passed directly to log_tool_action_warning with the full error object, bypassing the credential-safety check.[2]

src/agents/mcp/server.py loads StreamableHTTPTransport and streamablehttp_client dynamically from mcp.client.streamable_http at import time, supporting both MCP v1 and v2 without hard import failures.[2]

Sources

10.2MCPServerManager

Updated 2026-08-28

MCPServerManager is an async context manager in the OpenAI Agents SDK that orchestrates a pool of MCP (Model Context Protocol) servers, handling concurrent connection setup and per-server task affinity during cleanup. It accepts timeout, failure-handling, and parallelism options: failures can be strict (halt immediately) or lenient (record and continue), and connections run serially or in parallel while preserves cleanup ordering per server. MCP (Model Context Protocol) is a standard interface through which an AI agent communicates with external tool servers by invoking callable tools each server exposes.

MCPServerManager in src/agents/mcp/manager.py is an async context manager that calls connect_all() on __aenter__ and cleanup_all() on __aexit__.[1] During construction, MCPServerManager deduplicates the server list so the same server object cannot appear twice in all_servers.[1]

MCPServerManager defaults to connect_timeout_seconds=10.0 and cleanup_timeout_seconds=10.0; both accept a positive finite number of seconds or None to disable the timeout.[1] A value of zero is rejected because it would create an immediate deadline — only positive finite numbers or None are accepted.[1] Timeout validation runs on both construction and property assignment, so assigning an invalid value after construction also raises.[1]

When drop_failed_servers=True (the default), MCPServerManager.active_servers excludes servers that failed to connect; when False, failed servers remain in active_servers.[1] When strict=True, MCPServerManager raises on the first connection failure; when False (the default), failures are recorded in failed_servers/errors and the run proceeds with the remaining servers.[1] Setting connect_in_parallel=True spawns a dedicated _ServerWorker task per server so connects run concurrently while preserving the task affinity required for cleanup.[1]

MCPServerManager.active_servers and related properties return snapshots (new lists/dicts), so callers do not hold live references to internal state.[1] _ServerWorker in src/agents/mcp/manager.py serializes all connect and cleanup commands through an asyncio.Queue, ensuring per-server operations are ordered and run in the same task.[1] _ServerWorker.cleanup in src/agents/mcp/manager.py is idempotent: a second call reuses the existing _cleanup_future rather than enqueuing another cleanup command.[1] _run_with_timeout_in_task in src/agents/mcp/manager.py uses asyncio.timeout (Python ≥ 3.11) when available, and falls back to a loop.call_later cancel handle on older Python to preserve task affinity for MCP server cleanup.[1]

Canonical usage of MCPServerManager in src/agents/mcp/manager.py as an async context manager, passing active_servers to an Agent:

async with MCPServerManager([server_a, server_b]) as manager:
    agent = Agent(
        name="Assistant",
        instructions="...",
        mcp_servers=manager.active_servers,
    )

[1]

MCPServerManager in src/agents/mcp/manager.py can be used in a FastAPI lifespan to share managed MCP servers across requests:

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with MCPServerManager([server_a, server_b]) as manager:
        app.state.mcp_manager = manager
        yield

app = FastAPI(lifespan=lifespan)

[1]

Sources

src/agents/mcp/manager.py

Search

1Introduction

2Getting started

2.1Orientation

2.2Installation and requirements

2.3Quickstart and how-to

2.4Examples and developer workflow

3Agent behavior

3.1Handoffs

3.2Guardrails

3.3Lifecycle hooks

3.4Result and items

3.5Exceptions and redaction

4Run lifecycle

4.1RunState and resume

4.2Sessions and memory

4.3Sandbox

5Observability and testing

5.1Tracing

5.2Testing

5.3Upgrading

6Agents

6.1Agent types

7Tools

7.1Tool catalog

7.2Function schema

7.3Programmatic tool calling

7.4Hosted tool search

8Runner and run loop

8.1Runner API

8.2RunConfig

8.3Run loop internals

8.4Tool execution pipeline

8.5Streaming

8.6Run context

9Models

9.1Model interface

9.2OpenAIProvider

9.3OpenAI Responses model

9.4OpenAI Chat Completions model

9.5Model settings

10MCP

10.1MCP server internals

10.2MCPServerManager