feat: Add streaming tool output support

[pokliu]

Summary

This PR adds support for streaming tool output in the OpenAI Agents SDK. Previously, users could only receive tool results after complete execution. Now, tools can yield incremental output during execution, enabling real-time progress feedback for long-running operations.

Key improvements:

• New ToolOutputStreamEvent for receiving incremental tool output
• Support for async generator functions as tools (returning AsyncIterator[str])
• Automatic chunk accumulation for final LLM output
• Full backward compatibility with existing non-streaming tools
• Works in both Runner.run_streamed() and Runner.run() modes

Type system fix:
Fixed the type signature of FunctionTool.on_invoke_tool from:

Callable[[...], Awaitable[Any] | AsyncIterator[str]] 

to:

Callable[[...], Awaitable[Any | AsyncIterator[str]]]  

This properly represents that the function always returns an awaitable, which then yields either a regular value or an async iterator.

Test plan

New tests (tests/test_tool_streaming.py):

1. test_basic_streaming_tool - Verifies basic streaming functionality
2. test_streaming_tool_with_arguments - Tests parameter handling
3. test_mixed_streaming_and_non_streaming_tools - Validates coexistence
4. test_streaming_tool_accumulation - Confirms proper output accumulation
5. test_streaming_tool_in_non_streaming_mode - Tests Runner.run() mode
6. test_streaming_tool_agent_association - Checks event metadata

Example (examples/tools/streaming_tool_example.py):

• Runnable example demonstrating streaming tools with realistic use case

Validation:

make format  # ✅ All files formatted
make lint    # ✅ All checks passed
make mypy    # ✅ 347 files type-checked successfully
make tests   # ✅ 926 tests passed (including 6 new streaming tests)

Issue number

N/A - New feature enhancement

Checks

• I've added new tests (6 comprehensive tests)
• I've added/updated the relevant documentation (4 language versions)
• I've run make lint and make format (all passed)
• I've made sure tests pass (926 passed, 3 skipped)

---

Usage Example

from collections.abc import AsyncIterator
from agents import Agent, Runner, function_tool

@function_tool
async def search_documents(query: str) -> AsyncIterator[str]:
    """Search through documents and stream results."""
    documents = ["Doc 1...", "Doc 2...", "Doc 3..."]
    for doc in documents:
        await asyncio.sleep(0.5)
        yield doc

agent = Agent(
    name="Research Assistant",
    tools=[search_documents],
)

result = Runner.run_streamed(agent, input="Search for AI info")

async for event in result.stream_events():
    if event.type == "tool_output_stream_event":
        print(f"[{event.tool_name}] {event.delta}", end="", flush=True)

Documentation

Updated streaming documentation in 4 languages:

• English: docs/streaming.md
• Chinese: docs/zh/streaming.md
• Japanese: docs/ja/streaming.md
• Korean: docs/ko/streaming.md

All include complete examples and key points about streaming tools.

[seratch]

Thanks for sharing this idea. I see this could be useful for some use cases and it's an interesting enhancement idea. That being said, this requires a careful design decision, so we'll look into it later on.