how to send events/progress of a tool call through ag-ui

[happyxgang]

when a long running tool is executing, we often want to send some realtime events to the user. how can we achive this with ag-ui, custom event? is there anything demo for this?

[ltanme]

AG-UI Long-Running Tool Progress Event Guide

Overview

In AG-UI, when executing long-running tools, you can send real-time progress updates and events to the user through CustomEvent. This is one of the core capabilities of the AG-UI protocol.

CustomEvent Type Definition

In the TypeScript SDK, the CustomEvent interface is defined as follows:

export interface CustomEvent {
  type: EventType.CUSTOM;
  name: string;           // Event name
  value: any;             // Event data (any JSON-serializable value)
  timestamp?: number;     // Optional timestamp
  rawEvent?: any;         // Optional raw event payload
}

How to Send Custom Events

Python Example

In Python, you can yield a CustomEvent from an async generator function:

from ag_ui.core import CustomEvent, EventType
import asyncio

async def long_running_tool_with_progress():
    """Example of a long-running tool with progress updates"""
    
    # Send progress start event
    yield CustomEvent(
        type=EventType.CUSTOM,
        name="progress_start",
        value={
            "message": "Starting process...",
            "total_steps": 5
        }
    )
    
    # Simulate work and send progress updates
    for i in range(1, 6):
        await asyncio.sleep(1)
        yield CustomEvent(
            type=EventType.CUSTOM,
            name="progress_update",
            value={
                "current_step": i,
                "total_steps": 5,
                "percentage": (i / 5) * 100,
                "message": f"Processing step {i}/5..."
            }
        )
    
    # Send completion event
    yield CustomEvent(
        type=EventType.CUSTOM,
        name="progress_complete",
        value={
            "message": "Processing complete!",
            "result": "success"
        }
    )

Complete Endpoint Example (Python + FastAPI)

import uuid
import asyncio
from fastapi import Request
from fastapi.responses import StreamingResponse
from ag_ui.core import (
    RunAgentInput,
    EventType,
    RunStartedEvent,
    RunFinishedEvent,
    CustomEvent,
    ToolCallStartEvent,
    ToolCallArgsEvent,
    ToolCallEndEvent
)
from ag_ui.encoder import EventEncoder

async def agent_endpoint(input_data: RunAgentInput, request: Request):
    """Agent endpoint with progress updates"""
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():
        # Run start event
        yield encoder.encode(
            RunStartedEvent(
                type=EventType.RUN_STARTED,
                thread_id=input_data.thread_id,
                run_id=input_data.run_id
            )
        )

        # Start tool call
        tool_call_id = str(uuid.uuid4())
        yield encoder.encode(
            ToolCallStartEvent(
                type=EventType.TOOL_CALL_START,
                tool_call_id=tool_call_id,
                tool_call_name="process_large_file"
            )
        )

        # Send tool arguments
        yield encoder.encode(
            ToolCallArgsEvent(
                type=EventType.TOOL_CALL_ARGS,
                tool_call_id=tool_call_id,
                delta='{"file": "large_data.csv"}'
            )
        )

        # Progress events during processing
        total_chunks = 10
        for chunk in range(1, total_chunks + 1):
            yield encoder.encode(
                CustomEvent(
                    type=EventType.CUSTOM,
                    name="file_processing_progress",
                    value={
                        "chunk": chunk,
                        "total": total_chunks,
                        "percentage": (chunk / total_chunks) * 100,
                        "status": f"Processing chunk {chunk}/{total_chunks}"
                    }
                )
            )
            await asyncio.sleep(0.5)

        # End tool call
        yield encoder.encode(
            ToolCallEndEvent(
                type=EventType.TOOL_CALL_END,
                tool_call_id=tool_call_id
            )
        )

        # Run finished event
        yield encoder.encode(
            RunFinishedEvent(
                type=EventType.RUN_FINISHED,
                thread_id=input_data.thread_id,
                run_id=input_data.run_id
            )
        )

    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )

TypeScript Client Handling of Custom Events

On the client side, use AgentSubscriber to listen for and handle custom events:

import { AgentSubscriber, CustomEvent } from '@ag-ui/client';

const subscriber: AgentSubscriber = {
  onCustomEvent: async ({ event, messages, state }) => {
    const customEvent = event as CustomEvent;
    
    switch (customEvent.name) {
      case 'progress_update':
        console.log('Progress update:', customEvent.value);
        updateProgressBar(customEvent.value.percentage);
        break;
        
      case 'file_processing_progress':
        console.log('File progress:', customEvent.value);
        showProcessingStatus(customEvent.value.status);
        break;
        
      case 'progress_complete':
        console.log('Complete:', customEvent.value);
        showCompletionMessage(customEvent.value.message);
        break;
        
      default:
        console.log('Unknown custom event:', customEvent.name, customEvent.value);
    }
    
    return {
      state: {
        ...state,
        lastProgress: customEvent.value
      }
    };
  },
  
  onEvent: async ({ event }) => {
    console.log('Received event:', event.type);
  }
};

const agent = new HttpAgent({
  url: 'http://localhost:8000/agent',
  subscribers: [subscriber]
});

agent.runAgent({
  threadId: 'thread-123',
  runId: 'run-456',
  messages: [],
  state: {},
  tools: [],
  context: []
});

Common Use Cases

1. File Upload/Download Progress

yield CustomEvent(
    type=EventType.CUSTOM,
    name="upload_progress",
    value={
        "bytes_uploaded": 1024000,
        "total_bytes": 5120000,
        "percentage": 20,
        "speed": "1.5 MB/s"
    }
)

2. Data Processing Progress

yield CustomEvent(
    type=EventType.CUSTOM,
    name="data_processing",
    value={
        "records_processed": 500,
        "total_records": 1000,
        "current_operation": "Data Cleaning",
        "estimated_time_remaining": "30s"
    }
)

3. Multi-Step Task Progress

yield CustomEvent(
    type=EventType.CUSTOM,
    name="task_step_complete",
    value={
        "step_name": "Data Validation",
        "step_number": 2,
        "total_steps": 5,
        "step_result": "Passed"
    }
)

4. Real-Time Logs/Debug Messages

yield CustomEvent(
    type=EventType.CUSTOM,
    name="debug_log",
    value={
        "level": "info",
        "message": "Connecting to database...",
        "timestamp": time.time()
    }
)

Predictive State Update Example

AG-UI also supports predictive state updates, allowing the UI to anticipate and display changes before the tool finishes:

# Send predictive state event
yield CustomEvent(
    type=EventType.CUSTOM,
    name="PredictState",
    value=[
        {
            "state_key": "document",
            "tool": "write_document_local",
            "tool_argument": "document"
        }
    ]
)

# Then stream tool arguments
yield ToolCallArgsEvent(
    type=EventType.TOOL_CALL_ARGS,
    tool_call_id=tool_call_id,
    delta='{"document":"'
)

for chunk in content_chunks:
    yield ToolCallArgsEvent(
        type=EventType.TOOL_CALL_ARGS,
        tool_call_id=tool_call_id,
        delta=chunk + " "
    )
    await asyncio.sleep(0.2)

Reference Resources

• Core Event Definitions: sdks/typescript/packages/core/src/events.ts

• Client Subscriber Interface: sdks/typescript/packages/client/src/agent/subscriber.ts

• Python Examples: integrations/server-starter-all-features/python/examples/example_server/

  • predictive_state_updates.py — Predictive state update example
  • backend_tool_rendering.py — Backend tool rendering example
  • tool_based_generative_ui.py — Tool-based generative UI example

Best Practices

1. Use meaningful event names — e.g., "progress_update", "file_processing_progress".
2. Keep consistent value structures — same structure for same event type.
3. Include sufficient context — ensure the front-end can render progress meaningfully.
4. Reasonable update frequency — avoid sending updates too frequently.
5. Error reporting — use custom events for warnings and error info.
6. Timestamps — include timestamps for debugging and logging when needed.

Summary

AG-UI’s CustomEvent mechanism enables flexible progress reporting for long-running tools.
By yielding CustomEvent on the server and handling them via onCustomEvent on the client, you can build responsive, user-friendly AI application experiences.

[NathanTarbert]

Thanks @ltanme for jumping in here!

[NathanTarbert]

Also, where did you find this guide?

[ltanme]

Also, where did you find this guide?

https://github.com/ag-ui-protocol/ag-ui/tree/main/sdks/typescript