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)
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")
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
Sources