ヒューマンインザループとのAG-UI

MAFツール承認の担当者は、ツールに承認が必要かどうかを判断する責任を引き続き負います。 AG-UI は、承認要求をクライアントに転送し、クライアントの決定をサーバーに戻します。

承認ポリシー、条件付きルール、および一般的な安全ガイダンスについては、「人間のループ内承認で関数ツールを使用する」を参照してください。

承認を要求

MAF 関数を ApprovalRequiredAIFunction でラップし、エージェントを通常どおりに公開します。

using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

AIFunction deleteFile = AIFunctionFactory.Create(
    (string path) => $"Deleted {path}",
    name: "delete_file",
    description: "Delete a file.");

AITool approvalRequiredTool = new ApprovalRequiredAIFunction(deleteFile);
AIAgent agent = chatClient.AsAIAgent(tools: [approvalRequiredTool]);

app.MapAGUIServer("/", agent);

モデルがツールを呼び出すと、AG-UI アダプターは、関数を実行するのではなく、ツール呼び出し割り込みで実行を完了します。

.NET クライアントからの割り込みを解決する

AGUIChatClient は、割り込みを ToolApprovalRequestContentとして表面に表示します。通常の MAF 承認の種類を使用して、応答を作成して送信します。

ToolApprovalRequestContent? request = null;

await foreach (AgentResponseUpdate update in
    remoteAgent.RunStreamingAsync(messages, session))
{
    request ??= update.Contents
        .OfType<ToolApprovalRequestContent>()
        .FirstOrDefault();
}

if (request is not null)
{
    ToolApprovalResponseContent response = request.CreateResponse(approved: true);
    ChatMessage resume = new(ChatRole.User, [response]);

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

クライアントが中断された実行を続行できるように、応答を送信するときに同じ AgentSession を再利用します。 approved: falseを使用して呼び出しを拒否します。アダプターは、MAF 応答を正規の AG-UI 再開ペイロードに変換します。

次のステップ

AG-UI を使用して共有状態を管理する

このチュートリアルでは、ユーザーが実行する前にツールの実行を承認する必要がある AG UI を使用して、ループ内の人間のワークフローを実装する方法について説明します。これは、財務トランザクション、データの変更、重大な影響を及ぼすアクションなどの機密性の高い操作に不可欠です。

前提条件

開始する前に、バックエンドツールレンダリングのチュートリアルを完了していることを確認し、次のことを理解してください。

関数ツールを作成する方法
AG-UIがツールイベントをストリームする方法
基本的なサーバーとクライアントのセットアップ

Human-in-the-Loop とは

Human-in-the-Loop (HITL) は、エージェントが特定の操作を実行する前にユーザーの承認を要求するパターンです。 AG-UI の場合:

エージェントが通常どおりツール呼び出しを生成する
サーバーは、すぐに実行するのではなく、承認要求をクライアントに送信します。
クライアントは要求を表示し、ユーザーにメッセージを表示します。
ユーザーがアクションを承認または拒否する
サーバーは応答を受信し、それに応じて処理を続行します。

Benefits

安全性: 意図しないアクションが実行されないようにする
透明性:ユーザーはエージェントが何をしたいかを正確に確認します
制御: ユーザーは機密性の高い操作に関して最終的な決定を下します
コンプライアンス: 人による監視に関する規制要件を満たす

承認のためのマーキングツール

ツールの承認を要求するには、approval_mode デコレーターで @tool パラメーターを使用します。

from agent_framework import tool
from typing import Annotated
from pydantic import Field


@tool(approval_mode="always_require")
def send_email(
    to: Annotated[str, Field(description="Email recipient address")],
    subject: Annotated[str, Field(description="Email subject line")],
    body: Annotated[str, Field(description="Email body content")],
) -> str:
    """Send an email to the specified recipient."""
    # Send email logic here
    return f"Email sent to {to} with subject '{subject}'"


@tool(approval_mode="always_require")
def delete_file(
    filepath: Annotated[str, Field(description="Path to the file to delete")],
) -> str:
    """Delete a file from the filesystem."""
    # Delete file logic here
    return f"File {filepath} has been deleted"

承認モード

always_require: 常に実行前に承認を要求する
never_require: 承認を要求しない (既定の動作)
conditional: 特定の条件に基づいて承認を要求する (カスタムロジック)

Human-in-the-Loop を使用したサーバーの作成

承認が必要なツールを使用した完全なサーバー実装を次に示します。

"""AG-UI server with human-in-the-loop."""

import os
from typing import Annotated

from agent_framework import Agent, tool
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework_ag_ui import AgentFrameworkAgent, add_agent_framework_fastapi_endpoint
from azure.identity import AzureCliCredential
from fastapi import FastAPI
from pydantic import Field


# Tools that require approval
@tool(approval_mode="always_require")
def transfer_money(
    from_account: Annotated[str, Field(description="Source account number")],
    to_account: Annotated[str, Field(description="Destination account number")],
    amount: Annotated[float, Field(description="Amount to transfer")],
    currency: Annotated[str, Field(description="Currency code")] = "USD",
) -> str:
    """Transfer money between accounts."""
    return f"Transferred {amount} {currency} from {from_account} to {to_account}"


@tool(approval_mode="always_require")
def cancel_subscription(
    subscription_id: Annotated[str, Field(description="Subscription identifier")],
) -> str:
    """Cancel a subscription."""
    return f"Subscription {subscription_id} has been cancelled"


# Regular tools (no approval required)
@tool
def check_balance(
    account: Annotated[str, Field(description="Account number")],
) -> str:
    """Check account balance."""
    # Simulated balance check
    return f"Account {account} balance: $5,432.10 USD"


# 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 agent with tools
agent = Agent(
    name="BankingAssistant",
    instructions="You are a banking assistant. Help users with their banking needs. Always confirm details before performing transfers.",
    client=chat_client,
    tools=[transfer_money, cancel_subscription, check_balance],
)

# Wrap agent to enable human-in-the-loop
wrapped_agent = AgentFrameworkAgent(
    agent=agent,
    require_confirmation=True,  # Enable human-in-the-loop
)

# Create FastAPI app
app = FastAPI(title="AG-UI Banking Assistant")
add_agent_framework_fastapi_endpoint(app, wrapped_agent, "/")

if __name__ == "__main__":
    import uvicorn

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

主な概念

AgentFrameworkAgent ラッパー: human-in-the-loop などの AG-UI プロトコル機能を有効にします
require_confirmation=True: マークされたツールの承認ワークフローをアクティブ化します
ツールレベルの制御: approval_mode="always_require" でマークされたツールのみが承認を要求します

承認の中断について

ツールで承認が必要な場合は、正規の AG-UI 割り込みで実行が完了します。

承認中断

{
  "type": "RUN_FINISHED",
  "threadId": "thread-1",
  "runId": "run-1",
  "outcome": {
    "type": "interrupt",
    "interrupts": [
      {
        "id": "approval-1",
        "reason": "tool_call",
        "message": "Approve tool call transfer_money?",
        "toolCallId": "call-1",
        "responseSchema": {
          "type": "object",
          "properties": {
            "accepted": { "type": "boolean" },
            "arguments": { "type": "object" }
          },
          "required": ["accepted"]
        },
        "metadata": {
          "agent_framework": {
            "type": "function_approval_request",
            "function_call": {
              "call_id": "call-1",
              "name": "transfer_money",
              "arguments": {
                "from_account": "1234567890",
                "to_account": "0987654321",
                "amount": 500.00,
                "currency": "USD"
              }
            }
          }
        }
      }
    ]
  }
}

ツールの承認割り込みでは、 reason: "tool_call" が使用され、 toolCallIdが含まれます。 ChatResponseUpdateからの最後のAGUIChatClientでは、outcomeとinterruptsの値がadditional_propertiesに保持されます。 Interrupt および ResumeEntry は、エージェントフレームワーク固有のモデルではなく、 ag_ui.coreからのプロトコルの種類です。

履歴書の形式

正規の resume 配列で同じスレッドを再開します。 accepted: falseを使用して、エージェントの続行を許可しながら操作を拒否します。ペイロードなしで status: "cancelled" を使用して、中断された実行を取り消します。

{
  "threadId": "thread-1",
  "messages": [],
  "resume": [
    {
      "interruptId": "approval-1",
      "status": "resolved",
      "payload": {
        "accepted": true
      }
    }
  ]
}

認可サポートを持つクライアント

承認要求を処理する AGUIChatClient を使用するクライアントを次に示します。

"""AG-UI client with human-in-the-loop support."""

import asyncio
import os

from agent_framework import Agent
from agent_framework_ag_ui import AGUIChatClient


def display_approval_request(update) -> None:
    """Display approval request details to the user."""
    print("\n\033[93m" + "=" * 60 + "\033[0m")
    print("\033[93mAPPROVAL REQUIRED\033[0m")
    print("\033[93m" + "=" * 60 + "\033[0m")

    # Display tool call details from update contents
    for i, content in enumerate(update.contents, 1):
        if content.type == "function_approval_request":
            function_call = content.function_call
            print(f"\nAction {i}:")
            print(f"  Tool: \033[95m{function_call.name}\033[0m")
            print(f"  Arguments:")
            for key, value in (function_call.arguments or {}).items():
                print(f"    {key}: {value}")

    print("\n\033[93m" + "=" * 60 + "\033[0m")


async def main():
    """Main client loop with approval handling."""
    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:
            message = input("\nUser (:q or quit to exit): ")
            if not message.strip():
                continue

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

            print("\nAssistant: ", end="", flush=True)
            pending_interrupts = []

            async for update in agent.run(message, session=thread, stream=True):
                # Check if this update carries an approval request.
                if any(content.type == "function_approval_request" for content in update.contents):
                    display_approval_request(update)

                if update.text:
                    print(f"\033[96m{update.text}\033[0m", end="", flush=True)

                properties = update.additional_properties or {}
                outcome = properties.get("outcome")
                if isinstance(outcome, dict) and outcome.get("type") == "interrupt":
                    pending_interrupts = outcome.get("interrupts", [])

            if pending_interrupts:
                resume_entries = []
                for interrupt in pending_interrupts:
                    prompt = interrupt.get("message", "Approve this action?")
                    user_choice = input(f"\n{prompt} (yes/no): ").strip().lower()
                    resume_entries.append({
                        "interruptId": interrupt["id"],
                        "status": "resolved",
                        "payload": {"accepted": user_choice in ("yes", "y")},
                    })

                print("\nAssistant: ", end="", flush=True)
                async for update in agent.run(
                    [],
                    session=thread,
                    stream=True,
                    options={
                        "available_interrupts": pending_interrupts,
                        "resume": resume_entries,
                    },
                ):
                    if update.text:
                        print(f"\033[96m{update.text}\033[0m", end="", flush=True)

            print()

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


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

相互作用の例

サーバーとクライアントが実行されている場合:

User (:q or quit to exit): Transfer $500 from account 1234567890 to account 0987654321

[Run Started]
============================================================
APPROVAL REQUIRED
============================================================

Action 1:
  Tool: transfer_money
  Arguments:
    from_account: 1234567890
    to_account: 0987654321
    amount: 500.0
    currency: USD

============================================================

Approve this action? (yes/no): yes

[Sending approval response: True]

[Tool Result: Transferred 500.0 USD from 1234567890 to 0987654321]
The transfer of $500 from account 1234567890 to account 0987654321 has been completed successfully.
[Run Finished]

ユーザーが拒否した場合:

Approve this action? (yes/no): no

[Sending approval response: False]

I understand. The transfer has been cancelled and no money was moved.
[Run Finished]

カスタム確認メッセージ

サーバーからの承認割り込みをレンダリングするときに、AG-UI クライアント UI で承認メッセージと確認メッセージをカスタマイズします。 Python AgentFrameworkAgentは、承認要求と割り込みメタデータを公開します。サーバー側の確認戦略オブジェクトは取得しません。

ベストプラクティス

ツールの説明を明確にする

ユーザーが承認内容を理解できるように、詳細な説明を入力します。

@tool(approval_mode="always_require")
def delete_database(
    database_name: Annotated[str, Field(description="Name of the database to permanently delete")],
) -> str:
    """
    Permanently delete a database and all its contents.

    WARNING: This action cannot be undone. All data in the database will be lost.
    Use with extreme caution.
    """
    # Implementation
    pass

詳細な承認

バッチ処理ではなく、個々の機密性の高いアクションの承認を要求します。

# Good: Individual approval per transfer
@tool(approval_mode="always_require")
def transfer_money(...): pass

# Avoid: Batching multiple sensitive operations
# Users should approve each operation separately

有益な議論

説明的なパラメーター名を使用し、コンテキストを指定します。

@tool(approval_mode="always_require")
def purchase_item(
    item_name: Annotated[str, Field(description="Name of the item to purchase")],
    quantity: Annotated[int, Field(description="Number of items to purchase")],
    price_per_item: Annotated[float, Field(description="Price per item in USD")],
    total_cost: Annotated[float, Field(description="Total cost including tax and shipping")],
) -> str:
    """Purchase items from the store."""
    pass

タイムアウト処理

承認要求に適切なタイムアウトを設定します。

# Client side
async with httpx.AsyncClient(timeout=120.0) as client:  # 2 minutes for user to respond
    # Handle approval
    pass

選択的承認

承認が必要なツールと必要でないツールを組み合わせることができます。

# No approval needed for read-only operations
@tool
def get_account_balance(...): pass

@tool
def list_transactions(...): pass

# Approval required for write operations
@tool(approval_mode="always_require")
def transfer_funds(...): pass

@tool(approval_mode="always_require")
def close_account(...): pass

バッチ承認と取り消し

1 つのモデル応答には、承認が必要なツールと、承認を必要としないツールの両方を含めることができます。目に見える割り込みを解決すると、承認の決定に従って、そのバッチからの他のツール呼び出しも完了します。たとえば、承認が必要な兄弟要素が拒否された場合でも、never_require の兄弟要素は実行され、その TOOL_CALL_RESULT は再開後の実行でストリーミングされます。

status: "cancelled"でキャンセルすると、承認の再開が中止され、スレッドのキューに登録された承認状態がクリアされます。それ以降の要求では、取り消されたバッチから古いツール呼び出しを再表面化したり、実行したりすることはできません。

次のステップ

MCP アプリの互換性

その他のリソース

Go は、承認必須のツールを備えた AG-UI の human-in-the-loop フローをサポートしています。関数ツールを tool.ApprovalRequiredFuncでラップし、 aguiproviderを使用してエージェントをホストします。

approveExpense := functool.MustNew(functool.Config{
    Name:        "approve_expense_report",
    Description: "Approve the expense report.",
}, func(ctx context.Context, expenseReportID string) (string, error) {
    return fmt.Sprintf("Expense report %s approved", expenseReportID), nil
})

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Tools: []tool.Tool{tool.ApprovalRequiredFunc(approveExpense)},
    },
})

Tip

完全な実行可能な例については、 AG-UI human-in-the-loop サンプルを参照してください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-25

ヒューマンインザループとのAG-UI

承認を要求

.NET クライアントからの割り込みを解決する

次のステップ

前提条件

Human-in-the-Loop とは

Benefits

承認のためのマーキング ツール

承認モード

Human-in-the-Loop を使用したサーバーの作成

主な概念

承認の中断について

承認中断

履歴書の形式

認可サポートを持つクライアント

相互作用の例

カスタム確認メッセージ

ベスト プラクティス

ツールの説明を明確にする

詳細な承認

有益な議論

タイムアウト処理

選択的承認

バッチ承認と取り消し

次のステップ

その他のリソース

フィードバック

その他のリソース

承認のためのマーキングツール

ベストプラクティス