In this guide, you will learn how to integrate LlamaIndex agents with the AG-UI protocol. Also, we will cover how to integrate the AG-UI + LlamaIndex agents with CopilotKit in order to chat with the agent and stream its responses to the frontend.

Before we jump in, here is what we will cover:

• What is AG-UI protocol?
• Integrating LlamaIndex agents with AG-UI protocol
• Integrating a frontend to the AG-UI + LlamaIndex agent using CopilotKit

Here’s a preview of what we will be building:

What is AG-UI protocol?

The Agent User Interaction Protocol (AG-UI), developed by CopilotKit, is an open-source, lightweight, event-based protocol that facilitates rich, real-time interactions between the frontend and AI agents.

The AG-UI protocol enables event-driven communication, state management, tool usage, and streaming AI agent responses.

• Check out the AG-UI ⭐️ GitHub
  

To send information between the frontend and your AI agent, AG-UI uses events such as:

• Lifecycle events: These events mark the start or end of an agent’s task execution. Lifecycle events include RUN_STARTED and RUN_FINISHED events.
• Text message events: These events handle streaming agent responses to the frontend. Text message events include TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, and TEXT_MESSAGE_END events.
• Tool call events: These events manage the agent’s tool executions. Tool call events include TOOL_CALL_START, TOOL_CALL_ARGS, and TOOL_CALL_END events.
• State management events: These events keep the frontend and the AI agent state in sync. State management events include STATE_SNAPSHOT and STATE_DELTA events.

You can learn more about the AG-UI protocol and its architecture here on AG-UI docs.

Now that we have learned what the AG-UI protocol is, let us see how to integrate it with the LlamaIndex agent framework.

Let’s get started!

Want to learn more?

• Book a call and connect with our team
• Please tell us who you are --> what you're building, --> company size in the meeting description‍

Prerequisites

To fully understand this tutorial, you need to have a basic understanding of React or Next.js.

We'll also make use of the following:

• Python - a popular programming language for building AI agents with LangGraph; make sure it is installed on your computer.
• LlamaIndex - a simple, flexible framework for building knowledge assistants using LLMs connected to your enterprise data.
• OpenAI API Key - an API key to enable us to perform various tasks using the GPT models; for this tutorial, ensure you have access to the GPT-4 model.
• CopilotKit - an open-source copilot framework for building custom AI chatbots, in-app AI agents, and text areas.

Integrating LlamaIndex agents with the AG-UI protocol

To get started, clone the Open AG-UI Demo repository that consists of a Python-based backend (agent) and a Next.js frontend (frontend).

Next, navigate to the backend directory:

cd agent

Then install the dependencies using Poetry:

poetry install

After that, create a .env file with OpenAI API Key API key:

OPENAI_API_KEY=<<your-OpenAI-key-here>>

Then run the agent using the command below:

poetry run python main.py

To test the AG-UI + LlamaIndex integration, run the curl command below on https://reqbin.com/curl.

curl -X POST "http://localhost:8000/llamaindex-agent" \
  -H "Content-Type: application/json" \
  -d '{
    "thread_id": "test_thread_123",
    "run_id": "test_run_456",
    "messages": [
      {
        "id": "msg_1",
        "role": "user",
        "content": "Analyze AAPL stock with a $10000 investment from 2023-01-01"
      }
    ],
    "tools": [],
    "context": [],
    "forwarded_props": {},
    "state": {}
  }'

Let us now see how to integrate AG-UI protocol with the LlamaIndex agents framework.

Step 1: Create your LlamaIndex agent workflow

Before integrating AG-UI protocol with the LlamaIndex agent, create your LlamaIndex agent workflow, as shown in the agent/stock_analysis.py file.

from llama_index.core.workflow import (
    Context,
    Workflow,
    StartEvent,
    StopEvent,
    step,
)

class FuncationCallingAgent(Workflow):
    """
    Main workflow class for stock analysis and portfolio simulation.
    This agent processes user queries, fetches stock data, performs portfolio calculations,
    and generates investment insights using AI function calling.
    """
    def __init__(
        self,
        *args: Any,
        llm: FunctionCallingLLM | None = None,
        tools: List[BaseTool] | None = None,
        messages: List[Any],  # Conversation history
        emit_event: Callable,  # Function to emit UI state updates
        available_cash: int,  # Total cash available for investment
        investment_portfolio: List[Any],  # Current portfolio holdings
        **kwargs: Any,
    ) -> None:
        """
        Initialize the stock analysis workflow agent.

        Args:
            messages: Chat message history
            emit_event: Function to emit UI state changes
            available_cash: Available cash for investments
            investment_portfolio: Current portfolio data
        """
        super().__init__(*args, **kwargs)

        // ...

    @step
    async def chat_function(self, ctx: Context, ev: StartEvent) -> SimulationEvent:
        """
        First step: Analyze the user query and extract investment parameters.
        Uses AI function calling to parse ticker symbols, investment amounts, dates, etc.
        """

        // ...

    @step
    async def simulation_function(
        self, ctx: Context, ev: SimulationEvent
    ) -> CashAllocationEvent | StopEvent:
        """
        Second step: Fetch historical stock data for the specified tickers and time period.
        Downloads stock prices using Yahoo Finance API.
        """
        // ...

    @step
    async def cash_allocation_function(
        self, ctx: Context, ev: CashAllocationEvent
    ) -> InsightsEvent:
        """
        Third step: Perform portfolio simulation and cash allocation calculations.
        Simulates investment strategy (single shot or DCA) and calculates returns.
        """

        // ...

    @step
    async def insights_function(self, ctx: Context, ev: InsightsEvent) -> InsightsEvent:
        """
        Fourth step: Generate AI-powered insights about the investment portfolio.
        Uses GPT to create bull (positive) and bear (negative) insights for each stock.
        """

        // ...

Step 2: Create an endpoint with FastAPI

Once you have defined your LlamaIndex agent workflow, create a FastAPI endpoint and import the LlamaIndex agent workflow as shown in the agent/main.py file.

# Import necessary libraries for FastAPI web server and async operations
from fastapi import FastAPI
from fastapi.responses import StreamingResponse  # For streaming real-time responses to the client
import uuid  # For generating unique identifiers
from typing import Any  # Type hints for flexible data types
import os  # Environment variable access
import uvicorn  # ASGI server for running the FastAPI application
import asyncio  # Asynchronous programming support

# Import AG UI core components for event-driven agent communication
from ag_ui.core import (
    RunAgentInput,          # Input schema for agent runs
    StateSnapshotEvent,     # Event for sharing current state
    EventType,              # Enumeration of all event types
    RunStartedEvent,        # Event emitted when the agent run begins
    RunFinishedEvent,       # Event emitted when the agent run completes
    TextMessageStartEvent,  # Event emitted when text message begins
    TextMessageEndEvent,    # Event emitted when text message ends
    TextMessageContentEvent, # Event containing text message content chunks
    ToolCallStartEvent,     # Event emitted when tool call begins
    ToolCallEndEvent,       # Event emitted when tool call completes
    ToolCallArgsEvent,      # Event containing tool call arguments
    StateDeltaEvent,        # Event for incremental state updates
)

# Import encoder for converting events to streamable format
from ag_ui.encoder import EventEncoder

# Import CopilotKit state management
from copilotkit import CopilotKitState
from typing import List

# Import the stock analysis agent workflow
from stock_analysis import FunctionCallingAgent

# Create FastAPI application instance
app = FastAPI()

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """

    // ...

def main():
    """
    Server startup function.

    Initializes and runs the FastAPI server using Uvicorn ASGI server.
    Configures the server to listen on all interfaces with auto-reload for development.
    """
    # Step 1: Get port from environment variable or use default 8000
    port = int(os.getenv("PORT", "8000"))

    # Step 2: Start the Uvicorn server with configuration
    uvicorn.run(
        "main:app",         # Module:app reference for the FastAPI application
        host="0.0.0.0",     # Listen on all network interfaces
        port=port,          # Port number (from environment or default)
        reload=True,        # Enable auto-reload for development (restarts on file changes)
    )

# Application entry point
if __name__ == "__main__":
    # Run the server when the script is executed directly
    main()

Step 3: Define an event generator

After creating a FastAPI endpoint, define an event generator that produces a stream of AG-UI protocol events, initialize the event encoder and return the streaming response to the client or the frontend, as shown in the agent/main.py file.

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """
    try:
        # Step 1: Define the async event generator function
        # This function orchestrates the entire agent execution and event streaming
        async def event_generator():
            # Step 2: Initialize event handling infrastructure
            encoder = EventEncoder()  # Converts events to streamable format
            event_queue = asyncio.Queue()  # Queue for managing events between agent and API

            # Step 3: Define event emission function for agent callbacks
            def emit_event(event):
                """Callback function for agent to emit events to the stream"""
                event_queue.put_nowait(event)

            # Step 4: Generate unique message ID for tracking
            message_id = str(uuid.uuid4())

            // ...

    except Exception as e:
        # Step 14: Handle any errors during agent execution
        print(e)  # Log the error for debugging
        # Note: In production, you might want to emit an error event to the client

    # Step 15: Return the streaming response to the client
    # The event_generator() produces a stream of server-sent events
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Step 4: Configure AG-UI protocol lifecycle events

Once you have defined an event generator, define the AG-UI protocol lifecycle events that represent the lifecycle of an AG-UI + LlamaIndex agent workflow run as shown in the agent/main.py file.

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """
    try:
        # Step 1: Define the async event generator function
        # This function orchestrates the entire agent execution and event streaming
        async def event_generator():

            // ...

            # Step 5: Emit the run started event to notify the client
            yield encoder.encode(
                RunStartedEvent(
                    type=EventType.RUN_STARTED,
                    thread_id=input_data.thread_id,  # Conversation thread identifier
                    run_id=input_data.run_id,        # Unique run identifier
                )
            )

            // ...

            # Step 13: Signal completion of the agent run
            yield encoder.encode(
                RunFinishedEvent(
                    type=EventType.RUN_FINISHED,
                    thread_id=input_data.thread_id,  # Match the original thread ID
                    run_id=input_data.run_id,        # Match the original run ID
                )
            )

    except Exception as e:
        # Step 14: Handle any errors during agent execution
        print(e)  # Log the error for debugging
        # Note: In production, you might want to emit an error event to the client

    # Step 15: Return the streaming response to the client
    # The event_generator() produces a stream of server-sent events
    return StreamingResponse(event_generator(), media_type="text/event-stream

Step 5: Configure AG-UI protocol state management events

After defining AG-UI protocol lifecycle events, integrate AG-UI protocol state management events using the STATE_DELTA event in your LlamaIndex agent workflow, as shown in the agent/stock_analysis.py file.

from ag_ui.core import EventType, StateDeltaEvent  # Event management for UI state updates

class FuncationCallingAgent(Workflow):
    """
    Main workflow class for stock analysis and portfolio simulation.
    This agent processes user queries, fetches stock data, performs portfolio calculations,
    and generates investment insights using AI function calling.
    """
    def __init__(
        self,
        *args: Any,
        llm: FunctionCallingLLM | None = None,
        tools: List[BaseTool] | None = None,
        messages: List[Any],  # Conversation history
        emit_event: Callable,  # Function to emit UI state updates
        available_cash: int,  # Total cash available for investment
        investment_portfolio: List[Any],  # Current portfolio holdings
        **kwargs: Any,
    ) -> None:
        """
        Initialize the stock analysis workflow agent.

        Args:
            messages: Chat message history
            emit_event: Function to emit UI state changes
            available_cash: Available cash for investments
            investment_portfolio: Current portfolio data
        """

        // ...

    @step
    async def chat_function(self, ctx: Context, ev: StartEvent) -> SimulationEvent:
        """
        First step: Analyze the user query and extract investment parameters.
        Uses AI function calling to parse ticker symbols, investment amounts, dates, etc.
        """
        # Step 1: Initialize error handling and progress tracking
        try:
            # Step 2: Create a unique tool log ID for UI progress tracking
            tool_log_id = str(uuid.uuid4())
            self.tool_logs.append(
                {
                    "id": tool_log_id,
                    "message": "Analyzing user query",
                    "status": "processing",
                }
            )

            # Step 3: Emit UI state update to show progress to the user
            self.emit_event(
                StateDeltaEvent(
                    type=EventType.STATE_DELTA,
                    delta=[
                        {
                            "op": "add",
                            "path": "/tool_logs/-",
                            "value": {
                                "message": "Analyzing user query",
                                "status": "processing",
                                "id": tool_log_id,
                            },
                        }
                    ],
                )
            )
            await asyncio.sleep(0)  # Yield control for async execution

            # Step 4: Call the OpenAI GPT model with function calling to extract investment data
            response = self.llm.chat.completions.create(
                model="gpt-4.1-mini",
                messages=self.messages,
                tools=[extract_relevant_data_from_user_prompt],  # Function schema for data extraction
            )
            print(response)  # Debug: Show AI response

            # Step 5: Handle AI response - check if it made function calls
            if response.choices[0].finish_reason == "tool_calls":
                # Step 5a: AI wants to call the data extraction function
                tool_calls = [
                    convert_tool_call(tc)
                    for tc in response.choices[0].message.tool_calls
                ]

                # Step 5b: Add AI message with tool calls to conversation history
                a_message = AssistantMessage(
                    role="assistant", tool_calls=tool_calls, id=response.id
                )
                self.messages.append(a_message)

                # Step 5c: Update UI to show completion of analysis step
                index = len(self.tool_logs) - 1
                self.emit_event(
                    StateDeltaEvent(
                        type=EventType.STATE_DELTA,
                        delta=[
                            {
                                "op": "replace",
                                "path": f"/tool_logs/{index}/status",
                                "value": "completed",
                            }
                        ],
                    )
                )
                await asyncio.sleep(0)
                # Step 5d: Proceed to next workflow step (simulation)
                return SimulationEvent(input=self.messages)
            else:
                # Step 6: AI provided a direct response without function calls
                a_message = AssistantMessage(
                    id=response.id,
                    content=response.choices[0].message.content,
                    role="assistant",
                )
                self.messages.append(a_message)

                # Step 6a: Update UI status and end workflow
                index = len(self.tool_logs) - 1
                self.emit_event(
                    StateDeltaEvent(
                        type=EventType.STATE_DELTA,
                        delta=[
                            {
                                "op": "replace",
                                "path": f"/tool_logs/{index}/status",
                                "value": "completed",
                            }
                        ],
                    )
                )
                await asyncio.sleep(0)
                return StopEvent(result=self.messages)
        except Exception as e:
            # Step 7: Handle any errors in the chat analysis step
            print(e)
            return StopEvent(
                result={"response": "Error in chat function", "sources": []}
            )

Then in the FastAPI endpoint, initialize your LlamaIndex agent workflow state using the STATE_SNAPSHOT state management event, as shown below.

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """
    try:
        # Step 1: Define the async event generator function
        # This function orchestrates the entire agent execution and event streaming
        async def event_generator():

            // ...

            # Step 6: Send initial state snapshot to client
            # This provides the frontend with the current portfolio and cash state
            yield encoder.encode(
                StateSnapshotEvent(
                    type=EventType.STATE_SNAPSHOT,
                    snapshot={
                        "available_cash": input_data.state["available_cash"],
                        "investment_summary": input_data.state["investment_summary"],
                        "investment_portfolio": input_data.state["investment_portfolio"],
                        "tool_logs": [],  # Start with empty tool logs
                    },
                )
            )

            // ...

Step 6: Configure your LlamaIndex agent workflow with AG-UI protocol

Once you have initialized the LlamaIndex agent workflow state, integrate your LlamaIndex agent workflow with AG-UI protocol, as in the agent/main.py file.

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """
    try:
        # Step 1: Define the async event generator function
        # This function orchestrates the entire agent execution and event streaming
        async def event_generator():

            // ...

            # Step 7: Initialize agent state object
            # Creates a structured state object with all necessary data for the workflow
            state = AgentState(
                tools=input_data.tools,                                    # Available function tools
                messages=input_data.messages,                              # Conversation history
                be_stock_data=None,                                        # Will be populated with stock data
                be_arguments=None,                                         # Will be populated with investment params
                available_cash=input_data.state["available_cash"],         # Cash available for investing
                investment_portfolio=input_data.state["investment_portfolio"], # Current portfolio holdings
                tool_logs=[],                                              # Progress tracking logs
            )

            # Step 8: Create and configure the stock analysis agent
            agent = FuncationCallingAgent(
                tools=input_data.tools,                                    # Function calling tools
                messages=input_data.messages,                              # Message history
                emit_event=emit_event,                                     # Callback for real-time events
                available_cash=input_data.state["available_cash"],         # Investment budget
                investment_portfolio=input_data.state["investment_portfolio"], # Portfolio data
            )

            # Step 9: Start the agent workflow execution asynchronously
            # The agent will run through: chat → simulation → allocation → insights
            agent_task = agent.run(input=input_data.messages[-1].content)

            # Step 10: Event streaming loop - relay agent events to client
            # This loop runs until the agent completes its workflow
            while True:
                try:
                    # Step 10a: Wait for events from agent (with timeout for responsiveness)
                    event = await asyncio.wait_for(event_queue.get(), timeout=0.1)
                    # Step 10b: Encode and stream the event to the client
                    yield encoder.encode(event)
                except asyncio.TimeoutError:
                    # Step 10c: Check if the agent workflow has completed
                    if agent_task.done():
                        break  # Exit the streaming loop

            # Step 11: Clear tool logs after workflow completion
            # Reset progress indicators since the analysis is complete
            yield encoder.encode(
                StateDeltaEvent(
                    type=EventType.STATE_DELTA,
                    delta=[{"op": "replace", "path": "/tool_logs", "value": []}],
                )
            )

            // ...

    except Exception as e:
        # Step 14: Handle any errors during agent execution
        print(e)  # Log the error for debugging
        # Note: In production, you might want to emit an error event to the client

    # Step 15: Return the streaming response to the client
    # The event_generator() produces a stream of server-sent events
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Step 7: Configure AG-UI protocol tool events to handle Human-in-the-Loop breakpoint

After integrating your LlamaIndex agent workflow with AG-UI protocol, append a tool call message with a tool call name to the state, as shown in the cash allocation step in the agent/stock_analysis.py file.

class FuncationCallingAgent(Workflow):
    """
    Main workflow class for stock analysis and portfolio simulation.
    This agent processes user queries, fetches stock data, performs portfolio calculations,
    and generates investment insights using AI function calling.
    """
    def __init__(
        self,
        *args: Any,
        llm: FunctionCallingLLM | None = None,
        tools: List[BaseTool] | None = None,
        messages: List[Any],  # Conversation history
        emit_event: Callable,  # Function to emit UI state updates
        available_cash: int,  # Total cash available for investment
        investment_portfolio: List[Any],  # Current portfolio holdings
        **kwargs: Any,
    ) -> None:
        """
        Initialize the stock analysis workflow agent.

        Args:
            messages: Chat message history
            emit_event: Function to emit UI state changes
            available_cash: Available cash for investments
            investment_portfolio: Current portfolio data
        """

        // ...

    @step
    async def cash_allocation_function(
        self, ctx: Context, ev: CashAllocationEvent
    ) -> InsightsEvent:
        """
        Third step: Perform portfolio simulation and cash allocation calculations.
        Simulates investment strategy (single shot or DCA) and calculates returns.
        """

        // ...

        # Step 20: Add tool message to conversation history
        # This confirms the data extraction was completed successfully
        self.messages.append(
            ToolMessage(
                role="tool",
                id=str(uuid.uuid4()),
                content="The relevant details had been extracted",
                tool_call_id=self.messages[-1].tool_calls[0].id,
            )
        )

        # Step 21: Add assistant message requesting chart/table rendering
        # This triggers the UI to display charts and allocation tables
        self.messages.append(
            AssistantMessage(
                role="assistant",
                tool_calls=[
                    {
                        "id": str(uuid.uuid4()),
                        "type": "function",
                        "function": {
                            "name": "render_standard_charts_and_table",
                            "arguments": json.dumps(
                                {"investment_summary": self.investment_summary}
                            ),
                        },
                    }
                ],
                id=str(uuid.uuid4()),
            )
        )

        # Step 22: Update UI to show completion of cash allocation step
        index = len(self.tool_logs) - 1
        self.emit_event(
            StateDeltaEvent(
                type=EventType.STATE_DELTA,
                delta=[
                    {
                        "op": "replace",
                        "path": f"/tool_logs/{index}/status",
                        "value": "completed",
                    }
                ],
            )
        )
        await asyncio.sleep(0)

        # Step 23: Proceed to insights generation step
        return InsightsEvent(input=self.messages)

Then, define AG-UI protocol tool call events that an agent can use to trigger frontend actions by calling the frontend action using a tool name in order to request user feedback, as shown in the agent/main.py file.

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """
    try:
        # Step 1: Define the async event generator function
        # This function orchestrates the entire agent execution and event streaming
        async def event_generator():

            // ...

            # Step 12: Process and stream the final agent response
            # Check the type of response (tool call vs text message)
            if agent_task.result()[-1].role == "assistant":
                if agent_task.result()[-1].tool_calls:
                    # Step 12a: Handle tool call response (e.g., chart rendering request)
                    # This occurs when the agent wants to display charts/tables

                    # Step 12a-i: Signal start of tool call
                    yield encoder.encode(
                        ToolCallStartEvent(
                            type=EventType.TOOL_CALL_START,
                            tool_call_id=agent_task.result()[-1].tool_calls[0].id,
                            toolCallName=agent_task.result()[-1].tool_calls[0].function.name,
                        )
                    )

                    # Step 12a-ii: Stream tool call arguments (investment data + insights)
                    yield encoder.encode(
                        ToolCallArgsEvent(
                            type=EventType.TOOL_CALL_ARGS,
                            tool_call_id=agent_task.result()[-1].tool_calls[0].id,
                            delta=agent_task.result()[-1].tool_calls[0].function.arguments,
                        )
                    )

                    # Step 12a-iii: Signal end of tool call
                    yield encoder.encode(
                        ToolCallEndEvent(
                            type=EventType.TOOL_CALL_END,
                            tool_call_id=agent_task.result()[-1].tool_calls[0].id,
                        )
                    )
                else:

                    // ...

    except Exception as e:
        # Step 14: Handle any errors during agent execution
        print(e)  # Log the error for debugging
        # Note: In production, you might want to emit an error event to the client

    # Step 15: Return the streaming response to the client
    # The event_generator() produces a stream of server-sent events
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Step 8: Configure AG-UI protocol text message events

Once you have configured AG-UI protocol tool events, define AG-UI protocol text message events in order to handle streaming agent responses to the frontend, as shown in the agent/main.py file.

@app.post("/llamaindex-agent")
async def llama_index_agent(input_data: RunAgentInput):
    """
    Main API endpoint for running the stock analysis agent.

    This endpoint handles incoming requests to analyze stock investments,
    manages the agent workflow execution, and streams real-time events
    back to the client for live progress updates.

    Args:
        input_data: RunAgentInput containing user message, state, tools, and metadata

    Returns:
        StreamingResponse: Real-time event stream of agent execution progress
    """
    try:
        # Step 1: Define the async event generator function
        # This function orchestrates the entire agent execution and event streaming
        async def event_generator():

            // ...

            # Step 12: Process and stream the final agent response
            # Check the type of response (tool call vs text message)
            if agent_task.result()[-1].role == "assistant":

                // ...

                else:
                    # Step 12b: Handle text message response
                    # This occurs when the agent provides a direct text response

                    # Step 12b-i: Signal start of text message
                    yield encoder.encode(
                        TextMessageStartEvent(
                            type=EventType.TEXT_MESSAGE_START,
                            message_id=message_id,
                            role="assistant",
                        )
                    )

                    # Step 12b-ii: Stream message content with chunking for smooth display
                    if agent_task.result()[-1].content:
                        content = agent_task.result()[-1].content

                        # Step 12b-ii-1: Split content into chunks for streaming effect
                        n_parts = 100  # Number of chunks for smooth streaming
                        part_length = max(1, len(content) // n_parts)
                        parts = [
                            content[i : i + part_length]
                            for i in range(0, len(content), part_length)
                        ]

                        # Step 12b-ii-2: Handle rounding issues by merging excess parts
                        if len(parts) > n_parts:
                            parts = parts[: n_parts - 1] + [
                                "".join(parts[n_parts - 1 :])
                            ]

                        # Step 12b-ii-3: Stream each content chunk with a delay for typing effect
                        for part in parts:
                            yield encoder.encode(
                                TextMessageContentEvent(
                                    type=EventType.TEXT_MESSAGE_CONTENT,
                                    message_id=message_id,
                                    delta=part,  # Chunk of the message content
                                )
                            )
                            await asyncio.sleep(0.05)  # Small delay for streaming effect
                    else:
                        # Step 12b-ii-4: Handle empty content case with error message
                        yield encoder.encode(
                            TextMessageContentEvent(
                                type=EventType.TEXT_MESSAGE_CONTENT,
                                message_id=message_id,
                                delta="Something went wrong! Please try again.",
                            )
                        )

                    # Step 12b-iii: Signal end of text message
                    yield encoder.encode(
                        TextMessageEndEvent(
                            type=EventType.TEXT_MESSAGE_END,
                            message_id=message_id,
                        )
                    )

            // ...

    except Exception as e:
        # Step 14: Handle any errors during agent execution
        print(e)  # Log the error for debugging
        # Note: In production, you might want to emit an error event to the client

    # Step 15: Return the streaming response to the client
    # The event_generator() produces a stream of server-sent events
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Congratulations! You have integrated a LlamaIndex agent workflow with AG-UI protocol. Let’s now see how to add a frontend to the AG-UI + LlamaIndex agent workflow.

Integrating a frontend to the AG-UI + LlamaIndex agent workflow using CopilotKit

In this section, you will learn how to create a connection between your AG-UI + LlamaIndex agent workflow and a frontend using CopilotKit.

Let’s get started.

First, navigate to the frontend directory:

cd frontend

Next, create a .env file with OpenAI API Key API key:

OPENAI_API_KEY=<<your-OpenAI-key-here>>

Then install the dependencies:

pnpm install

After that, start the development server:

pnpm run dev

Navigate to http://localhost:3000, and you should see the AG-UI + LlamaIndex agent frontend up and running.

Let’s now see how to build the frontend UI for the AG-UI + LlamaIndex agent using CopilotKit.

Step 1: Create an HttpAgent instance

Before creating an HttpAgent instance, let’s first understand what HttpAgent is.

HttpAgent is a client from the AG-UI library that bridges your frontend application with any AG-UI-compatible AI agent’s server.

To create an HttpAgent instance, define it in an API route as shown in the src/app/api/copilotkit/route.ts file.

// Import necessary CopilotKit components for runtime and AI integration
import {
  CopilotRuntime, // Core runtime for managing AI agents and conversations
  copilotRuntimeNextJSAppRouterEndpoint, // Next.js App Router integration helper
  OpenAIAdapter, // Adapter for OpenAI API integration
} from "@copilotkit/runtime";

// Import Next.js request handling types
import { NextRequest } from "next/server";

// Import AG UI client for external agent communication
import { HttpAgent } from "@ag-ui/client";

// Step 1: Configure the LlamaIndex agent connection
// Create an HTTP agent client to communicate with the Python stock analysis service
const llamaIndexAgent = new HttpAgent({
  // Step 1a: Set agent URL from environment variable or use local development default
  // This points to the FastAPI server running the stock analysis workflow
  url:
    process.env.NEXT_PUBLIC_LLAMAINDEX_URL ||
    "http://0.0.0.0:8000/llamaindex-agent",

  // Alternative production URL (commented out for development)
  // url: "https://open-ag-ui-demo-llamaindex.onrender.com/llamaindex-agent",
});

// Step 2: Initialize OpenAI service adapter
// This adapter handles communication with OpenAI's API for language model interactions
const serviceAdapter = new OpenAIAdapter();

// Step 3: Create CopilotRuntime with agent configuration
// The runtime orchestrates conversations between the frontend and backend agents
const runtime = new CopilotRuntime({
  agents: {
    // Step 3a: Register the LlamaIndex agent for stock analysis
    // @ts-ignore - Suppress TypeScript warnings for agent type compatibility
    llamaIndexAgent: llamaIndexAgent,
  },
});

// Alternative simpler runtime configuration (commented out)
// const runtime = new CopilotRuntime()
// Step 4: Define the POST endpoint handler for CopilotKit API requests
// This endpoint receives requests from the frontend React components and routes them to agents
export const POST = async (req: NextRequest) => {
  // Step 4a: Create request handler using CopilotKit's Next.js integration
  // This configures the endpoint to work with Next.js App Router
  const { handleRequest } = copilotRuntimeNextJSAppRouterEndpoint({
    runtime, // The configured CopilotRuntime with our agents
    serviceAdapter, // OpenAI adapter for language model communication
    endpoint: "/api/copilotkit", // API endpoint path for client connections
  });

  // Step 4b: Delegate request handling to CopilotKit
  // This processes the incoming request and routes it to the appropriate agent
  // Returns streaming responses for real-time communication
  return handleRequest(req);
};

Step 2: Set up CopilotKit provider

To set up the CopilotKit Provider, the [<CopilotKit>](https://docs.copilotkit.ai/reference/components/CopilotKit) component must wrap the Copilot-aware parts of your application.

For most use cases, it's appropriate to wrap the CopilotKit provider around the entire app, e.g., in your layout.tsx, as shown below in the src/app/layout.tsx file.

// Next.js imports for metadata and font handling
import type { Metadata } from "next";
import { Geist, Geist_Mono } from "next/font/google";
// Global styles for the application
import "./globals.css";
// CopilotKit UI styles for AI components
import "@copilotkit/react-ui/styles.css";
// CopilotKit core component for AI functionality
import { CopilotKit } from "@copilotkit/react-core";

// Configure Geist Sans font with CSS variables for consistent typography
const geistSans = Geist({
  variable: "--font-geist-sans",
  subsets: ["latin"],
});

// Configure Geist Mono font for code and monospace text
const geistMono = Geist_Mono({
  variable: "--font-geist-mono",
  subsets: ["latin"],
});

// Metadata configuration for SEO and page information
export const metadata: Metadata = {
  title: "AI Stock Portfolio",
  description: "AI Stock Portfolio",
};

// Root layout component that wraps all pages in the application
export default function RootLayout({
  children,
}: Readonly<{
  children: React.ReactNode;
}>) {
  return (
    <html lang="en">
      <body
        className={`${geistSans.variable} ${geistMono.variable} antialiased`}>
        {/* CopilotKit wrapper that enables AI functionality throughout the app */}
        {/* runtimeUrl points to the API endpoint for AI backend communication */}
        {/* agent specifies which AI agent to use (stockAgent for stock analysis) */}
        <CopilotKit runtimeUrl="/api/copilotkit" agent="llamaIndexAgent">
          {children}
        </CopilotKit>
      </body>
    </html>
  );
}

Step 3: Set up a Copilot chat component

CopilotKit ships with a number of built-in chat components, which include CopilotPopup, CopilotSidebar, and CopilotChat.

To set up a Copilot chat component, define it as shown in the src/app/components/prompt-panel.tsx file.

// Client-side component directive for Next.js
"use client";

import type React from "react";
// CopilotKit chat component for AI interactions
import { CopilotChat } from "@copilotkit/react-ui";

// Props interface for the PromptPanel component
interface PromptPanelProps {
  // Amount of available cash for investment, displayed in the panel
  availableCash: number;
}

// Main component for the AI chat interface panel
export function PromptPanel({ availableCash }: PromptPanelProps) {
  // Utility function to format numbers as USD currency
  // Removes decimal places for cleaner display of large amounts
  const formatCurrency = (amount: number) => {
    return new Intl.NumberFormat("en-US", {
      style: "currency",
      currency: "USD",
      minimumFractionDigits: 0,
      maximumFractionDigits: 0,
    }).format(amount);
  };

  return (
    // Main container with full height and white background
    <div className="h-full flex flex-col bg-white">
      {/* Header section with title, description, and cash display */}
      <div className="p-4 border-b border-[#D8D8E5] bg-[#FAFCFA]">
        {/* Title section with icon and branding */}
        <div className="flex items-center gap-2 mb-2">
          <span className="text-xl">🪁</span>
          <div>
            <h1 className="text-lg font-semibold text-[#030507] font-['Roobert']">
              Portfolio Chat
            </h1>
            {/* Pro badge indicator */}
            <div className="inline-block px-2 py-0.5 bg-[#BEC9FF] text-[#030507] text-xs font-semibold uppercase rounded">
              PRO
            </div>
          </div>
        </div>
        {/* Description of the AI agent's capabilities */}
        <p className="text-xs text-[#575758]">
          Interact with the LangGraph-powered AI agent for portfolio
          visualization and analysis
        </p>

        {/* Available Cash Display section */}
        <div className="mt-3 p-2 bg-[#86ECE4]/10 rounded-lg">
          <div className="text-xs text-[#575758] font-medium">
            Available Cash
          </div>
          <div className="text-sm font-semibold text-[#030507] font-['Roobert']">
            {formatCurrency(availableCash)}
          </div>
        </div>
      </div>
      {/* CopilotKit chat interface with custom styling and initial message */}
      {/* Takes up majority of the panel height for conversation */}
      <CopilotChat
        className="h-[78vh] p-2"
        labels={{
          // Initial welcome message explaining the AI agent's capabilities and limitations
          initial: `I am a Crew AI agent designed to analyze investment opportunities and track stock performance over time. How can I help you with your investment query? For example, you can ask me to analyze a stock like "Invest in Apple with 10k dollars since Jan 2023". \n\nNote: The AI agent has access to stock data from the past 4 years only.
        }}
      />
    </div>
  );
}

Step 4: Sync AG-UI + LlamaIndex agent state with the frontend using CopilotKit hooks

In CopilotKit, CoAgents maintain a shared state that seamlessly connects your frontend UI with the agent's execution. This shared state system allows you to:

• Display the agent's current progress and intermediate results
• Update the agent's state through UI interactions
• React to state changes in real-time across your application

You can learn more about CoAgents’ shared state here on the CopilotKit docs.

To sync your AG-UI + LlamaIndex agent state with the frontend, use the CopilotKit useCoAgent hook to share the AG-UI + LlamaIndex agent state with your frontend, as shown in the src/app/page.tsx file.

"use client";

import {
  useCoAgent,
} from "@copilotkit/react-core";

// ...

export interface SandBoxPortfolioState {
  performanceData: Array<{
    date: string;
    portfolio: number;
    spy: number;
  }>;
}
export interface InvestmentPortfolio {
  ticker: string;
  amount: number;
}

export default function OpenStocksCanvas() {

  // ...

  const [totalCash, setTotalCash] = useState(1000000);

  const { state, setState } = useCoAgent({
    name: "llamaIndexAgent",
    initialState: {
      available_cash: totalCash,
      investment_summary: {} as any,
      investment_portfolio: [] as InvestmentPortfolio[],
    },
  });

    // ...

  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
       {/* ... */}
    </div>
  );
}

Then render the AG-UI + LlamaIndex agent's state in the chat UI, which is useful for informing the user about the agent's state in a more in-context way.

To render the AG-UI + LlamaIndex agent's state in the chat UI, you can use the useCoAgentStateRender hook, as shown in the src/app/page.tsx file.

"use client";

import {
  useCoAgentStateRender,
} from "@copilotkit/react-core";

import { ToolLogs } from "./components/tool-logs";

// ...

export default function OpenStocksCanvas() {

  // ...

  useCoAgentStateRender({
    name: "llamaIndexAgent",
    render: ({ state }) => <ToolLogs logs={state.tool_logs} />,
  });

  // ...

  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
      {/* ... */}
    </div>
  );
}

If you execute a query in the chat, you should see the AG-UI + LlamaIndex agent’s state task execution rendered in the chat UI, as shown below.

Step 5: Implementing Human-in-the-Loop (HITL) in the frontend

Human-in-the-loop (HITL) allows agents to request human input or approval during execution, making AI systems more reliable and trustworthy. This pattern is essential when building AI applications that need to handle complex decisions or actions that require human judgment.

You can learn more about Human in the Loop here on CopilotKit docs.

To implement Human-in-the-Loop (HITL) in the frontend, you need to use the CopilotKit useCopilotKitAction hook with the renderAndWaitForResponse method, which allows returning values asynchronously from the render function, as shown in the src/app/page.tsx file.

"use client";

import {
  useCopilotAction,
} from "@copilotkit/react-core";

// ...

export default function OpenStocksCanvas() {

  // ...

  useCopilotAction({
    name: "render_standard_charts_and_table",
    description:
      "This is an action to render a standard chart and table. The chart can be a bar chart or a line chart. The table can be a table of data.",
    renderAndWaitForResponse: ({ args, respond, status }) => {
      useEffect(() => {
        console.log(args, "argsargsargsargsargsaaa");
      }, [args]);
      return (
        <>
          {args?.investment_summary?.percent_allocation_per_stock &&
            args?.investment_summary?.percent_return_per_stock &&
            args?.investment_summary?.performanceData && (
              <>
                <div className="flex flex-col gap-4">
                  <LineChartComponent
                    data={args?.investment_summary?.performanceData}
                    size="small"
                  />
                  <BarChartComponent
                    data={Object.entries(
                      args?.investment_summary?.percent_return_per_stock
                    ).map(([ticker, return1]) => ({
                      ticker,
                      return: return1 as number,
                    }))}
                    size="small"
                  />
                  <AllocationTableComponent
                    allocations={Object.entries(
                      args?.investment_summary?.percent_allocation_per_stock
                    ).map(([ticker, allocation]) => ({
                      ticker,
                      allocation: allocation as a number,
                      currentValue:
                        args?.investment_summary.final_prices[ticker] *
                        args?.investment_summary.holdings[ticker],
                      totalReturn:
                        args?.investment_summary.percent_return_per_stock[
                          ticker
                        ],
                    }))}
                    size="small"
                  />
                </div>

                <button
                  hidden={status == "complete"}
                  className="mt-4 rounded-full px-6 py-2 bg-green-50 text-green-700 border border-green-200 shadow-sm hover:bg-green-100 transition-colors font-semibold text-sm"
                  onClick={() => {
                    debugger;
                    if (respond) {
                      setTotalCash(args?.investment_summary?.cash);
                      setCurrentState({
                        ...currentState,
                        returnsData: Object.entries(
                          args?.investment_summary?.percent_return_per_stock
                        ).map(([ticker, return1]) => ({
                          ticker,
                          return: return1 as number,
                        })),
                        allocations: Object.entries(
                          args?.investment_summary?.percent_allocation_per_stock
                        ).map(([ticker, allocation]) => ({
                          ticker,
                          allocation: allocation as a number,
                          currentValue:
                            args?.investment_summary?.final_prices[ticker] *
                            args?.investment_summary?.holdings[ticker],
                          totalReturn:
                            args?.investment_summary?.percent_return_per_stock[
                              ticker
                            ],
                        })),
                        performanceData:
                          args?.investment_summary?.performanceData,
                        bullInsights: args?.insights?.bullInsights || [],
                        bearInsights: args?.insights?.bearInsights || [],
                        currentPortfolioValue:
                          args?.investment_summary?.total_value,
                        totalReturns: (
                          Object.values(
                            args?.investment_summary?.returns
                          ) as number[]
                        ).reduce((acc, val) => acc + val, 0),
                      });
                      setInvestedAmount(
                        (
                          Object.values(
                            args?.investment_summary?.total_invested_per_stock
                          ) as number[]
                        ).reduce((acc, val) => acc + val, 0)
                      );
                      setState({
                        ...state,
                        available_cash: totalCash,
                      });
                      respond(
                        "Data rendered successfully. Provide a summary of the investments by not making any tool calls."
                      );
                    }
                  }}>
                  Accept
                </button>
                <button
                  hidden={status == "complete"}
                  className="rounded-full px-6 py-2 bg-red-50 text-red-700 border border-red-200 shadow-sm hover:bg-red-100 transition-colors font-semibold text-sm ml-2"
                  onClick={() => {
                    debugger;
                    if (respond) {
                      respond(
                        "Data rendering rejected. Just give a summary of the rejected investments by not making any tool calls."
                      );
                    }
                  }}>
                  Reject
                </button>
              </>
            )}
        </>
      );
    },
  });

  // ...

  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
      {/* ... */}
    </div>
  );
}

When an agent triggers frontend actions by tool/action name to request human input or feedback during execution, the end-user is prompted with a choice (rendered inside the chat UI). Then the user can choose by pressing a button in the chat UI, as shown below.

Step 6: Streaming AG-UI + LlamaIndex agent responses in the frontend

To stream your AG-UI + LlamaIndex agent responses or results in the frontend, pass the agent’s state field values to the frontend components, as shown in the src/app/page.tsx file.

"use client";

import { useEffect, useState } from "react";
import { PromptPanel } from "./components/prompt-panel";
import { GenerativeCanvas } from "./components/generative-canvas";
import { ComponentTree } from "./components/component-tree";
import { CashPanel } from "./components/cash-panel";

// ...

export default function OpenStocksCanvas() {
  const [currentState, setCurrentState] = useState<PortfolioState>({
    id: "",
    trigger: "",
    performanceData: [],
    allocations: [],
    returnsData: [],
    bullInsights: [],
    bearInsights: [],
    currentPortfolioValue: 0,
    totalReturns: 0,
  });
  const [sandBoxPortfolio, setSandBoxPortfolio] = useState<
    SandBoxPortfolioState[]
  >([]);
  const [selectedStock, setSelectedStock] = useState<string | null>(null);


  return (
    <div className="h-screen bg-[#FAFCFA] flex overflow-hidden">
      {/* Left Panel - Prompt Input */}
      <div className="w-85 border-r border-[#D8D8E5] bg-white flex-shrink-0">
        <PromptPanel availableCash={totalCash} />
      </div>

      {/* Center Panel - Generative Canvas */}
      <div className="flex-1 relative min-w-0">
        {/* Top Bar with Cash Info */}
        <div className="absolute top-0 left-0 right-0 bg-white border-b border-[#D8D8E5] p-4 z-10">
          <CashPanel
            totalCash={totalCash}
            investedAmount={investedAmount}
            currentPortfolioValue={
              totalCash + investedAmount + currentState.totalReturns || 0
            }
            onTotalCashChange={setTotalCash}
            onStateCashChange={setState}
          />
        </div>

        <div className="pt-20 h-full">
          <GenerativeCanvas
            setSelectedStock={setSelectedStock}
            portfolioState={currentState}
            sandBoxPortfolio={sandBoxPortfolio}
            setSandBoxPortfolio={setSandBoxPortfolio}
          />
        </div>
      </div>

      {/* Right Panel - Component Tree (Optional) */}
      {showComponentTree && (
        <div className="w-64 border-l border-[#D8D8E5] bg-white flex-shrink-0">
          <ComponentTree portfolioState={currentState} />
        </div>
      )}
    </div>
  );
}

If you query your agent and approve its feedback request, you should see the agent’s response or results streaming in the UI, as shown below.

Conclusion

In this guide, we have walked through the steps of integrating LlamaIndex agents with AG-UI protocol and then adding a frontend to the agents using CopilotKit.

While we’ve explored a couple of features, we have barely scratched the surface of the countless use cases for CopilotKit, ranging from building interactive AI chatbots to building agentic solutions—in essence, CopilotKit lets you add a ton of useful AI capabilities to your products in minutes.

Hopefully, this guide makes it easier for you to integrate AI-powered Copilots into your existing application.

Do you want to contribute to AG-UI?
Join the weekly Working Group!

Follow CopilotKit on Twitter and say hi, and if you'd like to build something cool, join the Discord community.