How to stream events from a LlamaIndex agent

An agent can start a tool call, return its result, and produce answer text during one run. Streaming those transitions lets an application update a progress view or capture tool activity without waiting for the final response.

Calling FunctionAgent.run() without await returns a workflow handler. The handler's stream_events() iterator exposes ToolCall, ToolCallResult, and AgentStream objects while awaiting the same handler afterward returns the completed response.

The runnable smoke test uses MockFunctionCallingLLM and default tool arguments, so it needs no provider credentials and produces repeatable output. A provider-backed function-calling LLM can replace the mock in an application without changing the handler or event loop.

1. Install LlamaIndex core in the active Python environment.

   $ python3 -m pip install --upgrade llama-index-core

   A project virtual environment keeps this package separate from a shared system Python installation.
   Related: How to install LlamaIndex with pip

2. Create stream-agent-events.py with the event imports and arithmetic tool.

   $ cat > stream-agent-events.py <<'PY'
   import asyncio
    
   from llama_index.core.agent.workflow import (
       AgentStream,
       FunctionAgent,
       ToolCall,
       ToolCallResult,
   )
   from llama_index.core.llms.mock import MockFunctionCallingLLM
    
    
   def add_numbers(a: int = 6, b: int = 7) -> int:
       """Add two integers."""
       return a + b
   PY

   The mock calls registered tools with their default arguments, which makes 6 and 7 the deterministic inputs for this smoke test.

3. Append the agent setup and workflow handler to stream-agent-events.py.

   $ cat >> stream-agent-events.py <<'PY'
    
    
   async def main() -> None:
       agent = FunctionAgent(
           tools=[add_numbers],
           llm=MockFunctionCallingLLM(is_chat_model=True),
           system_prompt="Use tools for arithmetic.",
       )
       handler = agent.run(user_msg="Add 6 and 7.")
   PY

   The returned handler remains unawaited until after event iteration; awaiting agent.run() immediately would return only the completed response.

4. Append the event loop and final-response check to stream-agent-events.py.

   $ cat >> stream-agent-events.py <<'PY'
    
       async for event in handler.stream_events():
           if isinstance(event, ToolCall):
               print(f"Tool call: {event.tool_name} {event.tool_kwargs}")
           elif isinstance(event, ToolCallResult):
               print(f"Tool result: {event.tool_name} -> {event.tool_output.content}")
           elif isinstance(event, AgentStream) and event.delta:
               print(f"Agent delta: {event.delta}")
    
       response = await handler
       print(f"Final response: {response}")
    
    
   asyncio.run(main())
   PY

   ToolCall identifies the requested function and arguments, ToolCallResult exposes the returned value, and AgentStream carries streamed assistant deltas.

5. Inspect the completed stream-agent-events.py source.

   stream-agent-events.py

   import asyncio
    
   from llama_index.core.agent.workflow import (
       AgentStream,
       FunctionAgent,
       ToolCall,
       ToolCallResult,
   )
   from llama_index.core.llms.mock import MockFunctionCallingLLM
    
    
   def add_numbers(a: int = 6, b: int = 7) -> int:
       """Add two integers."""
       return a + b
    
    
   async def main() -> None:
       agent = FunctionAgent(
           tools=[add_numbers],
           llm=MockFunctionCallingLLM(is_chat_model=True),
           system_prompt="Use tools for arithmetic.",
       )
       handler = agent.run(user_msg="Add 6 and 7.")
    
       async for event in handler.stream_events():
           if isinstance(event, ToolCall):
               print(f"Tool call: {event.tool_name} {event.tool_kwargs}")
           elif isinstance(event, ToolCallResult):
               print(f"Tool result: {event.tool_name} -> {event.tool_output.content}")
           elif isinstance(event, AgentStream) and event.delta:
               print(f"Agent delta: {event.delta}")
    
       response = await handler
       print(f"Final response: {response}")
    
    
   asyncio.run(main())

6. Run stream-agent-events.py to confirm that tool events and an agent delta arrive before the final response.

   $ python3 stream-agent-events.py
   Tool call: add_numbers {'a': 6, 'b': 7}
   Tool result: add_numbers -> 13
   Agent delta: Tool calls complete.
   Final response: Tool calls complete.