How to stream events from a LlamaIndex workflow

Long-running asynchronous jobs can look stalled when callers receive nothing until the final value is ready. LlamaIndex Workflows can publish typed progress events during a run while continuing toward its StopEvent result.

Each run returns a WorkflowHandler that represents both sides of this interaction. Workflow steps send ProgressEvent objects through Context.write_event_to_stream(), and the caller consumes them through handler.stream_events() without changing the typed events passed between steps.

The standalone llama-index-workflows package is sufficient for the local smoke test and does not require an LLM or API key. A handler stream can be consumed once, so applications with several listeners should read it through one consumer and fan out those events separately.

1. Define the event contracts at the top of stream-workflow-events.py.

   stream-workflow-events.py

   import asyncio
    
   from workflows import Context, Workflow, step
   from workflows.events import Event, StartEvent, StopEvent
    
    
   class ProgressEvent(Event):
       message: str
    
    
   class DraftEvent(Event):
       text: str

   ProgressEvent is visible to the caller, while DraftEvent remains the typed handoff between workflow steps.
   Related: How to install LlamaIndex with pip

2. Append the workflow steps after the DraftEvent class.

   class DraftWorkflow(Workflow):
       @step
       async def prepare(self, ctx: Context, ev: StartEvent) -> DraftEvent:
           ctx.write_event_to_stream(
               ProgressEvent(message=f"received topic: {ev.topic}")
           )
           return DraftEvent(text=ev.topic.upper())
    
       @step
       async def finish(self, ctx: Context, ev: DraftEvent) -> StopEvent:
           ctx.write_event_to_stream(
               ProgressEvent(message=f"prepared text: {ev.text}")
           )
           return StopEvent(result=f"final: {ev.text}")

   write_event_to_stream() exposes progress without replacing the DraftEvent returned to the next step.

3. Append the stream consumer after the DraftWorkflow class.

   async def main() -> None:
       workflow = DraftWorkflow(timeout=10)
       handler = workflow.run(topic="stream events")
    
       async for event in handler.stream_events():
           if isinstance(event, ProgressEvent):
               print(f"{type(event).__name__}: {event.message}")
    
       result = await handler
       print(f"Final result: {result}")
    
    
   if __name__ == "__main__":
       asyncio.run(main())

   The handler returned by run() supplies both streamed events and the final result from one run.

4. Run the completed workflow script to confirm both progress events precede the final result.

   $ python3 stream-workflow-events.py
   ProgressEvent: received topic: stream events
   ProgressEvent: prepared text: STREAM EVENTS
   Final result: final: STREAM EVENTS