fix(voice): support context in single-agent workflow

src/agents/voice/workflow.py

@@ -2,12 +2,12 @@
 
 import abc
 from collections.abc import AsyncIterator
-from typing import Any
 
 from ..agent import Agent
 from ..items import TResponseInputItem
 from ..result import RunResultStreaming
 from ..run import Runner
+from ..run_context import TContext
 
 
 class VoiceWorkflowBase(abc.ABC):
@@ -66,16 +66,24 @@ class SingleAgentVoiceWorkflow(VoiceWorkflowBase):
     custom configs), subclass `VoiceWorkflowBase` and implement your own logic.
     """
 
-    def __init__(self, agent: Agent[Any], callbacks: SingleAgentWorkflowCallbacks | None = None):
+    def __init__(
+        self,
+        agent: Agent[TContext],
+        callbacks: SingleAgentWorkflowCallbacks | None = None,
+        *,
+        context: TContext | None = None,
+    ):
         """Create a new single agent voice workflow.
 
         Args:
             agent: The agent to run.
             callbacks: Optional callbacks to call during the workflow.
+            context: Optional application context forwarded to every agent run.
         """
         self._input_history: list[TResponseInputItem] = []
         self._current_agent = agent
         self._callbacks = callbacks
+        self._context = context
 
     async def run(self, transcription: str) -> AsyncIterator[str]:
         if self._callbacks is not None:
@@ -90,7 +98,11 @@ async def run(self, transcription: str) -> AsyncIterator[str]:
         )
 
         # Run the agent
-        result = Runner.run_streamed(self._current_agent, self._input_history)
+        result = Runner.run_streamed(
+            self._current_agent,
+            self._input_history,
+            context=self._context,
+        )
 
         # Stream the text from the result
         async for chunk in VoiceWorkflowHelper.stream_text_from(result):

tests/voice/test_workflow.py

@@ -5,7 +5,8 @@
 import pytest
 from inline_snapshot import snapshot
 
-from agents import Agent
+from agents import Agent, RunContextWrapper
+from agents.decorators import tool
 from agents.testing import ScriptedModel
 
 from ..test_responses import get_function_tool, get_function_tool_call, get_text_message
@@ -136,3 +137,35 @@ async def test_single_agent_workflow(monkeypatch) -> None:
         ]
     )
     assert workflow._current_agent == agent
+
+
+@pytest.mark.asyncio
+async def test_single_agent_workflow_forwards_context_on_every_turn() -> None:
+    @tool
+    def read_user_id(ctx: RunContextWrapper[dict[str, str]]) -> str:
+        """Return the current user ID."""
+        return ctx.context["user_id"]
+
+    model = ScriptedModel()
+    model.extend(
+        [
+            [get_function_tool_call("read_user_id", "{}", call_id="context_call_1")],
+            [get_text_message("first turn done")],
+            [get_function_tool_call("read_user_id", "{}", call_id="context_call_2")],
+            [get_text_message("second turn done")],
+        ]
+    )
+    agent = Agent("context_agent", model=model, tools=[read_user_id])
+    workflow = SingleAgentVoiceWorkflow(agent, context={"user_id": "user-123"})
+
+    first_output = [chunk async for chunk in workflow.run("first transcription")]
+    second_output = [chunk async for chunk in workflow.run("second transcription")]
+
+    assert first_output == ["first turn done"]
+    assert second_output == ["second turn done"]
+    tool_outputs = [
+        item["output"]
+        for item in workflow._input_history
+        if item.get("type") == "function_call_output"
+    ]
+    assert tool_outputs == ["user-123", "user-123"]