AG-UI Protocol Python SDK

0.1.15 · active · verified Thu Apr 09

AG-UI (Agent-User Interaction Protocol) is an open, lightweight, event-based protocol that standardizes how AI agents connect to user-facing applications. The `ag-ui-protocol` Python SDK provides strongly-typed data structures and event encoding for building AG-UI compatible agent servers. It's built on Pydantic for robust validation and automatic camelCase serialization for seamless frontend integration, and is currently at version 0.1.15. The library has a continuous release cadence, reflecting ongoing development in the AG-UI ecosystem.

Warnings

breaking The AG-UI protocol is under active development. While efforts are made to use versioned schemas, specific event types, data models, or API structures may change in future releases, potentially requiring updates to integrated agents and UIs.
Fix: Refer to the official AG-UI documentation and release notes for migration guides and updated API specifications with each new major version. Test integrations thoroughly on upgrade.
gotcha AG-UI servers should not be directly exposed to untrusted clients (e.g., browsers, mobile apps) due to security risks. A trusted frontend server should mediate communication to validate and control the construction of AG-UI protocol messages, preventing malicious client input.
Fix: Implement a trusted intermediary server between the AG-UI agent server and untrusted clients. This server is responsible for sanitizing inputs, validating requests, and constructing valid AG-UI messages.
gotcha Integrating AG-UI with existing agent frameworks or custom tooling often requires explicit configuration for features like tool exposure and event translation. Automatic handling might not be sufficient or desired. For instance, in `ag-ui-adk`, client tools are not automatically added to the root agent's toolset and require explicit `AGUIToolset` addition.
Fix: Carefully review the integration guides for your specific agent framework (e.g., LangGraph, Agent Framework, Pydantic AI) when implementing AG-UI. Expect to explicitly define and configure how AG-UI events, tools, and state are handled and mapped.
gotcha Effective state management in AG-UI requires understanding when to use `STATE_SNAPSHOT` (full state replacement) versus `STATE_DELTA` (incremental JSON Patch updates). Incorrect usage can lead to inefficient data transfer or inconsistent state between agent and UI.
Fix: Utilize `STATE_SNAPSHOT` for initial state establishment or major refreshes. Employ `STATE_DELTA` events (which use JSON Patch format) for frequent, small, incremental updates to optimize bandwidth and maintain real-time synchronization. Design state objects to support partial updates.

Install

pip install ag-ui-protocol Install stable version

Imports

RunAgentInput
```
from ag_ui.core import RunAgentInput
```
`RunAgentInput` is directly available from `ag_ui.core`, `ag_ui.core.types` is an unnecessary submodule access.
AgentEventEncoder
```
from ag_ui.encoder import AgentEventEncoder
```
Used for serializing AG-UI events into a streaming format (e.g., Server-Sent Events).
TextMessageContent
```
from ag_ui.core import TextMessageContent
```
A common event type for streaming text messages from the agent.

Quickstart

This quickstart demonstrates how to set up a basic AG-UI compatible FastAPI endpoint that receives `RunAgentInput` and streams `AgentEvent` responses, specifically `TextMessageContent` and `RunFinished` events, using Server-Sent Events (SSE). It echoes the last user message received.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from ag_ui.core import RunAgentInput, TextMessageContent, RunFinished, AgentEvent
from ag_ui.encoder import AgentEventEncoder
import uvicorn
import asyncio
import json
import os

app = FastAPI(title="AG-UI Endpoint Example")

@app.post("/awp")
async def my_endpoint(input_data: RunAgentInput):
    async def event_generator():
        # Initial greeting event
        yield AgentEventEncoder.encode(TextMessageContent(text="Hello from AG-UI!"))
        
        # Echo the last text message from the input
        if input_data.messages:
            last_message = input_data.messages[-1]
            # Ensure the last message is a TextMessageContent for echoing
            if isinstance(last_message, TextMessageContent):
                yield AgentEventEncoder.encode(TextMessageContent(text=f"You said: {last_message.text}"))
            else:
                yield AgentEventEncoder.encode(TextMessageContent(text="Received a non-text message type."))
        else:
            yield AgentEventEncoder.encode(TextMessageContent(text="No messages provided in input."))

        # Simulate some asynchronous work
        await asyncio.sleep(0.5)
        yield AgentEventEncoder.encode(TextMessageContent(text="Performing some agent actions..."))
        await asyncio.sleep(0.5)

        # End the run with a RunFinished event
        yield AgentEventEncoder.encode(RunFinished())

    # Use StreamingResponse for Server-Sent Events (SSE)
    return StreamingResponse(event_generator(), media_type="text/event-stream")

if __name__ == "__main__":
    # To run this, you'll need fastapi and uvicorn installed:
    # pip install "fastapi[all]" uvicorn
    # Then, run this script directly:
    # python your_script_name.py
    # Or, if saved as main.py:
    # uvicorn main:app --host 0.0.0.0 --port 8000
    uvicorn.run(app, host="0.0.0.0", port=8000)

view raw JSON →