# Streaming Tool Results in Strands Agents

## The Problem

When implementing custom `AgentTool` classes that stream partial results, the intermediate results don't reach the client when using `agent.stream_async()`. Only the final result appears.

## Why It Happens

The issue occurs because of how the `stream_async()` method filters events:

```python
# In agent.py
async def stream_async(self, prompt, **kwargs):
    # ...
    events = self._run_loop(message, invocation_state=kwargs)
    async for event in events:
        if "callback" in event:  # <-- CRITICAL FILTERING POINT
            callback_handler(**event["callback"])
            yield event["callback"]  # Only these reach client
```

Standard tool results don't have a `callback` key, so they never reach the client.

## The Solution

Streaming tool events require specific formatting:

1. **Intermediate results** must be wrapped with `{"callback": {...}}` to reach the client
2. **Final result** must remain in standard `ToolResult` format (without wrapper)

## Example Implementation

```python
async def stream(self, tool_use: ToolUse, invocation_state: dict, **kwargs) -> ToolGenerator:
    tool_id = tool_use.get("toolUseId", "unknown")
    
    # Intermediate result - WITH callback wrapper
    yield {"callback": {
        "current_tool_use": tool_use,
        "event": {
            "tool_stream": {
                "progress": 0.5,
                "message": "Processing...",
                # Any custom data here
            }
        }
    }}
    
    await asyncio.sleep(1)  # Processing time
    
    # Final result - WITHOUT callback wrapper
    yield {
        "toolUseId": tool_id,
        "status": "success",
        "content": [{"text": "Processing complete."}]
    }
```

## Consuming Streaming Results

```python
async for event in agent.stream_async("Use the streaming tool"):
    # Check for streaming tool events
    if "event" in event and "tool_stream" in event["event"]:
        tool_stream = event["event"]["tool_stream"]
        print(f"Progress: {tool_stream['progress'] * 100}%")
        
    # Check for regular text content
    elif "delta" in event:
        print(event["delta"]["text"], end="", flush=True)
```

## Full Example Class

```python
from strands.types.tools import AgentTool, ToolSpec, ToolUse, ToolGenerator
import asyncio

class StreamingTool(AgentTool):
    @property
    def tool_name(self) -> str:
        return "streaming_tool"
    
    @property
    def tool_spec(self) -> ToolSpec:
        return {
            "name": "streaming_tool",
            "description": "A tool that streams results",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Input query"
                        }
                    }
                }
            }
        }
    
    @property
    def tool_type(self) -> str:
        return "custom"
    
    async def stream(self, tool_use: ToolUse, invocation_state: dict, **kwargs) -> ToolGenerator:
        tool_id = tool_use.get("toolUseId", "unknown")
        
        # First update - requires callback wrapper
        yield {"callback": {
            "current_tool_use": tool_use,
            "event": {
                "tool_stream": {
                    "progress": 0.33,
                    "message": "First step complete"
                }
            }
        }}
        
        await asyncio.sleep(1)
        
        # Second update - requires callback wrapper
        yield {"callback": {
            "current_tool_use": tool_use,
            "event": {
                "tool_stream": {
                    "progress": 0.66,
                    "message": "Second step complete"
                }
            }
        }}
        
        await asyncio.sleep(1)
        
        # Final result - standard format without wrapper
        yield {
            "toolUseId": tool_id,
            "status": "success",
            "content": [{"text": "All processing complete"}]
        }
```

## Summary

1. Strands Agents filters events in `stream_async()`, only yielding those with a "callback" key
2. Tool events must be wrapped in `{"callback": {...}}` to reach the client
3. Inside the callback wrapper, you have complete flexibility for your custom data
4. The final result must remain in standard ToolResult format without the wrapper
