feat(realtime): expose current_agent and context_wrapper on RealtimeSession

src/agents/realtime/session.py

@@ -217,6 +217,31 @@ def model(self) -> RealtimeModel:
         """Access the underlying model for adding listeners or other direct interaction."""
         return self._model
 
+    @property
+    def current_agent(self) -> RealtimeAgent:
+        """Return the agent that is currently active for this session.
+
+        This reflects the initial agent and is updated whenever the active agent changes,
+        for example after a handoff or a call to `update_agent()`. Use it to read the active
+        agent from code that runs outside the session's event loop, such as telemetry,
+        routing, or background timers. This returns a live, unsynchronized snapshot, so a
+        background reader can observe the agent mid-handoff because it is reassigned in
+        `update_agent()` and after a handoff without any locking.
+        """
+        return self._current_agent
+
+    @property
+    def context_wrapper(self) -> RunContextWrapper[Any]:
+        """Return the run context wrapper backing this session.
+
+        The returned wrapper is the same object the session was constructed with and exposes
+        the caller-provided context along with usage tracking. Use it to reach the run context
+        from code that runs outside the session's event loop. This returns a live,
+        unsynchronized reference whose usage counters are mutated by the session's event loop,
+        so a background reader can observe values updated without any locking.
+        """
+        return self._context_wrapper
+
     async def __aenter__(self) -> RealtimeSession:
         """Start the session by connecting to the model. After this, you will be able to stream
         events from the model and send messages and audio to the model.

tests/realtime/test_session.py

@@ -3651,6 +3651,38 @@ async def test_update_agent_validation_failure_keeps_current_agent(self, mock_mo
         assert mock_model.sent_events == []
 
 
+class TestSessionAccessors:
+    """Tests for the public ``current_agent`` and ``context_wrapper`` accessors."""
+
+    @pytest.mark.asyncio
+    async def test_current_agent_reflects_initial_agent(self, mock_model):
+        agent = RealtimeAgent(name="initial", instructions="initial", tools=[], handoffs=[])
+        session = RealtimeSession(mock_model, agent, None)
+
+        assert session.current_agent is agent
+
+    @pytest.mark.asyncio
+    async def test_current_agent_updates_after_update_agent(self, mock_model):
+        first_agent = RealtimeAgent(name="first", instructions="first", tools=[], handoffs=[])
+        second_agent = RealtimeAgent(name="second", instructions="second", tools=[], handoffs=[])
+        session = RealtimeSession(mock_model, first_agent, None)
+
+        assert session.current_agent is first_agent
+
+        await session.update_agent(second_agent)
+
+        assert session.current_agent is second_agent
+
+    @pytest.mark.asyncio
+    async def test_context_wrapper_returns_backing_wrapper(self, mock_model):
+        context = object()
+        agent = RealtimeAgent(name="agent", instructions="agent", tools=[], handoffs=[])
+        session = RealtimeSession(mock_model, agent, context)
+
+        assert session.context_wrapper is session._context_wrapper
+        assert session.context_wrapper.context is context
+
+
 class TestTranscriptPreservation:
     """Tests ensuring assistant transcripts are preserved across updates."""