Overview
Pydantic AI’s realtime support lets an agent hold a live, spoken conversation. It streams the user’s audio to a speech-to-speech model and streams the model’s spoken reply back over one persistent connection, so latency is low and interruptions feel natural.
A realtime session uses the same agent tools, dependencies,
instructions, message history,
capabilities, usage limits, and
observability as the rest of Pydantic AI, and that’s the point: mid-call the
agent can look up an order, check availability, or act on the logged-in user’s data with the same
tools and dependencies a text agent would use. The call itself becomes ordinary message history
that you can hand to Agent.run() for summarization or
structured follow-up, the same code runs against four providers, and usage
limits and Logfire tracing are built in. Your application owns the audio transport —
bridged through your backend, or browser-direct over WebRTC
on OpenAI and Azure — while Pydantic AI runs the provider-agnostic agent loop.
Install Pydantic AI with the OpenAI realtime dependencies, and set OPENAI_API_KEY:
pip install "pydantic-ai-slim[openai-realtime]"
uv add "pydantic-ai-slim[openai-realtime]"
A complete voice agent is one agent, one session, and three small loops — microphone in, speaker out, and a transcript log. The model hears the user, calls your tool on your backend, and answers out loud:
import asyncio
import contextlib
from collections.abc import AsyncIterator
from pydantic_ai import Agent
from pydantic_ai.realtime import RealtimeSession
agent = Agent(instructions='You take reservations for The Terrace. Keep replies short.')
@agent.tool_plain
async def check_availability(day: str, party_size: int) -> str:
"""Check whether a table is free."""
return f'One table for {party_size} is free at 7 pm {day}.'
async def stream_microphone(session: RealtimeSession) -> None:
... # capture signed 16-bit mono PCM chunks and `await session.send_audio(chunk)`
async def play_audio(chunks: AsyncIterator[bytes]) -> None:
async for chunk in chunks:
... # write the PCM chunk to your speaker
async def main():
async with agent.realtime('openai:gpt-realtime').session() as session:
microphone = asyncio.create_task(stream_microphone(session))
speaker = asyncio.create_task(play_audio(session.stream_audio()))
async for part in session.stream_transcripts():
print(f'{part.speaker}: {part.transcript}')
#> user: Hi! Do you have a table for two tomorrow night?
#> assistant: We do: 7 pm, table for two. Want me to book it?
if part.speaker == 'assistant':
break # keep listening in a real call; we stop after one exchange
# Leaving the `async with` block closes the session, which ends the speaker's audio stream —
# but the microphone reads an external source, so stop it explicitly.
microphone.cancel()
with contextlib.suppress(asyncio.CancelledError):
await microphone
await speaker
if __name__ == '__main__':
asyncio.run(main())
(This example is complete, it can be run “as is” — after filling in the two audio placeholders, which depend on your audio stack)
Capture and play at the sample rates the model expects — they’re reported by the model’s profile
and can differ between input and output (see Provider support below). The
voice assistant example fills the placeholders in with
sounddevice for a runnable microphone-and-speaker loop; the
text-to-audio example skips audio input entirely by
sending a text prompt and saving the spoken reply to a WAV file.
Your backend opens the provider connection and runs a
RealtimeSession. Stream content in with
send() or
send_audio(), and iterate the session for its
event stream — content, tool, turn, error, and reconnect events — or consume the
dedicated stream_audio() and
stream_transcripts() views as the
quickstart does.
device ↔ media bridge ↔ RealtimeSession ↔ provider
├── typed tools
└── message history
(your backend)
The media bridge is whatever moves audio between the user’s device and your backend — a browser WebSocket or a telephony bridge. It’s how you deploy this beyond a local microphone; see Connecting a frontend for each shape. On OpenAI and Azure the browser can instead exchange media with the provider directly over WebRTC, with your backend running this same loop over a control-plane sideband rather than a media bridge.
- Audio, images, and transcripts covers the PCM wire contract, playback, captions, input transcription, and image input.
- Events covers the session event vocabulary, which events are shared with standard runs, and the turn boundary.
- Turns and interruptions covers automatic turn detection, barge-in, output truncation, and push-to-talk.
- Tools covers function tools, provider-native tools, concurrency, approval, and delegation during a call.
- Capabilities and hooks covers how capabilities and their hooks map onto a session.
- History and handoff covers retained transcripts, audio and images, session seeding, and continuing with a standard text agent.
- Connecting a frontend covers the transport shapes between user devices and your backend.
- Connection lifecycle covers the session lifecycle, reconnection, session limits, and errors.
- Usage and observability covers usage limits, cost accounting, Logfire, and gateway trace propagation.
- Troubleshooting indexes common problems by symptom.
- The API reference lists session and codec types and explains how to implement another provider.
All providers implement the same RealtimeModel interface.
Provider pages are the canonical source for installation, model names, settings, feature support,
and quirks:
| Provider | Audio output | Image input | Text output | Browser WebRTC | Async tool calls | Thinking | State-restoring reconnect |
|---|---|---|---|---|---|---|---|
| OpenAI | ✓ | ✓ | ✓ | ✓ | ✓ | gpt-realtime-2* models | Replays local history |
| Azure OpenAI | ✓ | ✓ | ✓ | ✓ | ✓ | gpt-realtime-2* models | Replays local history |
| Google Gemini | ✓ | ✓ | ✗ | ✗ | Opt-in, native-audio models | ✓ | ✓, when enabled |
| xAI | ✓ | ✗ | ✗ | ✗ | ✗ | grok-voice-latest and -think- models | ✓ |
For portable branching, inspect RealtimeModel.profile
or RealtimeSession.profile: the
RealtimeModelProfile reports the audio sample rates
to capture and play at, plus one flag per capability in the table above and beyond. Profiles resolve
the same way as for a standard Model
(see Inspecting a model’s profile) — defaults,
then the provider’s knowledge of the model name, then your profile= argument on top. Pass
profile= when the model name doesn’t identify the model and the inferred facts are wrong, most
often with an Azure deployment named something other than its model:
from pydantic_ai.realtime.azure import AzureRealtimeModel
# The deployment serves a reasoning model, but nothing in its name says so.
model = AzureRealtimeModel('voice-prod', profile={'supports_thinking': True})
A partial dict is merged over the resolved profile; pass a callable
(resolved) -> RealtimeModelProfile instead to replace it wholesale.
Realtime sessions have their own settings type, playing the role that
model run settings play for standard runs:
RealtimeModelSettings defines the settings shared
across realtime providers, from tool_choice to
turn_detection. Set defaults with settings= on the
realtime model constructor, or pass realtime(model_settings=...) for one session; per-session
values override model defaults:
from pydantic_ai import Agent
from pydantic_ai.realtime import RealtimeModelSettings
agent = Agent(instructions='You are a helpful voice assistant.')
realtime = agent.realtime(
'openai:gpt-realtime', model_settings=RealtimeModelSettings(output_modality='audio')
)
Voices and detailed controls are provider-specific — openai_voice, google_voice, xai_voice
and friends live on the corresponding provider settings classes, with defaults and limitations on
the provider pages.
The agent’s regular model_settings and capability get_model_settings() contributions do not
configure realtime sessions. Unsupported shared settings are ignored, matching request-response
models, with one deliberate exception:
Agent.realtime() is the long-lived, bidirectional sibling of
run() and
iter(), and its parameters mirror theirs:
agent.realtime(
model, # 'openai:gpt-realtime', or a RealtimeModel instance
deps=..., # dependencies, as in run()/iter()
model_settings=..., # RealtimeModelSettings
instructions=..., # combined with the agent's instructions
toolsets=..., # additional toolsets for the session
capabilities=..., # additional capabilities for the session
usage=..., usage_limits=...,
message_history=..., # prior conversation to seed the session with
)
It accepts the same dependencies, instructions,
toolsets, capabilities,
usage limits, and message_history as a
standard run. Input arrives through the live session instead of a single user_prompt:
| Standard-run feature | In a realtime session |
|---|---|
| Function tools and tool hooks | ✓ — validation, retries, and execution hooks run as in a standard run |
Run hooks (before_run, after_run, wrap_run, on_run_error) | ✓ — once around the session |
| Capabilities, including third-party | ✓ — resolved once at connect |
| Event stream | ✓ — iterate the session, or attach ProcessEventStream |
output_type and output validators | ✗ — delegate to a text agent |
Graph node and model-request hooks (e.g. before_model_request) | ✗ — no agent graph |
| History processors at seeding | ✗ — preprocess before opening |
event_stream_handler parameter | ✗ — use ProcessEventStream |
See Capabilities and hooks for the full mapping, and hand off to a text agent for structured output or deeper reasoning.
The same realtime loop deploys to a browser or phone over WebRTC or a WebSocket relay without changing the agent code. If the realtime agent loop isn’t the right fit for a product, two alternatives sit outside it:
- Batch STT → text agent → TTS. Compose a standard agent with your own speech-to-text and text-to-speech services when you want a specific text model, structured output, or independently chosen speech components.
- Browser directly to the provider. A provider-native, UI-only experience using an ephemeral token: the provider’s own SDK owns the session, so there is no server-side agent loop, tools, or shared history — unlike the WebRTC sideband, where the browser owns the media but your backend still runs the agent. Pydantic AI can still power separate backend workflows.
| Limitation | Tracking |
|---|---|
| SIP is not built in; bridge telephony through a provider such as Twilio. | Connecting a frontend |
New tools cannot be advertised mid-session, so defer_loading=True tools and tool-contributing capabilities are rejected. | #7288 |
| Realtime-specific exchange hooks are not yet available; use supported tool hooks and session events. | #7190, #7191 |
| Provider resumption handles cannot be persisted and resumed in another process. | #7302 |
| Dynamic instructions are resolved once when the session connects. | #7303 |
History processors do not transform message_history before realtime seeding; preprocess it before opening the session when filtering or redaction is required. | #7299 |
Interactive human-in-the-loop tool approval is not supported: a HandleDeferredToolCalls handler resolves approvals from policy, immediately. | #7301 |
RunContext.enqueue() accepts one plain-text prompt per call, unlike its standard-run form. | #7300 |
| Gemini Live tool results are JSON-only: binary content attached to a tool return raises rather than being delivered. | #7362 |