feat: Add streaming tool output support

Author: pokliu

docs/ja/streaming.md

@@ -88,4 +88,64 @@ async def main():
 
 if __name__ == "__main__":
     asyncio.run(main())
-```
+```
+
+## ツール出力ストリーミングイベント
+
+[`ToolOutputStreamEvent`][agents.stream_events.ToolOutputStreamEvent] を使用すると、ツールの実行中に増分出力を受け取ることができます。これは、長時間実行されるツールでユーザーにリアルタイムで進捗を表示したい場合に有用です。
+
+ストリーミングツールを作成するには、文字列チャンクを yield する非同期ジェネレータ関数を定義します：
+
+```python
+import asyncio
+from collections.abc import AsyncIterator
+from agents import Agent, Runner, ToolOutputStreamEvent, function_tool
+
+@function_tool
+async def search_documents(query: str) -> AsyncIterator[str]:
+    """ドキュメントを検索し、見つかった結果をストリーミングします。"""
+    documents = [
+        f"ドキュメント 1 には {query} に関する情報が含まれています...\n",
+        f"ドキュメント 2 には {query} に関する追加の詳細があります...\n",
+        f"ドキュメント 3 は {query} の分析を提供します...\n",
+    ]
+    
+    for doc in documents:
+        # 処理時間をシミュレート
+        await asyncio.sleep(0.5)
+        # 増分結果を yield
+        yield doc
+
+
+async def main():
+    agent = Agent(
+        name="Research Assistant",
+        instructions="あなたはユーザーの情報検索を支援します。",
+        tools=[search_documents],
+    )
+
+    result = Runner.run_streamed(
+        agent,
+        input="AI に関する情報を検索してください",
+    )
+
+    async for event in result.stream_events():
+        # ツールストリーミングイベントを処理
+        if event.type == "tool_output_stream_event":
+            print(f"[{event.tool_name}] {event.delta}", end="", flush=True)
+        # 最終ツール出力を処理
+        elif event.type == "run_item_stream_event" and event.name == "tool_output":
+            print(f"\n✓ ツール完了\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+ストリーミングツールに関する重要なポイント：
+
+- ストリーミングツールは `AsyncIterator[str]`（文字列を yield する非同期ジェネレータ）を返す必要があります
+- yield された各チャンクは `ToolOutputStreamEvent` として発行されます
+- すべてのチャンクは自動的に蓄積され、最終的なツール出力として LLM に送信されます
+- 非ストリーミングツールはストリーミングツールと一緒に正常に動作します
+- 非ストリーミングモード（`Runner.run()`）では、ストリーミングツールは返す前にすべてのチャンクを自動的に収集します

docs/ko/streaming.md

@@ -88,4 +88,64 @@ async def main():
 
 if __name__ == "__main__":
     asyncio.run(main())
-```
+```
+
+## 도구 출력 스트리밍 이벤트
+
+[`ToolOutputStreamEvent`][agents.stream_events.ToolOutputStreamEvent]를 사용하면 도구가 실행되는 동안 증분 출력을 받을 수 있습니다. 이는 장시간 실행되는 도구에서 사용자에게 실시간으로 진행 상황을 표시하려는 경우에 유용합니다.
+
+스트리밍 도구를 만들려면 문자열 청크를 yield하는 비동기 제너레이터 함수를 정의하세요:
+
+```python
+import asyncio
+from collections.abc import AsyncIterator
+from agents import Agent, Runner, ToolOutputStreamEvent, function_tool
+
+@function_tool
+async def search_documents(query: str) -> AsyncIterator[str]:
+    """문서를 검색하고 발견된 결과를 스트리밍합니다."""
+    documents = [
+        f"문서 1에는 {query}에 대한 정보가 포함되어 있습니다...\n",
+        f"문서 2에는 {query}에 대한 추가 세부정보가 있습니다...\n",
+        f"문서 3은 {query}에 대한 분석을 제공합니다...\n",
+    ]
+    
+    for doc in documents:
+        # 처리 시간 시뮬레이션
+        await asyncio.sleep(0.5)
+        # 증분 결과 yield
+        yield doc
+
+
+async def main():
+    agent = Agent(
+        name="Research Assistant",
+        instructions="당신은 사용자가 정보를 검색하도록 돕습니다.",
+        tools=[search_documents],
+    )
+
+    result = Runner.run_streamed(
+        agent,
+        input="AI에 관한 정보를 검색하세요",
+    )
+
+    async for event in result.stream_events():
+        # 도구 스트리밍 이벤트 처리
+        if event.type == "tool_output_stream_event":
+            print(f"[{event.tool_name}] {event.delta}", end="", flush=True)
+        # 최종 도구 출력 처리
+        elif event.type == "run_item_stream_event" and event.name == "tool_output":
+            print(f"\n✓ 도구 완료\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+스트리밍 도구에 대한 주요 사항:
+
+- 스트리밍 도구는 `AsyncIterator[str]`(문자열을 yield하는 비동기 제너레이터)을 반환해야 합니다
+- yield된 각 청크는 `ToolOutputStreamEvent`로 발행됩니다
+- 모든 청크는 자동으로 누적되어 최종 도구 출력으로 LLM에 전송됩니다
+- 비스트리밍 도구는 스트리밍 도구와 함께 정상적으로 작동합니다
+- 비스트리밍 모드(`Runner.run()`)에서 스트리밍 도구는 반환하기 전에 모든 청크를 자동으로 수집합니다

docs/streaming.md

@@ -85,3 +85,63 @@ async def main():
 if __name__ == "__main__":
     asyncio.run(main())
 ```
+
+## Tool output streaming events
+
+[`ToolOutputStreamEvent`][agents.stream_events.ToolOutputStreamEvent] allows you to receive incremental output from tools as they execute. This is useful for long-running tools where you want to show progress to the user in real-time.
+
+To create a streaming tool, define an async generator function that yields string chunks:
+
+```python
+import asyncio
+from collections.abc import AsyncIterator
+from agents import Agent, Runner, ToolOutputStreamEvent, function_tool
+
+@function_tool
+async def search_documents(query: str) -> AsyncIterator[str]:
+    """Search through documents and stream results as they are found."""
+    documents = [
+        f"Document 1 contains information about {query}...\n",
+        f"Document 2 has additional details on {query}...\n",
+        f"Document 3 provides analysis of {query}...\n",
+    ]
+    
+    for doc in documents:
+        # Simulate processing time
+        await asyncio.sleep(0.5)
+        # Yield incremental results
+        yield doc
+
+
+async def main():
+    agent = Agent(
+        name="Research Assistant",
+        instructions="You help users search for information.",
+        tools=[search_documents],
+    )
+
+    result = Runner.run_streamed(
+        agent,
+        input="Search for information about AI",
+    )
+
+    async for event in result.stream_events():
+        # Handle tool streaming events
+        if event.type == "tool_output_stream_event":
+            print(f"[{event.tool_name}] {event.delta}", end="", flush=True)
+        # Handle final tool output
+        elif event.type == "run_item_stream_event" and event.name == "tool_output":
+            print(f"\n✓ Tool completed\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+Key points about streaming tools:
+
+- Streaming tools must return `AsyncIterator[str]` (an async generator that yields strings)
+- Each yielded chunk is emitted as a `ToolOutputStreamEvent`
+- All chunks are automatically accumulated and sent to the LLM as the final tool output
+- Non-streaming tools continue to work normally alongside streaming tools
+- In non-streaming mode (`Runner.run()`), streaming tools automatically collect all chunks before returning

docs/zh/streaming.md

@@ -88,4 +88,64 @@ async def main():
 
 if __name__ == "__main__":
     asyncio.run(main())
-```
+```
+
+## 工具输出流式事件
+
+[`ToolOutputStreamEvent`][agents.stream_events.ToolOutputStreamEvent] 允许你在工具执行时接收增量输出。这对于长时间运行的工具非常有用，可以实时向用户显示进度。
+
+要创建流式工具，请定义一个异步生成器函数，逐块 yield 字符串：
+
+```python
+import asyncio
+from collections.abc import AsyncIterator
+from agents import Agent, Runner, ToolOutputStreamEvent, function_tool
+
+@function_tool
+async def search_documents(query: str) -> AsyncIterator[str]:
+    """搜索文档并流式返回找到的结果。"""
+    documents = [
+        f"文档 1 包含关于 {query} 的信息...\n",
+        f"文档 2 提供关于 {query} 的更多细节...\n",
+        f"文档 3 分析了 {query}...\n",
+    ]
+    
+    for doc in documents:
+        # 模拟处理时间
+        await asyncio.sleep(0.5)
+        # yield 增量结果
+        yield doc
+
+
+async def main():
+    agent = Agent(
+        name="Research Assistant",
+        instructions="你帮助用户搜索信息。",
+        tools=[search_documents],
+    )
+
+    result = Runner.run_streamed(
+        agent,
+        input="搜索关于人工智能的信息",
+    )
+
+    async for event in result.stream_events():
+        # 处理工具流式事件
+        if event.type == "tool_output_stream_event":
+            print(f"[{event.tool_name}] {event.delta}", end="", flush=True)
+        # 处理最终工具输出
+        elif event.type == "run_item_stream_event" and event.name == "tool_output":
+            print(f"\n✓ 工具完成\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+关于流式工具的要点：
+
+- 流式工具必须返回 `AsyncIterator[str]`（一个 yield 字符串的异步生成器）
+- 每个 yield 的块都会作为 `ToolOutputStreamEvent` 发出
+- 所有块会自动累积并作为最终工具输出发送给 LLM
+- 非流式工具可以与流式工具正常共存
+- 在非流式模式（`Runner.run()`）中，流式工具会在返回前自动收集所有块

examples/tools/streaming_tool_example.py

@@ -0,0 +1,97 @@
+"""
+Example of using streaming tools with the Agents SDK.
+
+This example demonstrates how to create a tool that yields incremental output,
+allowing you to stream tool execution results to the user in real-time.
+"""
+
+import asyncio
+from collections.abc import AsyncIterator
+
+from agents import Agent, Runner, ToolOutputStreamEvent, function_tool
+
+
+@function_tool
+async def search_documents(query: str) -> AsyncIterator[str]:
+    """Search through documents and stream results as they are found.
+
+    Args:
+        query: The search query.
+
+    Yields:
+        Incremental search results.
+    """
+    # Simulate searching through multiple documents
+    documents = [
+        f"Document 1 contains information about {query}...\n",
+        f"Document 2 has additional details on {query}...\n",
+        f"Document 3 provides analysis of {query}...\n",
+    ]
+
+    for doc in documents:
+        # Simulate processing time
+        await asyncio.sleep(0.5)
+        # Yield incremental results
+        yield doc
+
+
+@function_tool
+async def generate_report(topic: str) -> AsyncIterator[str]:
+    """Generate a report on a topic, streaming the output as it's generated.
+
+    Args:
+        topic: The topic to generate a report on.
+
+    Yields:
+        Incremental report content.
+    """
+    sections = [
+        f"# Report on {topic}\n\n",
+        f"## Introduction\n\nThis report covers {topic} in detail.\n\n",
+        f"## Analysis\n\nOur analysis of {topic} shows several key points...\n\n",
+        f"## Conclusion\n\nIn summary, {topic} is an important topic.\n\n",
+    ]
+
+    for section in sections:
+        await asyncio.sleep(0.3)
+        yield section
+
+
+async def main():
+    # Create an agent with streaming tools
+    agent = Agent(
+        name="Research Assistant",
+        instructions="You are a helpful research assistant that can search documents and generate reports.",
+        tools=[search_documents, generate_report],
+    )
+
+    # Run the agent in streaming mode
+    result = Runner.run_streamed(
+        agent,
+        input="Search for information about artificial intelligence and generate a brief report.",
+    )
+
+    print("Streaming agent output:\n")
+
+    # Stream events and display tool outputs in real-time
+    async for event in result.stream_events():
+        # Handle tool streaming events
+        if event.type == "tool_output_stream_event":
+            assert isinstance(event, ToolOutputStreamEvent)
+            print(f"[{event.tool_name}] {event.delta}", end="", flush=True)
+
+        # Handle run item events (final outputs)
+        elif event.type == "run_item_stream_event":
+            if event.name == "tool_output":
+                print(f"\n✓ Tool '{event.item.agent.name}' completed\n")
+            elif event.name == "message_output_created":
+                print(f"\n[Agent Response]: {event.item}\n")
+
+    # Get final result
+    print("\n" + "=" * 60)
+    print("Final output:", result.final_output)
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/agents/__init__.py

@@ -65,6 +65,7 @@
     RawResponsesStreamEvent,
     RunItemStreamEvent,
     StreamEvent,
+    ToolOutputStreamEvent,
 )
 from .tool import (
     CodeInterpreterTool,
@@ -263,6 +264,7 @@ def enable_verbose_stdout_logging():
     "RawResponsesStreamEvent",
     "RunItemStreamEvent",
     "AgentUpdatedStreamEvent",
+    "ToolOutputStreamEvent",
     "StreamEvent",
     "FunctionTool",
     "FunctionToolResult",