AG-UI 入门

本教程演示如何使用代理框架的 AG-UI 协议生成服务器和客户端应用程序。你将了解如何在 AG-UI 终结点后面托管代理，以及如何连接用于交互式对话的客户端。

您将构建什么

在本教程结束时，你将拥有：

托管可通过 HTTP 访问的 AI 代理的 AG-UI 服务器
连接到服务器并流式传输响应的客户端应用程序
了解 AG-UI 协议如何与代理框架协作

Prerequisites

.NET 8 或更高版本
ASP.NET Core项目
配置的 MAF AIAgent

此示例使用 Azure OpenAI，但MapAGUIServer适用于任何 MAF 代理。

创建 AG-UI 服务器

安装托管包：

dotnet add package Microsoft.Agents.AI.Hosting.AGUI.AspNetCore --prerelease

using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddAGUIServer();

AIAgent agent = CreateAgent();

WebApplication app = builder.Build();
app.MapAGUIServer("/", agent);
await app.RunAsync();

MapAGUIServer 接受 AG-UI RunAgentInput 请求，并通过服务器发送的事件（SSE）将代理的响应作为 AG-UI 事件流式传输。

在客户端示例使用的 URL 上运行服务器：

dotnet run --urls http://localhost:8888

Tip

有关完整的服务器和控制台客户端，请参阅.NET入门示例。

使用 .NET 客户端进行连接

AG-UI .NET SDK 提供 AGUIChatClient，后者实现 IChatClient 并且可以适应 MAF 智能体：

dotnet add package AGUI.Client --prerelease
dotnet add package Microsoft.Agents.AI --prerelease

using AGUI.Abstractions;
using AGUI.Client;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

using HttpClient httpClient = new() { BaseAddress = new Uri("http://localhost:8888") };
AGUIChatClient chatClient = new(new AGUIChatClientOptions(httpClient, "/"));
AIAgent remoteAgent = chatClient.AsAIAgent();
AgentSession session = await remoteAgent.CreateSessionAsync();

List<AgentResponseUpdate> firstTurnUpdates = [];
await foreach (AgentResponseUpdate update in
    remoteAgent.RunStreamingAsync("Hello", session))
{
    firstTurnUpdates.Add(update);

    foreach (TextContent text in update.Contents.OfType<TextContent>())
    {
        Console.Write(text.Text);
    }
}

还可以连接到任何实现 AG-UI 协议的客户端。

对话连续性

AG-UI 使用 threadId 和 parentRunId 标识延续请求。这些标识符是协议数据，而不是授权凭据。

AGUIChatClient 是无状态的。若要继续进行服务器拥有的对话，请从第一轮的 RunStartedEvent 中获取标识符，然后在下一次请求中包含相同的 threadId，并将前一个 runId 作为 parentRunId：

RunStartedEvent started = firstTurnUpdates
    .Select(update => update.AsChatResponseUpdate().RawRepresentation)
    .OfType<RunStartedEvent>()
    .FirstOrDefault()
    ?? throw new InvalidOperationException("The server didn't return a run-started event.");

ChatMessage nextMessage = new(ChatRole.User, "What did I just say?");
ChatClientAgentRunOptions continuationOptions = new()
{
    ChatOptions = new ChatOptions
    {
        RawRepresentationFactory = _ => new RunAgentInput
        {
            ThreadId = started.ThreadId,
            ParentRunId = started.RunId,
            Messages = new[] { nextMessage }.AsAGUIMessages().ToList(),
        },
    },
};

await foreach (AgentResponseUpdate update in
    remoteAgent.RunStreamingAsync([nextMessage], session, continuationOptions))
{
    // Process the continued response.
}

仅在延续请求中发送新消息。 MapAGUIServer 使用 threadId 来选择托管代理会话，并使用 parentRunId 来标识正在继续的运行。如果没有托管会话持久性，每个请求都会收到一个新的服务器会话;客户端可以改为重新发送会话历史记录。

若要在请求之间保留由服务器持有的 AgentSession 状态，请配置托管会话持久性和隔离，然后使用 MapAGUIServer 映射已命名的托管代理。有关 AG-UI 特定的信任边界，请参阅生产和安全注意事项。

后续步骤

将后端工具与 AG-UI 配合使用

Prerequisites

在开始之前，确保你拥有：

Python 3.10 或更高版本
已配置 Azure OpenAI 服务终结点和部署
已安装的 Azure CLI并且已进行身份验证
用户在 Azure OpenAI 资源中具有 Cognitive Services OpenAI Contributor 角色

注释

这些示例使用 Azure OpenAI 模型。有关详细信息，请参阅如何使用 Foundry 部署 Azure OpenAI 模型。

注释

这些示例使用 DefaultAzureCredential 进行身份验证。确保通过 Azure 进行身份验证（例如，通过 az login）。有关详细信息，请参阅 Azure 标识文档。

Warning

AG-UI 协议仍在开发中，可能会发生更改。随着协议的发展，我们将不断更新这些示例。

步骤 1：创建 AG-UI 服务器

AG-UI 服务器运行和托管你的智能代理，并使用 FastAPI 通过 HTTP 端点公开它。

安装所需的包

安装服务器所需的包：

pip install agent-framework-ag-ui

或使用紫外线：

uv pip install agent-framework-ag-ui

这将自动安装 agent-framework-core、fastapi、uvicorn 和 sse-starlette 作为依赖项。

服务器代码

创建名为 server.py：的文件

"""AG-UI server example."""

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 azure.identity import AzureCliCredential
from fastapi import FastAPI

# Read required configuration
endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT")
deployment_name = os.environ.get("AZURE_OPENAI_CHAT_COMPLETION_MODEL")

if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT environment variable is required")
if not deployment_name:
    raise ValueError("AZURE_OPENAI_CHAT_COMPLETION_MODEL environment variable is required")

chat_client = OpenAIChatCompletionClient(
    model=deployment_name,
    azure_endpoint=endpoint,
    api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
    credential=AzureCliCredential(),
)

# Create the AI agent
agent = Agent(
    name="AGUIAssistant",
    instructions="You are a helpful assistant.",
    client=chat_client,
)

# Create FastAPI app
app = FastAPI(title="AG-UI Server")

# Register the AG-UI endpoint
add_agent_framework_fastapi_endpoint(app, agent, "/", single_user=True)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8888)

关键概念

add_agent_framework_fastapi_endpoint：注册 AG-UI 端点，具有自动请求/响应处理功能和 SSE 流式传输功能
Agent：将处理传入请求的 Agent Framework 智能体
FastAPI 集成：使用 FastAPI 的本机异步支持进行响应流式处理
说明：代理是使用默认指令创建的，客户端消息可以覆盖默认指令
配置： OpenAIChatCompletionClient 接受显式 Azure 路由输入，例如 model、 azure_endpoint、 api_version和 credential，还可以从环境变量中读取

配置并运行服务器

设置所需的环境变量：

export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
export AZURE_OPENAI_CHAT_COMPLETION_MODEL="gpt-4o-mini"

运行服务器：

python server.py

或者直接使用 uvicorn：

uvicorn server:app --host 127.0.0.1 --port 8888

服务器将开始侦听 http://127.0.0.1:8888。

步骤 2：创建 AG-UI 客户端

AG-UI 客户端连接到远程服务器并显示流式响应。

AGUIChatClient 重用内部拥有的 HTTP 客户端，该客户端不保留响应 Cookie。

如果服务器需要 Cookie 来实现身份验证、会话或负载均衡器相关性，请通过 http_client= 传递由调用方所有的 httpx.AsyncClient。将客户端限定为一个经过身份验证的主体，并在应用程序中将其关闭。 AG-UI 线程 ID 是关联标识符，而不是身份验证边界。

安装所需程序包

已安装 AG-UI 包，其中包括 AGUIChatClient：

# Already installed with agent-framework-ag-ui
pip install agent-framework-ag-ui

客户端代码

创建名为 client.py：的文件

"""AG-UI client example."""

import asyncio
import os

from agent_framework import Agent
from agent_framework_ag_ui import AGUIChatClient


async def main():
    """Main client loop."""
    # Get server URL from environment or use default
    server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:8888/")
    print(f"Connecting to AG-UI server at: {server_url}\n")

    # Create AG-UI chat client
    chat_client = AGUIChatClient(endpoint=server_url)

    # Create agent with the chat client
    agent = Agent(
        name="ClientAgent",
        client=chat_client,
        instructions="You are a helpful assistant.",
    )

    # Get a thread for conversation continuity
    thread = agent.create_session()

    try:
        while True:
            # Get user input
            message = input("\nUser (:q or quit to exit): ")
            if not message.strip():
                print("Request cannot be empty.")
                continue

            if message.lower() in (":q", "quit"):
                break

            # Stream the agent response
            print("\nAssistant: ", end="", flush=True)
            async for update in agent.run(message, session=thread, stream=True):
                # Print text content as it streams
                if update.text:
                    print(f"\033[96m{update.text}\033[0m", end="", flush=True)

            print("\n")

    except KeyboardInterrupt:
        print("\n\nExiting...")
    except Exception as e:
        print(f"\n\033[91mAn error occurred: {e}\033[0m")


if __name__ == "__main__":
    asyncio.run(main())

发送多模式消息

AGUIChatClient 保留用户消息中的文本和媒体部件的顺序。使用 MIME 类型标识 URI 或内联图像、音频、视频和文档内容：

from pathlib import Path

from agent_framework import Content, Message

message = Message(
    role="user",
    contents=[
        Content.from_text("Compare this image with the attached brief."),
        Content.from_uri(
            "https://example.com/product.png",
            media_type="image/png",
        ),
        Content.from_data(
            Path("brief.pdf").read_bytes(),
            media_type="application/pdf",
        ),
    ],
)

await agent.run(message, session=thread)

新的入站媒体 URI 必须使用 http、https 或 data 方案。还支持内联二进制数据。在以 HTTPS、数据 URI 或内联内容的形式发送应用程序代码之前，在应用程序代码中解析并授权 file、FTP、相对、自定义方案以及协议级 id 或 file 引用。

当持久化线程快照包含较旧的不受支持的媒体引用时，Python适配器会在模型重播之前将其替换为无效占位符文本，并在下次保存快照时删除引用。

关键概念

Server-Sent 事件（SSE）：协议使用 SSE 格式（data: {json}\n\n）
事件类型：不同的事件提供元数据和内容（大写字母加下划线）：
- RUN_STARTED：代理已开始处理
- TEXT_MESSAGE_START：来自代理的短信开头
- TEXT_MESSAGE_CONTENT：从代理流式传输的增量文本（包含 delta 字段）
- TEXT_MESSAGE_END：短信结束
- RUN_FINISHED：成功完成
- RUN_ERROR：错误信息
字段命名：事件字段使用 camelCase（例如， threadId、 runId、 messageId）
线程管理：在跨 threadId 的请求中维护会话上下文
客户端说明：系统消息从客户端发送

配置并运行客户端

（可选）设置自定义服务器 URL：

export AGUI_SERVER_URL="http://127.0.0.1:8888/"

运行客户端（在单独的终端中）：

python client.py

步骤 3：测试完整系统

运行服务器和客户端后，现在可以测试完整的系统。

预期输出

$ python client.py
Connecting to AG-UI server at: http://127.0.0.1:8888/

User (:q or quit to exit): What is 2 + 2?

[Run Started - Thread: abc123, Run: xyz789]
2 + 2 equals 4.
[Run Finished - Thread: abc123, Run: xyz789]

User (:q or quit to exit): Tell me a fun fact about space

[Run Started - Thread: abc123, Run: def456]
Here's a fun fact: A day on Venus is longer than its year! Venus takes
about 243 Earth days to rotate once on its axis, but only about 225 Earth
days to orbit the Sun.
[Run Finished - Thread: abc123, Run: def456]

User (:q or quit to exit): :q

颜色编码输出

客户端显示具有不同颜色的不同内容类型：

黄色：运行开始通知
蓝绿色：代理文本响应（实时流式传输）
绿色：运行完成通知
红色：错误消息

使用 curl 进行测试（可选）

在运行客户端之前，可以使用 curl 手动测试服务器：

curl -N http://127.0.0.1:8888/ \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "messages": [
      {"role": "user", "content": "What is 2 + 2?"}
    ]
  }'

您应该会看到 Server-Sent Events 传回：

data: {"type":"RUN_STARTED","threadId":"...","runId":"..."}

data: {"type":"TEXT_MESSAGE_START","messageId":"...","role":"assistant"}

data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":"The"}

data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":" answer"}

...

data: {"type":"TEXT_MESSAGE_END","messageId":"..."}

data: {"type":"RUN_FINISHED","threadId":"...","runId":"..."}

对于处于空闲状态的流，curl 还可能显示 : keepalive 注释行。这些是 SSE 传输注释，而不是 AG-UI 事件。

工作原理

服务器端流程

客户端发送包含消息的 HTTP POST 请求
FastAPI 终结点接收请求
AgentFrameworkAgent 封装器管理执行
代理使用代理框架处理消息
AgentFrameworkEventBridge 将代理更新转换为 AG-UI 事件
响应作为服务器发送事件（SSE）被流式传输回
运行完成后，连接将关闭

客户端流

客户端将 HTTP POST 请求发送到服务器终结点
服务器使用 SSE 流进行响应
客户端将传入 data: 行分析为 JSON 事件
每个事件都根据其类型显示
threadId 会被捕获以保持对话的连续性
当 RUN_FINISHED 事件到达时，流传输完成

协议详细信息

AG-UI 协议使用：

用于发送请求的 HTTP POST
Server-Sent Events (SSE) 来流式处理响应
使用 JSON 进行事件序列化
用于维护会话上下文的线程 ID
运行 ID 来跟踪单个执行
事件类型命名：大写字母加下划线（例如 RUN_STARTED、TEXT_MESSAGE_CONTENT）
字段命名：camelCase（例如，threadId、runId、messageId）
当流处于空闲状态时，每 15 秒发送一次 SSE 保持连接注释。只处理 data: 行的客户端会自动忽略这些注释。

常见模式

添加 A2UI 交互界面

A2UI 使代理能够生成交互界面，并在响应流传输过程中由兼容的 AG-UI 客户端将其渲染出来。安装可选的 A2UI 依赖项：

pip install "agent-framework-ag-ui[a2ui]"

对于 uv，请运行 uv pip install "agent-framework-ag-ui[a2ui]"。

若要为某个终结点默认启用 A2UI，请传递 a2ui_config 并包含后端选择加入：

from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint

add_agent_framework_fastapi_endpoint(
    app=app,
    agent=agent,
    path="/a2ui",
    a2ui_config={"inject_a2ui_tool": True},
    single_user=True,
)

前端需要 AG-UI A2UI 中间件或具备等效的行为，以便：

在 AG-UI 请求上下文中提供其组件目录和生成指南。
将流式传输的 render_a2ui 工具参数呈现为界面更新。
为请求启用 A2UI 时设置 forwardedProps.injectA2UITool 。显式 false 值会覆盖该请求的后端选择加入设置。

适配器将客户端提供的组件目录和生成指南作为不受信任的用户数据提供给模型。在由服务器控制的代理指令或配置中保留受信任的策略和呈现说明。

对为 A2UI 端点提供服务的代理使用 OpenAIChatCompletionClient。它以流式方式传输工具调用参数的增量更新，以便客户端能够逐步渲染界面。响应 API 客户端不支持此渐进式呈现流。

适配器根据提供的目录验证已完成的组件树。如果验证失败，则会将验证错误添加到生成提示，并根据配置重试 recovery 。当恢复尝试全部失败后，客户端不会收到异常，而是会收到一个失败信封。

有关完整的实现，请参阅 Agent Framework 存储库中的 Python A2UI 代理和 A2UI 终结点设置。

自定义服务器配置

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Add CORS for web clients
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

add_agent_framework_fastapi_endpoint(
    app,
    agent,
    "/agent",
    keepalive_seconds=30,  # Defaults to 15; set to None to disable
    single_user=True,
)

keepalive_seconds 必须是正数或 None。

断开连接后继续有限次运行

默认情况下，断开 SSE 客户端的连接可以取消响应生成器并停止运行。当有限代理或工作流运行必须持续足够长的时间，以完成快照、检查点或审批状态的最终处理时，请设置 detached_runs=True。

add_agent_framework_fastapi_endpoint(
    app,
    agent,
    "/agent",
    detached_runs=True,
    max_detached_runs=32,
    single_user=True,
    detached_run_timeout_seconds=3600,
)

配置快照持久化时，还要提供可信的 snapshot_scope_resolver。重新连接客户端可恢复最新提交的线程快照;分离执行不会重播在未附加客户端时发出的 SSE 事件。

每个终结点注册最多可保留 max_detached_runs 个生产者，并在达到该容量上限时返回 HTTP 503。当分离运行处于活动状态时，对于针对同一 (Snapshot Scope, threadId) 的另一个修改请求，它会返回 HTTP 409。这种协调是进程本地的，因此具有多个辅助角色的应用程序必须提供分布式序列化。

FastAPI 依赖项和其他请求作用域内的资源可能在断开连接的响应结束后被释放。在运行开始前，先确定授权、快照范围及其他持久性值，并且不要在独立的工具和提供程序中保留请求所属的客户端或会话。

多个代理

app = FastAPI()

weather_agent = Agent(name="weather", ...)
finance_agent = Agent(name="finance", ...)

add_agent_framework_fastapi_endpoint(app, weather_agent, "/weather", single_user=True)
add_agent_framework_fastapi_endpoint(app, finance_agent, "/finance", single_user=True)

这些示例之所以使用 single_user=True ，是因为它们在本地运行并保留一个授权边界。对于共享服务，请将其替换为可从经过身份验证的请求上下文派生稳定作用域的 authorization_scope_resolver。

错误处理

try:
    async for event in client.send_message(message):
        if event.get("type") == "RUN_ERROR":
            error_msg = event.get("message", "Unknown error")
            print(f"Error: {error_msg}")
            # Handle error appropriately
except httpx.HTTPError as e:
    print(f"HTTP error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

AGUIChatClient 可以在没有 TOOL_CALL_END 事件的情况下执行完整的本地 client-tool 调用。如果 RUN_ERROR 在客户端公开本地工具调用之后到达，客户端会引发 ResponseInvalidatedException 并放弃调用，而不是从无效响应执行副作用。

Troubleshooting

连接被拒绝

在启动客户端之前，请确保服务器正在运行：

# Terminal 1
python server.py

# Terminal 2 (after server starts)
python client.py

身份验证错误

确保使用 Azure 进行身份验证：

az login

验证 Azure OpenAI 资源上是否分配了正确的角色。

流媒体无法播放

检查客户端超时时间是否足够：

httpx.AsyncClient(timeout=60.0)  # 60 seconds should be enough

对于长期运行的智能体，相应延长超时时间。

默认情况下，空闲的流每 15 秒发送一次 SSE 保持连接注释。如果代理更早关闭空闲连接，请在注册终结点时将 keepalive_seconds 配置为一个较小的正值。

线程上下文丢失

客户端会自动管理线程连续性。如果上下文丢失：

检查是否正在从 RUN_STARTED 事件中捕获 threadId
确保跨消息使用相同的客户端实例
验证服务器是否在后续请求中接收thread_id

后续步骤

了解 AG-UI 的基础知识后，可以：

添加后端工具：为域创建自定义函数工具

其他资源

Go 通过 provider/aguiprovider 同时支持服务器端和客户端的 AG-UI。

import "github.com/microsoft/agent-framework-go/provider/aguiprovider"

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(myAgent, aguiprovider.HandlerConfig{}))

if err := http.ListenAndServe(":8888", mux); err != nil {
    log.Fatal(err)
}

当 Go 应用需要将 AG-UI 服务器作为代理调用时使用 aguiprovider.NewAgent ：

import aguiSSEClient "github.com/ag-ui-protocol/ag-ui/sdks/community/go/pkg/client/sse"

a := aguiprovider.NewAgent(
    aguiSSEClient.NewClient(aguiSSEClient.Config{Endpoint: serverURL}),
    aguiprovider.AgentConfig{},
)

Tip

有关完整的可运行示例，请参阅 AG-UI 入门服务器和客户端示例。

反馈

此页面是否有帮助？

上次更新时间为 2026-10-23

AG-UI 入门

您将构建什么

Prerequisites

创建 AG-UI 服务器

使用 .NET 客户端进行连接

对话连续性

后续步骤

相关资源

Prerequisites

步骤 1：创建 AG-UI 服务器

安装所需的包

服务器代码

关键概念

配置并运行服务器

步骤 2：创建 AG-UI 客户端

安装所需程序包

客户端代码

发送多模式消息

关键概念

配置并运行客户端

步骤 3：测试完整系统

预期输出

颜色编码输出

使用 curl 进行测试（可选）

工作原理

服务器端流程

客户端流

协议详细信息

常见模式

添加 A2UI 交互界面

自定义服务器配置

断开连接后继续有限次运行

多个代理

错误处理

Troubleshooting

连接被拒绝

身份验证错误

流媒体无法播放

线程上下文丢失

后续步骤

其他资源

反馈

其他资源