agent-framework-agui-py

Microsoft Agent Framework — AG-UI Integration (Python)

AG-UI (Agent UI) is a protocol for streaming agent interactions over HTTP using Server-Sent Events. agent-framework-ag-ui lets you expose any Agent or Workflow as an AG-UI server with one call, and consume any AG-UI server through AGUIChatClient — including hybrid execution where some tools run on the server and others run in the client.

Architecture

   ┌──────────────────────────────────────┐         ┌──────────────────────────────────────┐
   │             AG-UI Server             │   SSE   │             AG-UI Client             │
   │                                      │ ◄─────► │                                      │
   │  FastAPI + add_agent_framework_      │  HTTP   │  AGUIChatClient                      │
   │  fastapi_endpoint(app, target, "/")  │         │  ├─ direct .get_response(...)        │
   │                                      │         │  └─ wrapped: Agent(client=AGUI…)     │
   │  target ∈ {Agent, Workflow,          │         │                                      │
   │            AgentFrameworkAgent,      │         │  - streaming + non-streaming         │
   │            AgentFrameworkWorkflow}   │         │  - thread_id continuity              │
   │                                      │         │  - hybrid tool execution             │
   │  Optional: dependencies=[Depends(…)] │         │  - interrupt / resume passthrough    │
   └──────────────────┬───────────────────┘         └──────────────────┬───────────────────┘
                      │                                                │
                      ▼                                                ▼
            Server-side tools                                Client-side tools
            (executed in-process)                            (executed locally via
                                                              function-invocation mixin)

Both sides use the same Agent Framework primitives (Agent, @tool, Message, Content, AgentSession). The AG-UI layer translates between the wire protocol and those primitives.

⚠ Frontend tools + GitHubCopilotAgent server. When the server-side agent is a GitHubCopilotAgent (i.e. the model is the local GitHub Copilot CLI rather than an OpenAIChatCompletionClient), the CLI's tool list is frozen at session-creation time. Frontend tools advertised by the client are not merged into the running Copilot session, so the model will often hallucinate the tool call ("notification sent") rather than dispatch it back to the client. For deterministic, machine-checkable evidence that a client tool fired, expose a parallel non-AG-UI route on the same FastAPI app (e.g. POST /scripted-reservation) and have the client invoke its local @tool after a direct call to that route. See GitHub Copilot CLI as the server backbone below.

Installation

pip install agent-framework-ag-ui
# or
uv pip install agent-framework-ag-ui

# Server also needs FastAPI + uvicorn (transitively pulled, pin explicitly for production):
pip install fastapi uvicorn

# Examples package with seven feature demos:
pip install agent-framework-ag-ui-examples

The integration exports from two equivalent paths:

from agent_framework.ag_ui import (
    AGUIChatClient,
    AgentFrameworkAgent,
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
    state_update,
)
# Or directly:
from agent_framework_ag_ui import AGUIChatClient

Status: the AG-UI protocol is still evolving — pin to a known-good agent-framework-ag-ui version in production.

Environment Variables

Server samples use an Azure OpenAI deployment exposed through OpenAIChatCompletionClient:

Variable	Purpose
AZURE_OPENAI_ENDPOINT	https://<resource>.openai.azure.com/
AZURE_OPENAI_MODEL	Deployment name (e.g. gpt-4o-mini)
AZURE_OPENAI_API_KEY	API key, or omit and rely on DefaultAzureCredential (az login)
AG_UI_API_KEY	Optional shared secret for the X-API-Key dependency
AGUI_SERVER_URL	Client-side: defaults to http://127.0.0.1:5100/

Any SupportsChatGetResponse client works (OpenAIChatClient, OpenAIChatCompletionClient, Azure AI Foundry, custom) — AG-UI does not lock the model provider.

Core Workflow

1. Host an Agent as an AG-UI endpoint

import os
from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
from fastapi import FastAPI

agent = Agent(
    name="AGUIAssistant",
    instructions="You are a helpful assistant.",
    client=OpenAIChatCompletionClient(
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        model=os.environ["AZURE_OPENAI_MODEL"],
        api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
    ),
)

app = FastAPI(title="AG-UI Server")
add_agent_framework_fastapi_endpoint(app, agent, "/")

# uvicorn server:app --host 127.0.0.1 --port 5100

add_agent_framework_fastapi_endpoint(app, target, path, *, dependencies=None, ...) accepts any SupportsAgentRun (an Agent or an AgentFrameworkAgent wrapper) or a Workflow / AgentFrameworkWorkflow. It registers a streaming POST route that emits AG-UI events as SSE.

2. Consume an AG-UI server with AGUIChatClient

import asyncio
import os
from typing import cast
from agent_framework import ChatResponse, ChatResponseUpdate, Message, ResponseStream
from agent_framework.ag_ui import AGUIChatClient


async def main():
    server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:5100/")
    async with AGUIChatClient(endpoint=server_url) as client:
        thread_id: str | None = None
        while True:
            message = input("\nUser (:q to quit): ")
            if message.lower() in (":q", "quit"):
                break

            metadata = {"thread_id": thread_id} if thread_id else None
            stream = client.get_response(
                [Message(role="user", contents=[message])],
                stream=True,
                options={"metadata": metadata} if metadata else None,
            )
            stream = cast(ResponseStream[ChatResponseUpdate, ChatResponse], stream)

            print("Assistant: ", end="", flush=True)
            async for update in stream:
                if not thread_id and update.additional_properties:
                    thread_id = update.additional_properties.get("thread_id")
                for content in update.contents:
                    if content.type == "text" and content.text:
                        print(content.text, end="", flush=True)
            print()


asyncio.run(main())

• AGUIChatClient is a BaseChatClient — it slots into any place a chat client is expected.
• Use async with to clean up the underlying HTTP connection automatically.
• The first streamed update carries additional_properties["thread_id"] — pass it back via metadata on subsequent calls to keep the conversation on the same server-side thread.

3. Host a Workflow

add_agent_framework_fastapi_endpoint accepts a built Workflow directly. Outputs from ctx.yield_output(...) are streamed back as AG-UI events:

from agent_framework import WorkflowBuilder, WorkflowContext, executor
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
from fastapi import FastAPI

@executor(id="start")
async def start(message: str, ctx: WorkflowContext) -> None:
    await ctx.yield_output(f"Workflow received: {message}")

workflow = WorkflowBuilder(start_executor=start).build()

app = FastAPI()
add_agent_framework_fastapi_endpoint(app, workflow, "/")

For workflows with runtime state (e.g. pending ctx.request_info interrupts, in-flight super-steps), build a fresh instance per thread with AgentFrameworkWorkflow(workflow_factory=...):

from agent_framework import Workflow, WorkflowBuilder
from agent_framework.ag_ui import AgentFrameworkWorkflow, add_agent_framework_fastapi_endpoint

def build_workflow_for_thread(thread_id: str) -> Workflow:
    return WorkflowBuilder(start_executor=...).build()

thread_scoped = AgentFrameworkWorkflow(
    workflow_factory=build_workflow_for_thread,
    name="my_workflow",
)
add_agent_framework_fastapi_endpoint(app, thread_scoped, "/")

Without a factory, all threads share one Workflow instance — fine for stateless graphs, but unsafe for graphs that buffer per-conversation state.

Hybrid tool execution

The hallmark of AG-UI: client-side tools and server-side tools coexist in one conversation. The server LLM sees every tool definition (its own + the client's) and decides which to call; the framework routes execution to the right side.

Client defines:        Server defines:
  get_weather()          get_time_zone()

User: "What's the weather in SF and what time is it?"
   │
   ▼
Agent sends full history + client tool definitions to the server
   │
   ▼
Server LLM plans: get_weather('SF'), get_time_zone('SF')
   │
   ▼
Server executes get_time_zone('SF') → "Pacific Time (UTC-8)"
Server emits function-call request → get_weather('SF')
   │
   ▼
Agent's function-invocation mixin intercepts → runs get_weather locally
   │
   ▼
Result ("Sunny, 72°F") returns to server
   │
   ▼
Server LLM combines both → "It's sunny and 72°F in SF, and the timezone is PT (UTC-8)."

Required wiring:

Side	Where	What
Server	Agent(..., tools=[server_only_tools])	Tools the server executes. Do not include client tools here.
Client	Agent(name=..., client=AGUIChatClient(...), tools=[client_tools])	Tools the client executes; metadata is forwarded to the server.

Without the Agent wrapper the client cannot execute client-side tools (the function-invocation middleware lives on Agent). Direct AGUIChatClient.get_response(..., tools=[...]) sends tool definitions o

---

Content truncated.