大AIエージェント時代！AG-UI準拠でアプリを構築する

こんにちは。最近、AI将棋モデルの対局解説動画を見るのにハマっている大塚です。
将棋は全然わからないのですが、AIの予想外の手に驚いている解説を聞くのが結構面白いんですよね。

さて今回は、Agent–User Interaction（AG-UI）がどのような課題を解決するプロトコルなのかを確認します。
また、CopilotKitというフレームワークを利用して、AG-UIに準拠したアプリをどのように作成できるのかも試してみます。

1. AG-UIとは

AG-UIは、2025年4月にCopilotKitチームが提唱した、AIエージェントとUIの通信プロトコル（やり取りの形式）のことです。

github.com

AIエージェントとUIの間で、イベントを用いてやり取りすることで情報の連携をします。
イベントの例としては以下のようなものが定義されています。

• テキスト生成の開始/途中/完了を表すイベント
• ツール呼び出し開始/引数/結果のイベント
• UIに反映すべき状態更新の通知
• Human-in-the-loopを前提としたユーザーからのアクションイベント

こういったイベントなどの定義が、AG-UIによって標準化されたことで、
AIエージェントの開発にどのようなフレームワークを採用しているかにかかわらず、UIとの接続が容易になります。

1.1. AG-UIが解決しようとしている課題感

AIエージェントを開発するためのフレームワークがさまざまな会社から発表されており、それぞれが独自の形式を採用しているという状態です。

AIエージェントが当たり前の時代になれば、バックエンドのエージェント開発に使われるフレームワークも多様になるでしょう。
例えば、AチームではXというフレームワークを利用していて、BチームではYというフレームワークを利用している、といった形です。

その際に、標準化されたプロトコルがなければ、エージェントからのレスポンスをパースする側の実装は大変になります。
フレームワークごとに返してくるレスポンスの形式が異なるからです。

AG-UIがない場合は、以下のようにそれぞれのイベント形式についてパースする処理を実装する必要があります。

AG-UIがない場合

一方で、AG-UIがある場合は、フレームワークごとの差分を吸収してくれる層ができるので、AG-UIのイベントをパースするだけで済みます。

AG-UIがある場合

1.2. 他エージェントプロトコル（MCP・A2A）との違い

AIエージェント関連のプロトコルとして、AG-UI以外にもMCPやA2Aがあります。
これらはいずれもAIエージェントに関するプロトコルですが、それぞれ補完的な関係性になっています。

AIエージェント関連のプロトコル（https://github.com/ag-ui-protocol/ag-ui から引用）

プロトコル	標準化する対象
MCP	AIエージェントのツール利用
A2A	AIエージェント間の通信方法
AG-UI	AIエージェントとUI間のやり取り

また、AIエージェントが動的にUIを生成するGenerative UIの仕様も、標準化が進み始めています。

例えば、Googleが提唱しているA2UIなどです。
名前は似ていますが、AG-UIは「テキスト生成やツール呼び出しといったイベントをどのような形式でやり取りするか」を定めたプロトコルである一方、A2UIは「エージェントが画面のコンポーネント自体を動的に生成する」ための仕様であり、関心事が異なります。

A2UIに関しては、以前ブログに書いているので、よければ参照してください。

acro-engineer.hatenablog.com

1.3. AG-UIが標準化している主なイベント

前述した通り、AG-UI は、AIエージェントとUIの間の通信をイベントストリームとして扱うプロトコルです。
従来の「1リクエスト→1レスポンス」ではなく、逐次的にイベントが流れることでリアルタイム性を実現します。

AG-UIでは、イベントフローの形式として、以下の3つのパターンが定義されています。

パターン名	説明	主なイベント
Start-Content-End パターン	ストリーミングコンテンツ（テキストやツール呼び出し）に使用される	Start / Content / End
Snapshot-Delta パターン	状態同期のために使用される	Snapshot / Delta
Lifecycle パターン	エージェントの実行状況の監視に使用される	Started / Finished / Error

例えば、Start-Content-End パターンであるテキストメッセージイベントでは、以下のように TextMessageStart → TextMessageContent → TextMessageEnd という順序でイベントが発行されます。

テキストメッセージイベントのフロー（https://docs.ag-ui.com/concepts/events#text-message-events から引用）

テキストメッセージイベントの他にも、AG-UIでは、以下のようなイベントタイプが定義されています。
UI側は、AIエージェントフレームワーク独自のイベントをパースする必要はなく、これらのイベントを処理できれば良いことになります。

カテゴリ	説明	対応するフローパターン
ライフサイクルイベント	エージェントの実行進行状況を監視する	Lifecycle
テキストメッセージイベント	ストリーミングされるテキストコンテンツを処理する	Start-Content-End
ツール呼び出しイベント	エージェントによるツール実行を管理する	Start-Content-End
ステート管理イベント	エージェントとUI間の状態を同期する	Snapshot-Delta
アクティビティイベント	進行中のアクティビティの進捗を表す	Snapshot-Delta / Lifecycle

2. AG-UIを利用する

AG-UIのありがたみを理解するために、まずはAG-UIを利用しない場合、どのような実装になるのかを確認します。

2.1. AG-UIを利用しない場合

ここでは、AIエージェントフレームワークとしてStrands AgentsとGoogle ADKを利用してみます。 フロントエンドはReactで作成します。

AG-UIを利用しない場合の概要図

LLMのテキスト出力とツール実行出力をイベントとして返すことを考えます。

Strands Agentsを利用したバックエンド実装

@tool
def get_weather() -> str:
    return "sunny"


weather_agent = Agent(tools=[get_weather], callback_handler=None)

app = FastAPI()

@app.post("/")
async def run(request: Request) -> StreamingResponse:
    payload = await request.json()
    prompt = payload.get("prompt")
    if not prompt:
        raise HTTPException(status_code=400, detail="Missing 'prompt' in request body")

    async def event_stream():
        async for event in weather_agent.stream_async(prompt):
            if isinstance(event, dict):
                if "data" in event:
                    yield json.dumps({"data": event["data"]}, ensure_ascii=False) + "\n"
                elif "current_tool_use" in event:
                    yield json.dumps(
                        {"current_tool_use": event["current_tool_use"]},
                        ensure_ascii=False,
                    ) + "\n"

    return StreamingResponse(event_stream(), media_type="application/x-ndjson")

Google ADKを利用したバックエンド実装

@tool
def get_weather() -> str:
    return "sunny"


root_agent = Agent(
    model="gemini-3-flash-preview",
    name="root_agent",
    instruction="Answer user questions to the best of your knowledge",
    tools=[get_weather],
)

session_service = InMemorySessionService()
runner = Runner(
    agent=root_agent,
    app_name="sample-adk",
    session_service=session_service,
    auto_create_session=True,
)

app = FastAPI()

@app.post("/")
async def run(request: Request) -> StreamingResponse:
    payload = await request.json()
    prompt = payload.get("prompt")
    if not prompt:
        raise HTTPException(status_code=400, detail="Missing 'prompt' in request body")

    user_id = payload.get("user_id") or f"user_{uuid4().hex}"
    session_id = payload.get("session_id") or f"session_{uuid4().hex}"

    async def event_stream():
        user_content = types.Content(role="user", parts=[types.Part(text=prompt)])
        async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=user_content,
            run_config=RunConfig(streaming_mode=StreamingMode.SSE, max_llm_calls=50),
        ):
            calls = event.get_function_calls()
            if calls:
                for call in calls:
                    yield (
                        json.dumps(
                            {
                                "function_call": {
                                    "name": call.name,
                                    "args": call.args,
                                }
                            },
                            ensure_ascii=False,
                        )
                        + "\n"
                    )

            responses = event.get_function_responses()
            if responses:
                for resp in responses:
                    yield (
                        json.dumps(
                            {
                                "function_response": {
                                    "name": resp.name,
                                    "response": resp.response,
                                }
                            },
                            ensure_ascii=False,
                        )
                        + "\n"
                    )

            if event.content and event.content.parts:
                for part in event.content.parts:
                    text = getattr(part, "text", None)
                    if text:
                        yield (
                            json.dumps(
                                {
                                    "text": text,
                                    "partial": bool(getattr(event, "partial", False)),
                                },
                                ensure_ascii=False,
                            )
                            + "\n"
                        )

    return StreamingResponse(
        event_stream(),
        media_type="application/x-ndjson",
        headers={"X-User-Id": user_id, "X-Session-Id": session_id},
    )

フロントエンドのイベントパース処理

AG-UIのような標準化プロトコルを使わずにイベントを処理する場合、フロントエンド側の実装は、例えば以下のように個別のイベント形式に対して実装を追加することになります。 （もしくはバックエンド側でフレームワーク間の差分を吸収します）

// Strands Agents用: ストリームを読み取って onEvent を呼び出す
const parseStrandsStream = async (response, onEvent) => {
  for await (const line of parseNdjson(response)) {
    const event = JSON.parse(line);
    if (typeof event?.data === "string") {
      onEvent({ kind: "text", text: event.data });
    } else if (typeof event?.data?.text === "string") {
      onEvent({ kind: "text", text: event.data.text });
    } else if (typeof event?.text === "string") {
      onEvent({ kind: "text", text: event.text });
    } else if (event?.current_tool_use) {
      onEvent({ kind: "tool_call", tool: event.current_tool_use });
    }
  }
};

// Google ADK用: ストリームを読み取って onEvent を呼び出す
const parseAdkStream = async (response, onEvent) => {
  for await (const line of parseNdjson(response)) {
    const event = JSON.parse(line);
    if (event?.function_call) {
      onEvent({ kind: "tool_call", tool: event.function_call });
    } else if (event?.function_response) {
      onEvent({ kind: "tool_result", result: event.function_response });
    } else if (typeof event?.text === "string") {
      onEvent({ kind: "text", text: event.text });
    }
  }
};

// 利用側: フレームワークごとに呼び分けが必要
if (framework === "strands") {
  await parseStrandsStream(response, onEvent);
} else if (framework === "adk") {
  await parseAdkStream(response, onEvent);
}

これでは、AIエージェントのフレームワークごとに個別の実装が必要で、新しいフレームワークを追加するたびに対応処理を増やさなければなりません。

2.2. AG-UIを利用する場合

AG-UIを用いることで、「AG-UIを利用しない場合」で紹介したような個別実装をする必要がなくなります。

Strands Agentsを利用する場合、AG-UIが公式提供するライブラリ ag_ui_strands に create_strands_app などの便利関数が用意されているため、AG-UI対応のFastAPIサーバーを簡単に作成できます。

StrandsAgent がAG-UIが定義しているイベント形式に変換する役割を担っています。

Strands Agentsを利用したバックエンド実装

from strands import Agent, tool
from ag_ui_strands import StrandsAgent, create_strands_app


@tool
def get_weather() -> str:
    return "sunny"


weather_agent = Agent(tools=[get_weather], callback_handler=None)

# StrandsAgentでラップすることで、Strands Agentsのイベントを
# AG-UIが定義するイベント形式に変換して返すようになる
agui_agent = StrandsAgent(
    agent=weather_agent,
    name="WeatherAgent",
    description="今日の天気を返します。",
)

app = create_strands_app(agui_agent, path="/", ping_path="/ping")

Google ADKを利用したバックエンド実装

from ag_ui_adk import ADKAgent, create_adk_app
from google.adk.agents import Agent
from google.adk.tools import tool


@tool
def get_weather() -> str:
    return "sunny"


root_agent = Agent(
    model="gemini-3-flash-preview",
    name="root_agent",
    instruction="Answer user questions to the best of your knowledge",
    tools=[get_weather],
)

agui_agent = ADKAgent(
    adk_agent=root_agent,
    app_name="sample-adk",
    user_id="local-user",
)

app = create_adk_app(agui_agent, path="/")

フロントエンドのイベントパース処理

イベントを受け取る側は、以下のように、AG-UIのイベント形式のみを意識すればよく、AIエージェントフレームワークごとの個別対応をする必要がなくなりました。

const parseAguiStream = async (response, onEvent) => {
  await parseSseStream(response, (event) => {
    const type = event?.type;
    if (!type) return;

    if (type === "TEXT_MESSAGE_START") {
      onEvent({
        kind: "text_start",
        messageId: event.messageId,
        role: event.role,
      });
      return;
    }

    if (type === "TEXT_MESSAGE_CONTENT" || type === "TEXT_MESSAGE_CHUNK") {
      if (typeof event.delta !== "string") return;
      onEvent({
        kind: "text",
        messageId: event.messageId,
        role: event.role,
        text: event.delta,
      });
      return;
    }

    ...

  });
};

// 利用側: フレームワーク問わず同じ関数を使うだけ
await parseAguiStream(response, onEvent);

AG-UIを使うことで、parseAguiStream 一本に統一できます。
新しいフレームワークをバックエンドに追加した場合でも、フロントエンド側の変更は不要です。

3. CopilotKitとは

CopilotKitとは、AIエージェントを利用するフロントエンドアプリを作成するためのフレームワークです。 Generative UIやHuman-in-the-loopを備えたチャットアプリなども簡単に作成できます。

www.copilotkit.ai

また、AG-UIを提唱した企業がCopilotKitを作成しているため、AG-UIもサポートされています。

CopilotKit概要図（https://www.copilotkit.ai/ から引用）

3.1. CopilotKitを利用してチャットアプリを作成する

以下のコマンドで、UI・AIエージェント両方のテンプレートが生成されます。

npx copilotkit@latest create -f aws-strands-py

以下のようなディレクトリ構造で生成されます。
src がフロントエンド、 agent がバックエンドの実装になっています。
※デフォルトだとOpenAIのAPIキーを利用して、GPTモデルを呼び出す実装になっています。

$ tree
.
├── agent
│   ├── main.py
│   ├── pyproject.toml
│   └── uv.lock
├── eslint.config.mjs
├── LICENSE
├── next.config.ts
├── package.json
├── postcss.config.mjs
├── public
│   ├── file.svg
│   ├── globe.svg
│   ├── next.svg
│   ├── vercel.svg
│   └── window.svg
├── README.md
├── scripts
│   ├── run-agent.bat
│   ├── run-agent.sh
│   ├── setup-agent.bat
│   └── setup-agent.sh
├── src
│   ├── app
│   │   ├── api
│   │   │   └── copilotkit
│   │   │       └── route.ts
│   │   ├── favicon.ico
│   │   ├── globals.css
│   │   ├── layout.tsx
│   │   └── page.tsx
│   └── components
│       ├── default-tool-ui.tsx
│       └── weather.tsx
└── tsconfig.json

npm run dev を実行した後、 localhost:3000 にアクセスすることで、以下のようなチャット画面が表示されます。
試しに「こんにちは。」と会話すると、たしかにAG-UIで定義されている形で、イベントが返ってきていることを確認できました。

CopilotKitチャット画面

CopilotKitはフレームワーク内部で2.2.で実施したようなAG-UIのイベント購読処理を行っています。
そのため、フレームワーク利用者はCopilotKitが提供するコンポーネントを宣言するだけで、AIエージェントから受け取るイベントを画面に表示できます。

具体的には、以下のように CopilotKit コンポーネントを返すだけです。
CopilotKitを使わない場合はAG-UIの各イベントに対応するように自前で実装する必要があると考えると、実装がいかにシンプルになるかがわかりますね。

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

export default function RootLayout({
  children,
}: Readonly<{
  children: React.ReactNode;
}>) {
  return (
    <html lang="en">
      <body className={"antialiased"}>
        <CopilotKit runtimeUrl="/api/copilotkit" agent="strands_agent">
          {children}
        </CopilotKit>
      </body>
    </html>
  );
}

3.2. Human-in-the-loop（HITL）の簡単なサンプルを作成する

HITLとは、AIが処理を行うワークフローの中で人が介入し、判断を加えることで、正確性や安全性を確保する仕組みです。

ここでは、カスタマーサポートの返信内容を生成してくれるエージェンティックアプリを作成してみようと思います。
カスタマーサポート業務には、顧客からのクレーム対応などが含まれるため、AIエージェントに任せきりにせず、人が最終確認を行うことが望ましいです。

一方で、すべてのケースでHITLが必要なわけではありません。 例えばFAQのように、参照する情報源が明確で回答内容が定型的な場合は、AIエージェントのみで自動応答しても問題ないケースもあります。

以下のような業務フローを想定してサンプルを作成します。

カスタマーサポート業務フロー

実装

CopilotKitでHITLを実現するには、「フロントエンドツール」という仕組みを使います。
フロントエンドツールとは、AIエージェントが呼び出すツールのうち、実際の処理をフロントエンド側で行うものです。

AIがこのツールを呼び出すと、AG-UIのツール呼び出しイベントとしてフロントエンドに届き、
CopilotKitの useHumanInTheLoop フックがそれを検知して、人間が操作するためのUIをチャット内にインラインで表示します。

大まかな流れは以下のようになります。

1. バックエンドに「フロントエンドツール」を定義する
   • 通常のツールと同様に定義するが、処理は return None のみ（実際の処理はフロントエンド側）
2. AIエージェントがフロントエンドツールを呼び出す
   • ツール呼び出しイベントがAG-UIの形式でフロントエンドに届く
3. useHumanInTheLoop フックでイベントを捕捉する
   • ツール名を指定しておくと、AIがそのツールを呼び出したタイミングで render 関数が実行される
4. カスタムUIが表示され、人間が判断を入力する
   • respond() を呼ぶことで結果をバックエンドに返送
5. エージェントがツール結果を受け取り、処理を継続・完了する

シーケンス図

AIエージェントにHITL用のフロントエンドツールを登録する

return None のみのフロントエンドツールを定義し、システムプロンプトでAIエージェントがそのツールを呼び出すよう指示します。

"""Customer Support Reply Generator - Human in the Loop Demo.

HITL flow:
1. User inputs a customer message in the chat
2. AI generates a professional draft reply
3. AI calls present_draft_reply tool (frontend tool)
4. Frontend shows approval/edit UI inline in the chat
5. Operator approves (sends as-is) or edits then sends (both mocked)
6. Agent confirms the action in chat
"""

import os

from ag_ui_strands import (
    StrandsAgent,
    create_strands_app,
)
from dotenv import load_dotenv
from strands import Agent, tool

load_dotenv()


@tool
def present_draft_reply(customer_message: str, draft_reply: str):
    """Present a draft customer support reply to the human operator for review.

    This is a frontend tool. The operator will review the draft in the UI
    and either approve it or edit it before sending to the customer.

    Args:
        customer_message: The original customer inquiry text
        draft_reply: The AI-generated draft reply to present for review

    Returns:
        None - execution is handled by the frontend
    """
    return None


system_prompt = """You are a customer support assistant that drafts professional replies to customer inquiries.

When the user provides a customer message:
1. Carefully read the customer's inquiry
2. Generate a professional, empathetic, and helpful draft reply in the same language as the customer message
3. Call the present_draft_reply tool with the original customer_message and your draft_reply
4. After the tool returns with the operator's decision, do NOT send any confirmation message. The UI already shows the result to the operator.
"""

strands_agent = Agent(
    system_prompt=system_prompt,
    tools=[present_draft_reply],
)

agui_agent = StrandsAgent(
    agent=strands_agent,
    name="strands_agent",
    description="Customer support reply generator with human-in-the-loop review",
)

app = create_strands_app(agui_agent, "/")

フロントエンド側でツールが実行されたときのコールバックを設定する

useHumanInTheLoop は、バックエンドで定義したフロントエンドツールと、フロントエンドのUI処理を紐付けるためのフックです。
name にバックエンドのツール名と同じ名前を指定することで、AIがそのツールを呼び出した際に render 関数が実行され、チャット内にカスタムUIが表示されます。

"use client";

import { useHumanInTheLoop } from "@copilotkit/react-core";
import { CopilotChat } from "@copilotkit/react-ui";
import { useState } from "react";

export default function CustomerSupportPage() {
  // useHumanInTheLoop: バックエンドのフロントエンドツールとUIを紐付けるフック
  // AIがツールを呼び出すと render 関数が実行され、チャット内にカスタムUIが表示される
  useHumanInTheLoop({
    // バックエンドで定義したツール名と一致させる（AIはこの名前でツールを呼び出す）
    name: "present_draft_reply",
    description:
      "Present an AI-generated draft reply to the human operator for approval or editing before sending.",
    // バックエンドのツール定義と同じパラメーターを宣言する
    // ここで宣言したパラメーターが render の args として受け取れる
    parameters: [
      {
        name: "customer_message",
        type: "string",
        description: "The original customer inquiry message",
        required: true,
      },
      {
        name: "draft_reply",
        type: "string",
        description: "The AI-generated draft reply",
        required: true,
      },
    ],
    // AIがツールを呼び出したときに実行されるコールバック
    // args:    AIがツールに渡した引数（バックエンドのツール呼び出し時の値）
    // respond: 人間の判断結果をバックエンドに返す関数（呼ぶとエージェントが処理を再開する）
    // status:  "inProgress"（人間が操作中）| "complete"（respond が呼ばれた後）
    render({ args, respond, status }) {
      return (
        <DraftReviewPanel
          customerMessage={args.customer_message ?? ""}
          draftReply={args.draft_reply ?? ""}
          status={status === "complete" ? "complete" : "inProgress"}
          // respond を渡すことで DraftReviewPanel 内から結果をバックエンドへ返せる
          respond={respond}
        />
      );
    },
  });

  return (
    <main className="h-screen flex flex-col bg-slate-50">
      <CopilotChat
        labels={{
          title: "カスタマーサポート返信アシスタント",
          initial:
            "顧客からのお問い合わせメッセージを入力してください。AIが返信案を作成し、送信前に確認できます。",
        }}
        className="flex-1 max-w-3xl w-full mx-auto"
      />
    </main>
  );
}

実際に動かすと、AIが返信案を生成した後、チャット内にインラインで承認UIが表示されます。

HITLでの承認UI

「承認して送信」を押すと、完了状態に切り替わりエージェントが処理を継続します。

承認した場合のUI

「編集する」を押すと、返信案を自由に編集してから送信できます。

編集した場合のUI

AIエージェントフレームワークをStrands AgentsからGoogle ADKに変えたいとなった場合でも、画面側のロジックを修正することなく実施できました。

Google ADKを利用した場合

4. まとめ

AG-UIによって、AIエージェントフレームワーク間の差分が吸収されるため、フロントエンド側はAG-UIのイベント形式だけを意識すればよくなります。
また、CopilotKitを利用することで、AG-UIのイベント購読処理を自前で実装することなく、エージェンティックアプリを簡単に作成できました。
さらに、useHumanInTheLoop フックを使うことで、AIの出力を人間が確認・修正してから処理を継続するHITLの仕組みも比較的シンプルに実装できました。

CopilotKitはReact以外のサポートが薄い点が惜しいですが、AG-UIに準拠したエージェンティックアプリを素早く作成したい場合には有力な選択肢になりそうです。

Acroquest Technologyでは、キャリア採用を行っています。

• Azure OpenAI／Amazon Bedrock等を使った生成AIソリューションの開発
• ディープラーニング等を使った自然言語／画像／音声／動画解析の研究開発
• マイクロサービス、DevOps、最新のOSSやクラウドサービスを利用する開発プロジェクト
• 書籍・雑誌等の執筆や、社内外での技術の発信・共有によるエンジニアとしての成長

少しでも上記に興味を持たれた方は、是非以下のページをご覧ください。

www.wantedly.com