diff --git a/src/agents/voice/workflow.py b/src/agents/voice/workflow.py index b3b9734e98..68b9cb82ef 100644 --- a/src/agents/voice/workflow.py +++ b/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): diff --git a/tests/voice/test_workflow.py b/tests/voice/test_workflow.py index af27007962..2f561a2a15 100644 --- a/tests/voice/test_workflow.py +++ b/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"]