Events & Streaming

The Event Stream

In the previous examples, we used .collect() to get the final result. When you call a Runnable, it doesn’t just return the answer - it returns a stream of events that tell you what’s happening step by step (an async generator). The .collect() method waits for all events and gives you just the final answer. You can iterate through the async generator to process events in real-time:

async for event in add_tool(a=5, b=3):
    print(event)

# Output:
# StartEvent(run_id="068c4458382e79bb80006dc019ac3039", ...)
# OutputEvent(run_id="068c4458382e79bb80006dc019ac3039", output=8, ...)

This enables you to:

Monitor execution progress in real-time
Handle streaming responses from LLMs
Debug execution flow
Build reactive user interfaces

Event Logging

In reality, you don’t need to manually print events. By default, events are logged to standard output by the framework. You can control logging behavior with these environment variables:

TIMBAL_LOG_EVENTS: Which events to log (default: "START,OUTPUT")
TIMBAL_LOG_FORMAT: Log format - "dev" for human-readable or "json" for structured (default)
TIMBAL_LOG_LEVEL: Standard log level (default: "INFO")

Event Types

Events are the communication mechanism that Runnables use to stream information throughout their execution lifecycle. Every Runnable execution produces a sequence of events that can be consumed in real-time or collected for later processing. Events are designed to be lightweight. For comprehensive execution data, see Traces. Every execution produces at least a Start event (when it begins) and an Output event (when it finishes). LLMs and streaming operations also produce Delta events for intermediate results. Runs that pause for human input emit Approval or Interaction events. The following examples show what these events look like for an LLM interaction:

Start Event

Signals the beginning of an execution.

from timbal.types.events import StartEvent

start_event = StartEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    parent_run_id=None, 
    path="agent.llm",
    call_id="068c4458383678be800031537a8df42e",
    parent_call_id="068c4458382e708f8000cc0f9b19d970",
)

These fields are present in all events to identify their source - the framework handles this automatically. We’ll explore this in greater detail in advanced sections.

Delta Events

Streaming content is emitted as typed Delta Events — structured, semantic updates about text, tool calls, thinking, and other content types.

Delta Event Structure

Each DeltaEvent contains an item property with a typed delta item. All delta items inherit from DeltaItem and have an id and type field:

from timbal.types.events import DeltaEvent
from timbal.types.events.delta import Text, TextDelta, ToolUse

# Text block start (complete text content)
text_event = DeltaEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    path="agent.llm",
    call_id="068c4458383678be800031537a8df42e",
    item=Text(id="text_0", text="Hello")
)

# Text streaming delta
text_delta_event = DeltaEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    path="agent.llm",
    call_id="068c4458383678be800031537a8df42e",
    item=TextDelta(id="text_0", text_delta=" world")
)

# Tool call start
tool_event = DeltaEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    path="agent.llm",
    call_id="068c4458383678be800031537a8df42e",
    item=ToolUse(id="call_123", name="search", input="")
)

Delta Item Types

DeltaEvents use a pattern of paired types: a “start” type for the beginning of a content block, and a “delta” type for streaming incremental updates. All items have an id field to correlate deltas with their parent block. Text - Start of a text content block:

Text(
    id="text_0",
    text="Hello"  # Initial text content
)

TextDelta - Streaming text increments:

TextDelta(
    id="text_0",
    text_delta=" world"  # Incremental text
)

ToolUse - Start of a tool call:

ToolUse(
    id="call_abc123",
    name="search_web",
    input="",  # Initially empty
    is_server_tool_use=False
)

ToolUseDelta - Streaming tool input parameters:

ToolUseDelta(
    id="call_abc123",
    input_delta='{"query'  # Partial JSON
)

LLMs stream tool input parameters incrementally. Multiple ToolUseDelta events will be emitted for a single tool call. You need to accumulate these deltas and parse the complete JSON afterwards. Thinking - Start of LLM reasoning block:

Thinking(
    id="thinking_0",
    thinking="Let me analyze this problem..."
)

ThinkingDelta - Streaming thinking increments:

ThinkingDelta(
    id="thinking_0",
    thinking_delta="First, I'll consider..."
)

Custom - Arbitrary custom content:

Custom(
    id="custom_0",
    data={"type": "image_progress", "percent": 45}
)

Use Custom for streaming content that doesn’t fit standard types (e.g., multimodal content, provider-specific features, experimental data). This allows custom collectors and tools to emit arbitrary typed data while participating in the delta event system. ContentBlockStop - Signals end of a content block:

ContentBlockStop(id="text_0")

This event signals that a content block (text, tool use, thinking) has finished streaming. Use it to finalize processing of accumulated deltas.

Using Delta Events

from timbal.types.events import DeltaEvent
from timbal.types.events.delta import (
    Text, TextDelta, ToolUse, ToolUseDelta,
    Thinking, ThinkingDelta, Custom, ContentBlockStop
)

async for event in agent(prompt="Hello"):
    if isinstance(event, DeltaEvent):
        item = event.item
        # Type-safe handling of different content types
        if isinstance(item, Text):
            print(item.text, end="", flush=True)
        elif isinstance(item, TextDelta):
            print(item.text_delta, end="", flush=True)
        elif isinstance(item, ToolUse):
            print(f"\n[Calling {item.name}]")
        elif isinstance(item, ToolUseDelta):
            # Accumulate input_delta for later JSON parsing
            pass
        elif isinstance(item, Thinking):
            print(f"\n[Thinking: {item.thinking}]")
        elif isinstance(item, ThinkingDelta):
            print(item.thinking_delta, end="", flush=True)
        elif isinstance(item, Custom):
            print(f"\n[Custom: {item.data}]")
        elif isinstance(item, ContentBlockStop):
            print(f"\n[Block {item.id} complete]")

Benefits of Delta Events

Type Safety: Each delta item has specific fields and types
Semantic Information: Know exactly what type of content is streaming
Better Observability: Track tool calls, thinking, and text separately
Block Lifecycle: Start and stop events for each content block
UI Flexibility: Render different content types with appropriate components
Structured Logging: Monitor different content types independently

Approval and Interaction Events

When a run pauses for human input, the framework emits structured events before the final OutputEvent:

ApprovalEvent — a requires_approval gate fired before a runnable executed. Resume with resume={approval_id: True} (or an ApprovalResolution).
InteractionEvent — a handler called suspend() to ask the user for input. Resume with resume={suspension_id: value}.

See Human in the Loop for the full reference.

Output Event

Contains the final result and signals completion. Contains either the final output result (if successful) or error information (if something went wrong).

from timbal.types.events import OutputEvent

output_event = OutputEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    parent_run_id=None,
    path="agent.llm",
    call_id="068c4458383678be800031537a8df42e",
    parent_call_id="068c4458382e708f8000cc0f9b19d970",
    output={
        "role": "assistant",
        "content": [{"type": "text", "text": "Hello! How can I assist you today?"}]
    },
    error=None
)

Handling Errors

When a Runnable encounters an error during execution, it won’t raise an exception. Instead, it will always return an OutputEvent with error information. This ensures the event stream continues and you can handle errors gracefully:

try:
    result = await runnable(**kwargs).collect()
except Exception as e:
    # This won't happen - errors are in the OutputEvent
    pass

# Result will always be an instance of OutputEvent
if result.error:
    print(f"Error: {result.error['message']}")

Errors will always have this structure:

error = {
    "type": "ValueError",
    "message": "Invalid input provided",
    "traceback": "Traceback (most recent call last):\\n..."
}

​The Event Stream

​Event Logging

​Event Types

​Start Event

​Delta Events

​Delta Event Structure

​Delta Item Types

​Using Delta Events

​Benefits of Delta Events

​Approval and Interaction Events

​Output Event

​Handling Errors

The Event Stream

Event Logging

Event Types

Start Event

Delta Events

Delta Event Structure

Delta Item Types

Using Delta Events

Benefits of Delta Events

Approval and Interaction Events

Output Event

Handling Errors