Live Todo Streaming

See the agent's todo checklist tick off in real time

Subscribe to TODO_UPDATED to render a live checklist that updates as the agent adds and ticks off items.

The agent adds and updates todos with the built-in todo tool; each change emits a TODO_UPDATED event carrying the full ordered list, so any subscriber renders the current state without tracking deltas.

Quick Start

1. Enable the todo tool

   from praisonaiagents import Agent
   
   agent = Agent(
       name="Planner",
       instructions="Break tasks into todos, then execute them one by one.",
       tools=["todo_add", "todo_list", "todo_update"],
   )
   agent.start("Plan a blog launch — research, draft, publish.")

2. Subscribe to live updates

   from praisonaiagents import Agent
   from praisonaiagents.streaming import StreamEvent, StreamEventType
   
   MARK = {"pending": "[ ]", "in_progress": "[~]", "completed": "[x]", "cancelled": "[-]"}
   
   def on_event(event: StreamEvent):
       if event.type == StreamEventType.TODO_UPDATED:
           for t in event.metadata["todos"]:
               print(f"{MARK[t['status']]} {t['task']}")
   
   agent = Agent(
       name="Planner",
       instructions="Break the task into todos. Mark exactly one as in_progress at a time; complete it before moving on.",
       tools=["todo_add", "todo_list", "todo_update"],
   )
   agent.stream_emitter.add_callback(on_event)
   agent.start("Plan a blog launch — research, draft, publish.")

How It Works

Every todo_add and todo_update emits a TODO_UPDATED event with the complete, ordered list.

The event carries the full ordered list, so subscribers never track deltas — each event is a complete snapshot.

What’s in the event

The TODO_UPDATED event carries the current list in metadata.

Each todo dict looks like:

{
    "id": 1,
    "task": "research blog topic",
    "priority": "high",
    "category": "general",
    "status": "in_progress",
}

Status is one of pending | in_progress | completed | cancelled.

The single-in_progress rule

Setting a todo to in_progress demotes any other in_progress item back to pending, so exactly one item is active at a time.

todo_update(1, status="in_progress")   # todo 1 is now in_progress
todo_update(2, status="in_progress")   # todo 1 auto-demoted to pending; only todo 2 is in_progress

This keeps rendered checklists unambiguous — “what is the agent working on right now?” always has one answer. It matches the prompt policy recommended for the coding toolset.

Choosing a Renderer

Pick a renderer based on where the checklist appears.

Rich Live re-renders the same block on every event:

from rich.live import Live
from rich.text import Text
from praisonaiagents import Agent
from praisonaiagents.streaming import StreamEvent, StreamEventType

MARK = {"pending": "[ ]", "in_progress": "[~]", "completed": "[x]", "cancelled": "[-]"}
live = Live(Text(""), refresh_per_second=8)
live.start()

def on_event(event: StreamEvent):
    if event.type == StreamEventType.TODO_UPDATED:
        lines = [f"{MARK[t['status']]} {t['task']}" for t in event.metadata["todos"]]
        live.update(Text("\n".join(lines)))

agent = Agent(
    name="Planner",
    instructions="Break the task into todos. Keep exactly one in_progress.",
    tools=["todo_add", "todo_list", "todo_update"],
)
agent.stream_emitter.add_callback(on_event)
agent.start("Plan a blog launch — research, draft, publish.")
live.stop()

Best Practices

In AG-UI

Over the AG-UI wire, a TODO_UPDATED event reaches the client as a custom event named praisonai.todo_updated with value { content, tool_call, metadata }. See the AGUI API reference.

Related