How to Stream Tool Events

This guide covers how to stream real-time feedback to users when your agent executes tools. Tool streaming displays status messages like "Searching for documents..." that update to "Found 15 results" when complete.

Prerequisites

Before you begin, make sure you are familiar with How to Create Agents that Use Tools.

Step 1: Mark Tools as Streamable

Use the @streamable decorator to define display text templates for your tools:

from zav.agents_sdk import streamable

@streamable(
    running_text="Searching for {query}...",
    completed_text="Found {result_count} results for '{query}'"
)
async def search_documents(query: str) -> dict:
    """Search for documents matching the query."""
    results = await perform_search(query)
    return {"result_count": len(results), "documents": results}

Templates support {placeholder} interpolation from both input parameters and return values.

Alternatively, pass the streaming config explicitly when registering:

from zav.agents_sdk import ToolStreamingConfig

tools_registry.add(
    executable=search_documents,
    streaming_config=ToolStreamingConfig(
        running_text="Searching for {query}...",
        completed_text="Found {result_count} results"
    )
)

Step 2: Enable Streaming in Your Agent

Set stream_tool_events=True in the complete() call and use chat_response.to_chat_message() to convert responses:

from zav.agents_sdk import ChatAgentClassRegistry, ChatMessage, StreamableChatAgent
from zav.agents_sdk.adapters import ZAVChatCompletionClient

@ChatAgentClassRegistry.register()
class SearchAgent(StreamableChatAgent):
    agent_name = "search_agent"

    def __init__(self, client: ZAVChatCompletionClient):
        self.client = client
        self.tools_registry.add(search_documents)

    async def execute_streaming(self, conversation):
        response_stream = await self.client.complete(
            bot_setup_description="You are a helpful search assistant.",
            messages=conversation,
            tools=self.tools_registry,
            stream=True,
            execute_tools=True,
            stream_tool_events=True,
        )

        async for chat_response in response_stream:
            if chat_response.error:
                raise chat_response.error

            message = chat_response.to_chat_message()
            if message:
                yield message

The to_chat_message() method automatically includes tool_events as content_parts. Each tool event is a ContentPartTool object.

• name: Tool name
• status: "running", "completed", or "error"
• display_text: The interpolated template text
• params: Tool input parameters
• response: Tool output (when completed)

Step 3: Filter Data Sent to Frontend (Optional)

Tool responses may contain internal data or sensitive information not needed by the frontend. Use transform callbacks to filter what gets included:

from zav.agents_sdk import streamable, include_fields, exclude_fields, hide_response

# Only include specific fields in the response
@streamable(
    running_text="Searching...",
    completed_text="Found {result_count} results",
    response_transform=include_fields("result_count", "summary"),
)
async def search(query: str) -> dict:
    return {
        "result_count": 10,
        "summary": "Found documents about...",
        "internal_scores": [0.9, 0.8],  # Excluded from frontend
    }

# Exclude specific fields
@streamable(
    running_text="Fetching document...",
    response_transform=exclude_fields("raw_content", "metadata"),
)
async def fetch_document(doc_id: str) -> dict:
    ...

# Hide the entire response
@streamable(
    running_text="Processing...",
    response_transform=hide_response,
)
async def internal_operation(data: str) -> dict:
    ...

You can also write custom transform functions:

def redact_documents(response: dict | None) -> dict | None:
    if response is None:
        return None
    return {
        "result_count": response.get("result_count"),
        "documents": [
            {"id": doc["id"], "title": doc["title"]}
            for doc in response.get("documents", [])
        ]
    }

@streamable(
    running_text="Searching...",
    response_transform=redact_documents,
)
async def search(query: str) -> dict:
    ...

info

Transform callbacks only affect params and response in the ContentPartTool sent to the frontend. The full data remains available for LLM context and display_text interpolation.

MCP Tools

MCP tools discovered at runtime automatically get streaming enabled. To customize the display text, configure tool_streaming in your agent_setups.json:

agent_setups.json

{
  "agent_identifier": "mcp_agent",
  "agent_name": "mcp_agent",
  "agent_configuration": {
    "mcp_configuration": {
      "servers": [...],
      "tool_streaming": {
        "read_file": {
          "running_text": "Reading file {path}...",
          "completed_text": "Loaded file content"
        },
        "list_directory": {
          "running_text": "Listing directory {path}...",
          "completed_text": "Found {count} items"
        }
      }
    }
  }
}

Behavior:

• If tool_streaming is not set (default): all MCP tools get auto-generated streaming text
• If tool_streaming is set: only tools explicitly listed will stream