feat(realtime): add typed create_response helper to RealtimeSession

docs/realtime/guide.md

@@ -129,6 +129,19 @@ If you need lower-level control, you can also send raw client events such as `in
 
 `session.send_message()` sends user input using the high-level path and starts a response for you. Raw audio buffering does **not** automatically do the same in every configuration.
 
+Use [`session.create_response()`][agents.realtime.session.RealtimeSession.create_response] to trigger a model response without adding a new user message, for example after committing audio manually or when you want the model to respond on demand:
+
+```python
+await session.create_response()
+
+await session.create_response(
+    instructions="Reply in one short sentence.",
+    metadata={"turn": "greeting"},
+)
+```
+
+`instructions` and `metadata` are per-response overrides that apply to this response only. They do not mutate the agent's instructions or the session configuration. Both are optional, and when omitted the model responds using the current session configuration.
+
 At the Realtime API level, manual turn control means clearing `turn_detection` with a raw `session.update`, then sending `input_audio_buffer.commit` and `response.create` yourself.
 
 If you are managing turns manually, you can send raw client events through the model transport:

src/agents/realtime/__init__.py

@@ -80,6 +80,7 @@
     RealtimeModelSendEvent,
     RealtimeModelSendInterrupt,
     RealtimeModelSendRawMessage,
+    RealtimeModelSendResponseCreate,
     RealtimeModelSendSessionUpdate,
     RealtimeModelSendToolOutput,
     RealtimeModelSendUserInput,
@@ -178,6 +179,7 @@
     "RealtimeModelSendEvent",
     "RealtimeModelSendInterrupt",
     "RealtimeModelSendRawMessage",
+    "RealtimeModelSendResponseCreate",
     "RealtimeModelSendSessionUpdate",
     "RealtimeModelSendToolOutput",
     "RealtimeModelSendUserInput",

src/agents/realtime/model_inputs.py

@@ -107,11 +107,23 @@ class RealtimeModelSendSessionUpdate:
     """The updated session settings to send."""
 
 
+@dataclass
+class RealtimeModelSendResponseCreate:
+    """Trigger a new model response via a `response.create` event."""
+
+    instructions: str | None = None
+    """Optional instructions that override the session instructions for this response only."""
+
+    metadata: dict[str, Any] | None = None
+    """Optional metadata to attach to this response."""
+
+
 RealtimeModelSendEvent: TypeAlias = (
     RealtimeModelSendRawMessage
     | RealtimeModelSendUserInput
     | RealtimeModelSendAudio
     | RealtimeModelSendToolOutput
     | RealtimeModelSendInterrupt
     | RealtimeModelSendSessionUpdate
+    | RealtimeModelSendResponseCreate
 )

src/agents/realtime/openai_realtime.py

@@ -57,6 +57,9 @@
 from openai.types.realtime.realtime_function_tool import (
     RealtimeFunctionTool as OpenAISessionFunction,
 )
+from openai.types.realtime.realtime_response_create_params import (
+    RealtimeResponseCreateParams as OpenAIRealtimeResponseCreateParams,
+)
 from openai.types.realtime.realtime_server_event import (
     RealtimeServerEvent as OpenAIRealtimeServerEvent,
 )
@@ -140,6 +143,7 @@
     RealtimeModelSendEvent,
     RealtimeModelSendInterrupt,
     RealtimeModelSendRawMessage,
+    RealtimeModelSendResponseCreate,
     RealtimeModelSendSessionUpdate,
     RealtimeModelSendToolOutput,
     RealtimeModelSendUserInput,
@@ -726,6 +730,8 @@ async def send_event(self, event: RealtimeModelSendEvent) -> None:
             await self._send_interrupt(event)
         elif isinstance(event, RealtimeModelSendSessionUpdate):
             await self._send_session_update(event)
+        elif isinstance(event, RealtimeModelSendResponseCreate):
+            await self._send_response_create(event)
         else:
             assert_never(event)
             raise ValueError(f"Unknown event type: {type(event)}")
@@ -977,6 +983,16 @@ async def _send_session_update(self, event: RealtimeModelSendSessionUpdate) -> N
         """Send a session update to the model."""
         await self._update_session_config(event.session_settings)
 
+    async def _send_response_create(self, event: RealtimeModelSendResponseCreate) -> None:
+        """Send a response.create through the sequencer so it cannot collide with an active one."""
+        converted = _ConversionHelper.convert_response_create(event)
+        request_version = await self._reserve_response_create_request(manual=True)
+        self._start_response_create(
+            request_version,
+            response_create=converted,
+            manual=True,
+        )
+
     async def _handle_audio_delta(self, parsed: ResponseAudioDeltaEvent) -> None:
         """Handle audio delta events and update audio tracking state."""
         self._current_item_id = parsed.item_id
@@ -1705,6 +1721,24 @@ def try_convert_raw_message(
         except Exception:
             return None
 
+    @classmethod
+    def convert_response_create(
+        cls, event: RealtimeModelSendResponseCreate
+    ) -> OpenAIResponseCreateEvent:
+        response_params: dict[str, Any] = {}
+        if event.instructions is not None:
+            response_params["instructions"] = event.instructions
+        if event.metadata is not None:
+            response_params["metadata"] = event.metadata
+        # Omit the response field entirely when no params are provided. Passing response=None
+        # explicitly marks the field as set, so model_dump_json(exclude_unset=True) would emit
+        # {"response": null} instead of a bare {"type": "response.create"}, which the Realtime
+        # schema can reject.
+        if not response_params:
+            return OpenAIResponseCreateEvent(type="response.create")
+        response = OpenAIRealtimeResponseCreateParams(**response_params)
+        return OpenAIResponseCreateEvent(type="response.create", response=response)
+
     @classmethod
     def convert_tracing_config(
         cls, tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None

src/agents/realtime/session.py

@@ -69,6 +69,7 @@
 from .model_inputs import (
     RealtimeModelSendAudio,
     RealtimeModelSendInterrupt,
+    RealtimeModelSendResponseCreate,
     RealtimeModelSendSessionUpdate,
     RealtimeModelSendToolOutput,
     RealtimeModelSendUserInput,
@@ -300,6 +301,22 @@ async def interrupt(self) -> None:
         """Interrupt the model."""
         await self._model.send_event(RealtimeModelSendInterrupt())
 
+    async def create_response(
+        self,
+        *,
+        instructions: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> None:
+        """Trigger a new model response with optional per-response instructions and metadata.
+
+        The provided ``instructions`` and ``metadata`` override the session configuration for this
+        response only and do not mutate the agent instructions. Both arguments are optional; when
+        omitted, the model creates a response using the current session configuration.
+        """
+        await self._model.send_event(
+            RealtimeModelSendResponseCreate(instructions=instructions, metadata=metadata)
+        )
+
     async def update_agent(self, agent: RealtimeAgent) -> None:
         """Update the active agent for this session and apply its settings to the model."""
         updated_settings = await self._get_updated_model_settings_from_agent(

tests/realtime/test_openai_realtime.py

@@ -22,6 +22,7 @@
     RealtimeModelSendAudio,
     RealtimeModelSendInterrupt,
     RealtimeModelSendRawMessage,
+    RealtimeModelSendResponseCreate,
     RealtimeModelSendSessionUpdate,
     RealtimeModelSendToolOutput,
     RealtimeModelSendUserInput,
@@ -957,6 +958,96 @@ async def test_interrupt_respects_auto_cancellation_when_not_forced(self, model,
         assert all(call.args[0].type != "response.cancel" for call in send_raw.call_args_list)
         assert model._ongoing_response is True
 
+    @pytest.mark.asyncio
+    async def test_create_response_without_args_emits_minimal_response_create(
+        self, model, monkeypatch
+    ):
+        """create_response with no fields should emit a response.create with no response object."""
+        sent_events: list[Any] = []
+
+        async def fake_send_raw(event):
+            sent_events.append(event)
+
+        monkeypatch.setattr(model, "_send_raw_message", fake_send_raw)
+
+        await model.send_event(RealtimeModelSendResponseCreate())
+        await asyncio.sleep(0)
+
+        assert [e.type for e in sent_events] == ["response.create"]
+        # No fields were provided, so the response object is omitted entirely (unset, not null),
+        # and the serialized payload is a bare response.create.
+        assert sent_events[0].response is None
+        assert "response" not in sent_events[0].model_fields_set
+        dumped = json.loads(sent_events[0].model_dump_json(exclude_unset=True))
+        assert "response" not in dumped
+        # The request was routed through the sequencer rather than sent directly.
+        assert model._response_control == "create_requested"
+
+    @pytest.mark.asyncio
+    async def test_create_response_includes_only_provided_fields(self, model, monkeypatch):
+        """create_response should forward instructions/metadata and omit unset fields."""
+        sent_events: list[Any] = []
+
+        async def fake_send_raw(event):
+            sent_events.append(event)
+
+        monkeypatch.setattr(model, "_send_raw_message", fake_send_raw)
+
+        await model.send_event(
+            RealtimeModelSendResponseCreate(instructions="be brief", metadata={"turn": "1"})
+        )
+        await asyncio.sleep(0)
+
+        assert [e.type for e in sent_events] == ["response.create"]
+        response = sent_events[0].response
+        assert response is not None
+        assert response.instructions == "be brief"
+        assert response.metadata == {"turn": "1"}
+
+        dumped = json.loads(sent_events[0].model_dump_json(exclude_unset=True))
+        assert dumped["type"] == "response.create"
+        assert dumped["response"] == {"instructions": "be brief", "metadata": {"turn": "1"}}
+
+    @pytest.mark.asyncio
+    async def test_create_response_with_only_instructions_omits_metadata(self, model, monkeypatch):
+        """Unset fields should not appear in the serialized response.create payload."""
+        sent_events: list[Any] = []
+
+        async def fake_send_raw(event):
+            sent_events.append(event)
+
+        monkeypatch.setattr(model, "_send_raw_message", fake_send_raw)
+
+        await model.send_event(RealtimeModelSendResponseCreate(instructions="hello"))
+        await asyncio.sleep(0)
+
+        dumped = json.loads(sent_events[0].model_dump_json(exclude_unset=True))
+        assert dumped["response"] == {"instructions": "hello"}
+
+    @pytest.mark.asyncio
+    async def test_create_response_defers_while_response_active(self, model, monkeypatch):
+        """create_response should wait for an in-flight response instead of colliding with it."""
+        payload_types: list[str] = []
+
+        async def fake_send_raw(event):
+            payload_types.append(event.type)
+
+        monkeypatch.setattr(model, "_send_raw_message", fake_send_raw)
+        await model._mark_response_created()
+
+        await model.send_event(RealtimeModelSendResponseCreate())
+        await asyncio.sleep(0)
+
+        # A response is already active, so the new response.create is deferred.
+        assert payload_types == []
+        assert model._ongoing_response is True
+
+        await model._mark_response_done()
+        await asyncio.sleep(0)
+
+        # Once the active response completes, the queued response.create is sent.
+        assert payload_types == ["response.create"]
+
     @pytest.mark.asyncio
     async def test_send_user_input_defers_response_create_without_blocking_caller(
         self, model, monkeypatch

tests/realtime/test_openai_realtime_conversions.py

@@ -14,6 +14,7 @@
 from agents.realtime.config import RealtimeModelTracingConfig
 from agents.realtime.model_inputs import (
     RealtimeModelSendRawMessage,
+    RealtimeModelSendResponseCreate,
     RealtimeModelSendUserInput,
     RealtimeModelUserInputMessage,
 )
@@ -230,3 +231,26 @@ def test_tools_to_session_tools_rejects_deferred_function_tools():
 
     with pytest.raises(UserError, match="defer_loading=True"):
         m._tools_to_session_tools([tool], [])
+
+
+def test_convert_response_create_includes_only_provided_fields():
+    empty = _ConversionHelper.convert_response_create(RealtimeModelSendResponseCreate())
+    assert empty.type == "response.create"
+    assert empty.response is None
+    # The response field must be unset (not explicitly None) so it is omitted from the payload.
+    assert "response" not in empty.model_fields_set
+
+    populated = _ConversionHelper.convert_response_create(
+        RealtimeModelSendResponseCreate(instructions="be brief", metadata={"turn": "1"})
+    )
+    assert populated.response is not None
+    assert populated.response.instructions == "be brief"
+    assert populated.response.metadata == {"turn": "1"}
+
+
+def test_realtime_model_send_response_create_is_exported():
+    import agents.realtime as realtime_pkg
+    from agents.realtime import RealtimeModelSendResponseCreate as ExportedResponseCreate
+
+    assert ExportedResponseCreate is RealtimeModelSendResponseCreate
+    assert "RealtimeModelSendResponseCreate" in realtime_pkg.__all__

tests/realtime/test_session.py

@@ -56,6 +56,7 @@
 from agents.realtime.model_inputs import (
     RealtimeModelSendAudio,
     RealtimeModelSendInterrupt,
+    RealtimeModelSendResponseCreate,
     RealtimeModelSendSessionUpdate,
     RealtimeModelSendToolOutput,
     RealtimeModelSendUserInput,
@@ -164,6 +165,28 @@ async def test_property_and_send_helpers_and_enter_alias():
         assert any(isinstance(e, RealtimeModelSendInterrupt) for e in model.events)
 
 
+@pytest.mark.asyncio
+async def test_create_response_sends_typed_event():
+    model = _DummyModel()
+    agent = RealtimeAgent(name="agent")
+    session = RealtimeSession(model, agent, None)
+
+    async with await session.enter():
+        # No arguments should send an empty typed response.create input.
+        await session.create_response()
+        # Explicit instructions and metadata should be forwarded unchanged.
+        await session.create_response(instructions="be brief", metadata={"turn": "1"})
+
+    response_events = [e for e in model.events if isinstance(e, RealtimeModelSendResponseCreate)]
+    assert len(response_events) == 2
+
+    assert response_events[0].instructions is None
+    assert response_events[0].metadata is None
+
+    assert response_events[1].instructions == "be brief"
+    assert response_events[1].metadata == {"turn": "1"}
+
+
 @pytest.mark.asyncio
 async def test_aiter_cancel_propagates_cancelled_error():
     model = _DummyModel()