Skip to main content

​
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 Chunk or Delta events for intermediate results. 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.

​
Chunk Event

Contains streaming or intermediate results during execution. The chunk property contains each piece of the response as it’s generated during streaming. 
from timbal.types.events import ChunkEvent

chunk_event = ChunkEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    parent_run_id=None,
    path="agent.llm", 
    call_id="068c4458383678be800031537a8df42e",
    parent_call_id="068c4458382e708f8000cc0f9b19d970",
    chunk="Hello"
)
Deprecation Notice: ChunkEvents are simple and untyped, and will be deprecated in a future release in favor of Delta Events. Delta Events provide structured, typed streaming with semantic information about different content types (text, tool calls, thinking) and better observability. We recommend migrating to Delta Events for new projects.

​
Delta Events

For advanced use cases requiring fine-grained control over streaming content, Timbal provides Delta Events. Unlike simple ChunkEvents, DeltaEvents provide typed, structured information about different types of streaming content.
Delta events are opt-in via the environment variable:
export TIMBAL_DELTA_EVENTS=true

​
Delta Event Structure

from timbal.types.events import DeltaEvent
from timbal.types.events.delta import TextItem, ToolItem, ToolInputItem

# Text streaming
text_delta = DeltaEvent(
    run_id="068c4458382e79bb80006dc019ac3039",
    path="agent.llm",
    call_id="068c4458383678be800031537a8df42e",
    item=TextItem(delta="Hello")  # Typed as text
)

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

​
Delta Item Types

DeltaEvents contain one of several typed items: TextItem - Streaming text output: 
TextItem(delta="Hello world")
ToolItem - Start of a tool call: 
ToolItem(
    id="call_abc123",
    name="search_web",
    input="",  # Initially empty
    is_server_tool_use=False
)
ToolInputItem - Streaming tool parameters: 
ToolInputItem(
    id="call_abc123",
    delta='{"query'  # Partial JSON
)
LLMs stream tool input parameters incrementally. Multiple ToolInputItem events will be emitted for a single tool call. You need to accumulate these deltas and parse the complete JSON afterwards. ToolResultItem - Tool execution result: 
ToolResultItem(
    id="call_abc123",
    result={"temperature": 72}
)
This event is only emitted for is_server_tool_use tools (e.g., web search tools run by OpenAI/Anthropic). For local tools that Timbal executes, you’ll see the normal runnable event sequence (StartEvent, OutputEvent, etc.) instead. ThinkingItem - LLM reasoning process: 
ThinkingItem(delta="Let me analyze...")
CustomItem - Arbitrary custom content: 
CustomItem(
    data={"type": "image_progress", "percent": 45}
)
Use CustomItem 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.

​
Using Delta Events

from timbal.types.events import DeltaEvent
from timbal.types.events.delta import TextItem, ToolItem, ThinkingItem, CustomItem

async for event in agent(input="Hello"):
    if isinstance(event, DeltaEvent):
        # Type-safe handling of different content types
        if isinstance(event.item, TextItem):
            print(event.item.delta, end="", flush=True)
        elif isinstance(event.item, ToolItem):
            print(f"\n[Calling {event.item.name}]")
        elif isinstance(event.item, ThinkingItem):
            print(f"\n[Thinking: {event.item.delta}]")
        elif isinstance(event.item, CustomItem):
            print(f"\n[Custom: {event.item.data}]")

​
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
  • UI Flexibility: Render different content types with appropriate components
  • Structured Logging: Monitor different content types independently

​
Backward Compatibility

Delta Events are opt-in.When TIMBAL_DELTA_EVENTS is disabled (default):
  • Text content is emitted as ChunkEvent for backward compatibility
  • Other delta types (tool calls, thinking) are filtered out
  • Existing code continues to work unchanged
When enabled:
  • All content is emitted as structured DeltaEvent instances
  • No ChunkEvent instances for LLM text output
  • Full observability into all streaming content types

​
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..."
}