RunResultStreaming.stream_events() raises ValueError when reset_current=True in async environments

[JilinJL]

Please read this first

• ✅ Have you read the docs? Agents SDK docs
• ✅ Have you searched for related issues? Yes, no existing issue was found.

---

Describe the bug

When using the RunResultStreaming.stream_events() method within a streaming API endpoint in frameworks like FastAPI (or any ASGI server), the following exception is raised at the end of the stream:

ValueError: <Token var= at 0x...> was created in a different Context

This error originates from self._trace.finish(reset_current=True) being executed across async task boundaries, which violates how ContextVar tokens must be reset in the same context they were set.

---

Debug information

• Agents SDK version: v0.0.3
• Python version: 3.12.2
• Framework: FastAPI + Uvicorn

---

My codes:

# the agent_func code
      if stream:
          return Runner.run_streamed(
              starting_agent=agent,
              input=input_items,
          ).stream_events()

# Here is the server layer:
@doc_agent_router.post("/agent/run")
async def run_agent(request: AgentRequest):
    try:
        agent_func = agent_map.get(request.mode)
        if not agent_func:
            raise ValueError("Invalid mode.")
            agent_kwargs = {
                "query": request.query,
                "original_content": request.content,
                "stream": request.stream
            }

        print("\n params :\n", agent_kwargs)
        result = await agent_func(**agent_kwargs)

        if request.stream:
            async def event_stream() -> AsyncGenerator[str, None]:
                async for chunk in result:
                    yield sse_event(chunk)
                yield sse_done()
            
            return StreamingResponse(
                event_stream(), 
                media_type="text/event-stream"
            )
        else:
            return BaseResponse(code=200, data=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"SLevel Internal server error: {str(e)}")

# stream_events
# .venv/lib/python3.12/site-packages/agents/result.py
@dataclass
class RunResultStreaming(RunResultBase):
    """The result of an agent run in streaming mode. You can use the `stream_events` method to
    receive semantic events as they are generated.

    The streaming method will raise:
    - A MaxTurnsExceeded exception if the agent exceeds the max_turns limit.
    - A GuardrailTripwireTriggered exception if a guardrail is tripped.
    """

    current_agent: Agent[Any]
    """The current agent that is running."""

    current_turn: int
    """The current turn number."""

    max_turns: int
    """The maximum number of turns the agent can run for."""

    final_output: Any
    """The final output of the agent. This is None until the agent has finished running."""

    _current_agent_output_schema: AgentOutputSchema | None = field(repr=False)

    _trace: Trace | None = field(repr=False)

    is_complete: bool = False
    """Whether the agent has finished running."""

    # Queues that the background run_loop writes to
    _event_queue: asyncio.Queue[StreamEvent | QueueCompleteSentinel] = field(
        default_factory=asyncio.Queue, repr=False
    )
    _input_guardrail_queue: asyncio.Queue[InputGuardrailResult] = field(
        default_factory=asyncio.Queue, repr=False
    )

    # Store the asyncio tasks that we're waiting on
    _run_impl_task: asyncio.Task[Any] | None = field(default=None, repr=False)
    _input_guardrails_task: asyncio.Task[Any] | None = field(default=None, repr=False)
    _output_guardrails_task: asyncio.Task[Any] | None = field(default=None, repr=False)
    _stored_exception: Exception | None = field(default=None, repr=False)

    @property
    def last_agent(self) -> Agent[Any]:
        """The last agent that was run. Updates as the agent run progresses, so the true last agent
        is only available after the agent run is complete.
        """
        return self.current_agent

    async def stream_events(self) -> AsyncIterator[StreamEvent]:
        """Stream deltas for new items as they are generated. We're using the types from the
        OpenAI Responses API, so these are semantic events: each event has a `type` field that
        describes the type of the event, along with the data for that event.

        This will raise:
        - A MaxTurnsExceeded exception if the agent exceeds the max_turns limit.
        - A GuardrailTripwireTriggered exception if a guardrail is tripped.
        """
        while True:
            self._check_errors()
            if self._stored_exception:
                logger.debug("Breaking due to stored exception")
                self.is_complete = True
                break

            if self.is_complete and self._event_queue.empty():
                break

            try:
                item = await self._event_queue.get()
            except asyncio.CancelledError:
                break

            if isinstance(item, QueueCompleteSentinel):
                self._event_queue.task_done()
                # Check for errors, in case the queue was completed due to an exception
                self._check_errors()
                break

            yield item
            self._event_queue.task_done()

        if self._trace:
            self._trace.finish(reset_current=True)  # here is

        self._cleanup_tasks()

        if self._stored_exception:
            raise self._stored_exception
  ...

Repro steps

Use the following minimal example with FastAPI:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai_agents import create_openai_agent
from openai_agents.executor import AgentExecutor

app = FastAPI()
agent = create_openai_agent(tools=[])

@app.get("/stream")
async def stream():
    executor = AgentExecutor(agent=agent)
    result = await executor.run_stream(query="Hello!", stream=True)

    async def stream_gen():
        async for event in result.stream_events():  # ❗ Error occurs here after stream ends
            yield event.delta.text

    return StreamingResponse(stream_gen(), media_type="text/event-stream")

### Root Cause

This happens because `self._trace.finish(reset_current=True)` attempts to reset a `ContextVar` from a different asynchronous context than the one it was created in.

### Result
After successful streaming, the endpoint raises a ValueError.

### Expected behavior
The stream should close cleanly without internal exceptions. The tracing system should finalize safely, even across asynchronous contexts.

[JilinJL]

error message:

During handling of the above exception, another exception occurred:

  + Exception Group Traceback (most recent call last):
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/uvicorn/protocols/http/httptools_impl.py", line 409, in run_asgi
  |     result = await app(  # type: ignore[func-returns-value]
  |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
  |     return await self.app(scope, receive, send)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/fastapi/applications.py", line 1054, in __call__
  |     await super().__call__(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/applications.py", line 112, in __call__
  |     await self.middleware_stack(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/middleware/errors.py", line 187, in __call__
  |     raise exc
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/middleware/errors.py", line 165, in __call__
  |     await self.app(scope, receive, _send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/middleware/cors.py", line 93, in __call__
  |     await self.simple_response(scope, receive, send, request_headers=headers)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/middleware/cors.py", line 144, in simple_response
  |     await self.app(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
  |     await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
  |     raise exc
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
  |     await app(scope, receive, sender)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/routing.py", line 715, in __call__
  |     await self.middleware_stack(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/routing.py", line 735, in app
  |     await route.handle(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/routing.py", line 288, in handle
  |     await self.app(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/routing.py", line 76, in app
  |     await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
  |     raise exc
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
  |     await app(scope, receive, sender)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/routing.py", line 74, in app
  |     await response(scope, receive, send)
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/responses.py", line 261, in __call__
  |     async with anyio.create_task_group() as task_group:
  |                ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/home/gpower/.venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 767, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/responses.py", line 264, in wrap
    |     await func()
    |   File "/home/gpower/.venv/lib/python3.12/site-packages/starlette/responses.py", line 245, in stream_response
    |     async for chunk in self.body_iterator:
    |   File "/home/gpower/gpower/api/routes/doc_agents_server.py", line 42, in event_stream
    |     async for chunk in result:
    |   File "/home/gpower/.venv/lib/python3.12/site-packages/agents/result.py", line 181, in stream_events
    |     self._trace.finish(reset_current=True)
    |   File "/home/gpower/.venv/lib/python3.12/site-packages/agents/tracing/traces.py", line 173, in finish
    |     Scope.reset_current_trace(self._prev_context_token)
    |   File "/home/gpower/.venv/lib/python3.12/site-packages/agents/tracing/scope.py", line 45, in reset_current_trace
    |     _current_trace.reset(token)
    | ValueError: <Token var=<ContextVar name='current_trace' default=None at 0x7ff7b810c770> at 0x7ff76bcac0c0> was created in a different Context

![Image](https://github.com/user-attachments/assets/8b5de322-d343-41a2-a6dc-9cbab69be2cf)

[JilinJL]

As you can see, streaming output directly results in an error at the end.

[JilinJL]

I have fixed this bug and I plan to submit a PR.

[JilinJL]

Oh sry, I mistakenly reopened this issue 😂.