Skip to content

RealtimeRunner

A RealtimeRunner is the equivalent of Runner for realtime agents. It automatically handles multiple turns by maintaining a persistent connection with the underlying model layer.

The session manages the local history copy, executes tools, runs guardrails and facilitates handoffs between agents.

Since this code runs on your server, it uses WebSockets by default. You can optionally create your own custom model layer by implementing the RealtimeModel interface.

Source code in src/agents/realtime/runner.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class RealtimeRunner:
    """A `RealtimeRunner` is the equivalent of `Runner` for realtime agents. It automatically
    handles multiple turns by maintaining a persistent connection with the underlying model
    layer.

    The session manages the local history copy, executes tools, runs guardrails and facilitates
    handoffs between agents.

    Since this code runs on your server, it uses WebSockets by default. You can optionally create
    your own custom model layer by implementing the `RealtimeModel` interface.
    """

    def __init__(
        self,
        starting_agent: RealtimeAgent,
        *,
        model: RealtimeModel | None = None,
        config: RealtimeRunConfig | None = None,
    ) -> None:
        """Initialize the realtime runner.

        Args:
            starting_agent: The agent to start the session with.
            context: The context to use for the session.
            model: The model to use. If not provided, will use a default OpenAI realtime model.
            config: Override parameters to use for the entire run.
        """
        self._starting_agent = starting_agent
        self._config = config
        self._model = model or OpenAIRealtimeWebSocketModel()

    async def run(
        self, *, context: TContext | None = None, model_config: RealtimeModelConfig | None = None
    ) -> RealtimeSession:
        """Start and returns a realtime session.

        Returns:
            RealtimeSession: A session object that allows bidirectional communication with the
            realtime model.

        Example:
            ```python
            runner = RealtimeRunner(agent)
            async with await runner.run() as session:
                await session.send_message("Hello")
                async for event in session:
                    print(event)
            ```
        """
        model_settings = await self._get_model_settings(
            agent=self._starting_agent,
            disable_tracing=self._config.get("tracing_disabled", False) if self._config else False,
            initial_settings=model_config.get("initial_model_settings") if model_config else None,
            overrides=self._config.get("model_settings") if self._config else None,
        )

        model_config = model_config.copy() if model_config else {}
        model_config["initial_model_settings"] = model_settings

        # Create and return the connection
        session = RealtimeSession(
            model=self._model,
            agent=self._starting_agent,
            context=context,
            model_config=model_config,
            run_config=self._config,
        )

        return session

    async def _get_model_settings(
        self,
        agent: RealtimeAgent,
        disable_tracing: bool,
        context: TContext | None = None,
        initial_settings: RealtimeSessionModelSettings | None = None,
        overrides: RealtimeSessionModelSettings | None = None,
    ) -> RealtimeSessionModelSettings:
        context_wrapper = RunContextWrapper(context)
        model_settings = initial_settings.copy() if initial_settings else {}

        instructions, tools = await asyncio.gather(
            agent.get_system_prompt(context_wrapper),
            agent.get_all_tools(context_wrapper),
        )

        if instructions is not None:
            model_settings["instructions"] = instructions
        if tools is not None:
            model_settings["tools"] = tools

        if overrides:
            model_settings.update(overrides)

        if disable_tracing:
            model_settings["tracing"] = None

        return model_settings

__init__ 

__init__(
    starting_agent: RealtimeAgent,
    *,
    model: RealtimeModel | None = None,
    config: RealtimeRunConfig | None = None,
) -> None

Initialize the realtime runner.

Parameters:

Name Type Description Default
starting_agent RealtimeAgent

The agent to start the session with.

 required
context

The context to use for the session.

 required
model RealtimeModel | None

The model to use. If not provided, will use a default OpenAI realtime model.

 None
config RealtimeRunConfig | None

Override parameters to use for the entire run.

 None
Source code in src/agents/realtime/runner.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    starting_agent: RealtimeAgent,
    *,
    model: RealtimeModel | None = None,
    config: RealtimeRunConfig | None = None,
) -> None:
    """Initialize the realtime runner.

    Args:
        starting_agent: The agent to start the session with.
        context: The context to use for the session.
        model: The model to use. If not provided, will use a default OpenAI realtime model.
        config: Override parameters to use for the entire run.
    """
    self._starting_agent = starting_agent
    self._config = config
    self._model = model or OpenAIRealtimeWebSocketModel()

run async 

run(
    *,
    context: TContext | None = None,
    model_config: RealtimeModelConfig | None = None,
) -> RealtimeSession

Start and returns a realtime session.

Returns:

Name Type Description
RealtimeSession RealtimeSession

A session object that allows bidirectional communication with the
RealtimeSession

realtime model.
Example 
runner = RealtimeRunner(agent)
async with await runner.run() as session:
    await session.send_message("Hello")
    async for event in session:
        print(event)
Source code in src/agents/realtime/runner.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def run(
    self, *, context: TContext | None = None, model_config: RealtimeModelConfig | None = None
) -> RealtimeSession:
    """Start and returns a realtime session.

    Returns:
        RealtimeSession: A session object that allows bidirectional communication with the
        realtime model.

    Example:
        ```python
        runner = RealtimeRunner(agent)
        async with await runner.run() as session:
            await session.send_message("Hello")
            async for event in session:
                print(event)
        ```
    """
    model_settings = await self._get_model_settings(
        agent=self._starting_agent,
        disable_tracing=self._config.get("tracing_disabled", False) if self._config else False,
        initial_settings=model_config.get("initial_model_settings") if model_config else None,
        overrides=self._config.get("model_settings") if self._config else None,
    )

    model_config = model_config.copy() if model_config else {}
    model_config["initial_model_settings"] = model_settings

    # Create and return the connection
    session = RealtimeSession(
        model=self._model,
        agent=self._starting_agent,
        context=context,
        model_config=model_config,
        run_config=self._config,
    )

    return session