AG-UI Protocol Explained — Implementing the Standard Protocol Between Agents and Users with Microsoft Agent Framework

// The core AG-UI abstraction
type RunAgent = () => Observable<BaseEvent>
 
class MyAgent extends AbstractAgent {
  run(input: RunAgentInput): RunAgent {
    const { threadId, runId } = input
    return () =>
      from([
        { type: EventType.RUN_STARTED, threadId, runId },
        {
          type: EventType.MESSAGES_SNAPSHOT,
          messages: [
            { id: "msg_1", role: "assistant", content: "Hello!" }
          ],
        },
        { type: EventType.RUN_FINISHED, threadId, runId },
      ])
  }
}

import { HttpAgent, type AgentSubscriber } from "@ag-ui/client"
import { EventType } from "@ag-ui/core"
 
const agent = new HttpAgent({
  url: "https://your-agent-endpoint.com/agent",
  agentId: "my-agent",
  threadId: "conversation-123",
})
 
const subscriber: AgentSubscriber = {
  onEvent: ({ event }) => {
    switch(event.type) {
      case EventType.TEXT_MESSAGE_CONTENT:
        // Display text in UI
        break
      case EventType.TOOL_CALL_START:
        // Show tool call in progress
        break
    }
  },
  onRunFailed: ({ error }) => console.error("Agent error:", error),
  onRunFinalized: () => console.log("Run complete")
}
 
await agent.runAgent({
  tools: [...],
  context: [...]
}, subscriber)

interface BaseEvent {
  type: EventType       // Event type identifier
  timestamp?: number    // Event creation time (optional)
  rawEvent?: any        // Original event data if transformed (optional)
  metadata?: Record<string, any> // Extra event metadata (optional)
}

→ middleware1
  → middleware2
    → middleware3
      → agent.run()
    ← events flow back through middleware3
  ← events flow back through middleware2
← events flow back through middleware1

import { MiddlewareFunction } from "@ag-ui/client"
import { EventType } from "@ag-ui/core"
 
// Middleware that prefixes text messages
const prefixMiddleware: MiddlewareFunction = (input, next) => {
  return next.run(input).pipe(
    map(event => {
      if (
        event.type === EventType.TEXT_MESSAGE_CHUNK ||
        event.type === EventType.TEXT_MESSAGE_CONTENT
      ) {
        return { ...event, delta: `[AI]: ${event.delta}` }
      }
      return event
    })
  )
}
 
agent.use(prefixMiddleware)

import { Middleware } from "@ag-ui/client"
 
class MetricsMiddleware extends Middleware {
  private eventCount = 0
 
  run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
    const startTime = Date.now()
    return this.runNext(input, next).pipe(
      tap(event => {
        this.eventCount++
        this.metricsService.recordEvent(event.type)
      }),
      finalize(() => {
        this.metricsService.recordDuration(Date.now() - startTime)
      })
    )
  }
}

import { FilterToolCallsMiddleware } from "@ag-ui/client"
 
// Allow list
const filter = new FilterToolCallsMiddleware({
  allowedToolCalls: ["search", "calculate"]
})
 
// Block list
const blocker = new FilterToolCallsMiddleware({
  disallowedToolCalls: ["delete", "modify"]
})
 
agent.use(filter)

pip install agent-framework-ag-ui

# simple_agent.py
from typing import Any
from agent_framework import Agent, SupportsChatGetResponse
 
 
def simple_agent(client: SupportsChatGetResponse[Any]) -> Agent[Any]:
    """Create a simple chat agent."""
    return Agent[Any](
        name="simple_chat_agent",
        instructions="You are a helpful assistant. Be concise and friendly.",
        client=client,
    )

# server.py
from fastapi import FastAPI
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
from agents.simple_agent import simple_agent
 
app = FastAPI(title="AG-UI Demo Server")
 
# Create chat client
chat_client = OpenAIChatCompletionClient(model="gpt-4o")
 
# Register as AG-UI endpoint (local, single user)
add_agent_framework_fastapi_endpoint(
    app, simple_agent(chat_client), "/agentic_chat", single_user=True
)

→ RUN_STARTED(threadId, runId)
  → TEXT_MESSAGE_START(messageId, role="assistant")
    → TEXT_MESSAGE_CONTENT(messageId, delta="Hel")
    → TEXT_MESSAGE_CONTENT(messageId, delta="lo")
    → TEXT_MESSAGE_CONTENT(messageId, delta="!")
  → TEXT_MESSAGE_END(messageId)
→ RUN_FINISHED(threadId, runId)

# weather_agent.py
from typing import Any
from agent_framework import Agent, SupportsChatGetResponse, tool
 
 
@tool
def get_weather(location: str) -> dict[str, Any]:
    """Get current weather for a location.
 
    Args:
        location: The city to get weather for.
 
    Returns:
        Dictionary with temperature, conditions, humidity, etc.
    """
    weather_data = {
        "tokyo": {
            "temperature": 22,
            "conditions": "sunny",
            "humidity": 55,
            "wind_speed": 8,
        },
        "osaka": {
            "temperature": 24,
            "conditions": "partly cloudy",
            "humidity": 60,
            "wind_speed": 6,
        },
    }
    location_lower = location.lower()
    if location_lower in weather_data:
        return weather_data[location_lower]
    return {"temperature": 21, "conditions": "unknown", "humidity": 50, "wind_speed": 10}
 
 
@tool
def get_forecast(location: str, days: int = 3) -> str:
    """Get weather forecast for a location.
 
    Args:
        location: The city to get forecast for.
        days: Number of days to forecast (default: 3).
 
    Returns:
        Forecast information string.
    """
    forecast: list[str] = []
    for day in range(1, min(days, 7) + 1):
        forecast.append(f"Day {day}: Partly cloudy, {20 + day}°C")
    return f"{days}-day forecast for {location}:\n" + "\n".join(forecast)
 
 
def weather_agent(client: SupportsChatGetResponse[Any]) -> Agent[Any]:
    """Create a weather agent."""
    return Agent[Any](
        name="weather_agent",
        instructions=(
            "You are a helpful weather assistant. "
            "Use the get_weather and get_forecast functions to help users. "
            "Always provide friendly and informative responses."
        ),
        client=client,
        tools=[get_weather, get_forecast],
    )

→ RUN_STARTED
  → TOOL_CALL_START(toolCallId, toolCallName="get_weather")
    → TOOL_CALL_ARGS(toolCallId, delta='{"location":')
    → TOOL_CALL_ARGS(toolCallId, delta=' "Tokyo"}')
  → TOOL_CALL_END(toolCallId)
  → TOOL_CALL_RESULT(toolCallId, content='{"temperature": 22, ...}')
  → TEXT_MESSAGE_START(messageId, role="assistant")
    → TEXT_MESSAGE_CONTENT(messageId, delta="The current weather in Tokyo is...")
  → TEXT_MESSAGE_END(messageId)
→ RUN_FINISHED

# human_in_the_loop_agent.py
from typing import Any
from agent_framework import Agent, SupportsChatGetResponse, tool
 
 
@tool(approval_mode="always_require")
def delete_file(filename: str) -> str:
    """Delete a file (requires approval).
 
    Args:
        filename: Name of the file to delete.
 
    Returns:
        Deletion result message.
    """
    return f"File '{filename}' has been deleted."
 
 
@tool(approval_mode="always_require")
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email (requires approval).
 
    Args:
        to: Recipient address.
        subject: Email subject.
        body: Email body.
 
    Returns:
        Send result message.
    """
    return f"Email sent to {to} with subject '{subject}'."
 
 
def human_in_the_loop_agent(
    client: SupportsChatGetResponse[Any],
) -> Agent[Any]:
    """Create a Human-in-the-Loop agent."""
    return Agent[Any](
        name="hitl_agent",
        instructions=(
            "You are a helpful assistant that can delete files and send emails. "
            "Always confirm with the user before performing these actions."
        ),
        client=client,
        tools=[delete_file, send_email],
    )

# recipe_agent.py
from typing import Any
from agent_framework import Agent, SupportsChatGetResponse, tool
from agent_framework.ag_ui import AgentFrameworkAgent
 
 
@tool
def update_recipe(recipe_data: dict[str, Any]) -> str:
    """Update recipe information.
 
    Args:
        recipe_data: Recipe info (name, ingredients, etc.)
 
    Returns:
        Update result message.
    """
    return f"Recipe '{recipe_data.get('name', 'Unknown')}' has been updated."
 
 
def recipe_agent(client: SupportsChatGetResponse[Any]) -> AgentFrameworkAgent:
    """Create a recipe management agent."""
    agent = Agent[Any](
        name="recipe_agent",
        instructions=(
            "You are a recipe assistant. Help users create and modify recipes. "
            "Use the update_recipe tool to save changes."
        ),
        client=client,
        tools=[update_recipe],
    )
 
    # Define state schema
    state_schema = {
        "recipe": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "ingredients": {"type": "array"},
                "instructions": {"type": "array"},
            },
        }
    }
 
    return AgentFrameworkAgent(
        agent=agent,
        state_schema=state_schema,
        name="RecipeAgent",
        description="A recipe management agent with shared state",
    )

# document_writer_agent.py
from typing import Any
from agent_framework import Agent, SupportsChatGetResponse, tool
from agent_framework.ag_ui import AgentFrameworkAgent
 
 
@tool
def write_document(title: str, content: str) -> str:
    """Create a document.
 
    Args:
        title: Document title.
        content: Document content.
 
    Returns:
        Creation result message.
    """
    return f"Document '{title}' has been created."
 
 
def document_writer_agent(
    client: SupportsChatGetResponse[Any],
) -> AgentFrameworkAgent:
    """Create a document writer agent."""
    agent = Agent[Any](
        name="document_writer",
        instructions="You are a document writer. Create documents based on user requests.",
        client=client,
        tools=[write_document],
    )
 
    # Predictive State configuration
    predict_state_config = {
        "current_title": {"tool": "write_document", "tool_argument": "title"},
        "current_content": {"tool": "write_document", "tool_argument": "content"},
    }
 
    return AgentFrameworkAgent(
        agent=agent,
        state_schema={
            "current_title": {"type": "string"},
            "current_content": {"type": "string"},
        },
        predict_state_config=predict_state_config,
        require_confirmation=True,
    )

→ RUN_STARTED
  → TOOL_CALL_START(write_document)
    → TOOL_CALL_ARGS(delta='{"title": "AG-')
      → STATE_DELTA(op: replace, path: /current_title, value: "AG-")
    → TOOL_CALL_ARGS(delta='UI Guide"')
      → STATE_DELTA(op: replace, path: /current_title, value: "AG-UI Guide")
    → TOOL_CALL_ARGS(delta=', "content": "AG-UI is...')
      → STATE_DELTA(op: replace, path: /current_content, value: "AG-UI is...")
  → TOOL_CALL_END
→ RUN_FINISHED

from fastapi import FastAPI
from agent_framework import WorkflowBuilder, WorkflowContext, executor
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
 
 
@executor(id="start")
async def start(message: str, ctx: WorkflowContext) -> None:
    """Start step of the workflow."""
    await ctx.yield_output(f"Workflow received: {message}")
 
 
@executor(id="analyze")
async def analyze(data: str, ctx: WorkflowContext) -> None:
    """Analysis step."""
    result = f"Analysis of '{data}': positive sentiment detected."
    await ctx.yield_output(result)
 
 
workflow = (
    WorkflowBuilder(start_executor=start)
    .add_executor(analyze)
    .add_edge("start", "analyze")
    .build()
)
 
app = FastAPI()
add_agent_framework_fastapi_endpoint(app, workflow, "/workflow", single_user=True)

"""AG-UI Demo Server — All features integrated"""
import os
from fastapi import FastAPI
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
 
# Import agents
from agents.simple_agent import simple_agent
from agents.weather_agent import weather_agent
from agents.human_in_the_loop_agent import human_in_the_loop_agent
from agents.recipe_agent import recipe_agent
from agents.document_writer_agent import document_writer_agent
 
app = FastAPI(title="AG-UI Demo Server")
 
# Shared chat client
chat_client = OpenAIChatCompletionClient(
    model=os.getenv("OPENAI_CHAT_MODEL", "gpt-4o"),
    api_key=os.getenv("OPENAI_API_KEY"),
)
 
# Register endpoints for each feature
add_agent_framework_fastapi_endpoint(
    app, simple_agent(chat_client), "/agentic_chat",
    single_user=True,
)
add_agent_framework_fastapi_endpoint(
    app, weather_agent(chat_client), "/backend_tool_rendering",
    single_user=True,
)
add_agent_framework_fastapi_endpoint(
    app, human_in_the_loop_agent(chat_client), "/human_in_the_loop",
    single_user=True,
)
add_agent_framework_fastapi_endpoint(
    app, recipe_agent(chat_client), "/shared_state",
    single_user=True,
)
add_agent_framework_fastapi_endpoint(
    app, document_writer_agent(chat_client), "/predictive_state_updates",
    single_user=True,
)
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8888)

import asyncio
from agent_framework.ag_ui import AGUIChatClient
 
 
async def main():
    async with AGUIChatClient(endpoint="http://localhost:8888/agentic_chat") as client:
        # Streaming response
        async for update in client.get_response("Hello!", stream=True):
            for content in update.contents:
                if content.type == "text" and content.text:
                    print(content.text, end="", flush=True)
        print()
 
 
asyncio.run(main())

import os
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader
from agent_framework import Agent
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
 
# API key authentication setup
API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
EXPECTED_API_KEY = os.environ.get("AG_UI_API_KEY")
 
 
async def verify_api_key(
    api_key: str | None = Security(API_KEY_HEADER),
) -> None:
    """Verify the API key in request headers."""
    if not api_key or api_key != EXPECTED_API_KEY:
        raise HTTPException(status_code=401, detail="Invalid or missing API key")
 
 
app = FastAPI()
 
# Assume agent was created in a previous section
# agent = simple_agent(chat_client)
 
# Register endpoint WITH authentication
# Callers sharing one API key are treated as a single user
add_agent_framework_fastapi_endpoint(
    app,
    agent,
    "/",
    dependencies=[Depends(verify_api_key)],
    single_user=True,
)

import { useCopilotAction } from "@copilotkit/react-core"
 
// Define a tool on the frontend
useCopilotAction({
  name: "confirmAction",
  description: "Ask user to confirm an action",
  parameters: [
    { name: "action", type: "string", description: "Action to confirm" },
    { name: "importance", type: "string", enum: ["low", "medium", "high"] },
  ],
  handler: async ({ action, importance }) => {
    // Show confirmation dialog to user
    const confirmed = await showConfirmDialog(action, importance)
    return confirmed ? "approved" : "rejected"
  },
})