Foundry がホストするエージェント

Microsoft Foundry Agent Service でホストされるエージェントを使用すると、コンテナー化されたエージェント アプリケーションをMicrosoftマネージド インフラストラクチャにデプロイできます。 プラットフォームは、スケーリング、セッション状態の永続化、セキュリティ、およびライフサイクル管理を処理するため、エージェントのロジックに集中できます。 Microsoft Foundry Hosted Agents は一般提供されており、独自のコードまたは優先エージェント フレームワークを使用して構築されたエージェントをサポートしています。 この記事では、Agent Framework ホスティング統合について具体的に説明します。

Agent Framework ホスティング統合を使用すると、最小限のコードで Foundry 応答または呼び出しプロトコルを使用して Agent を公開できます。 Pythonでは、ネイティブ Workflowをエージェントに変換せずに直接ホストすることもできます。

Note

Azure Developer CLI (azd) ワークフローを使用すると、他のフレームワークで構築されたエージェント コードを Foundry でホストされるエージェントにデプロイすることもできます。 フレームワークに依存しない概念とデプロイ ガイダンスについては、「 ホストされるエージェントとは」 を参照してください。この記事の残りの部分では、Agent Framework の統合について説明します。

ホストされているエージェントを使用する場合

必要に応じ、Foundry でホストされるエージェントを選択します。

  • マネージド インフラストラクチャ - コンテナー、Web サーバー、またはスケーリング ルールを自分で構成する必要はありません。
  • 組み込みのセッション管理 — プラットフォームは、ターンとアイドル期間にわたって $HOME およびアップロードされたファイルを保持します。
  • 専用エージェント ID — デプロイされたすべてのエージェントは、モデル、ツール、ダウンストリーム サービスに安全にアクセスするための独自の Entra ID を取得します。
  • OpenAI と互換性のあるエンドポイント - クライアントは、応答プロトコルを介して任意の OpenAI 互換 SDK を使用してエージェントと対話できます。
  • リアルタイム オーディオ エージェントの場合は、サーバー側の音声アクティビティ検出、エコー キャンセル、ノイズリダクションのために、Foundry Tools (Voice Live) の Azure Speech でホストされたエージェントを使用します。 詳細については、「 ホストされているエージェントで Voice Live を使用する」を参照してください。

Note

Python agent-framework-foundry-hosting統合はプレリリースです。 マネージド ホスティング サービスMicrosoft Foundry Hosted Agents が一般提供されています。

前提条件

  • Azure サブスクリプション
  • Azure 開発者 CLI (azd) と AI エージェントの拡張機能: azd ext install azure.ai.agents

ローカル テストの場合は、次のものが必要です。

  • Microsoft Foundry プロジェクト(たとえば、モデル展開を含む gpt-4o)
  • Azure CLI のインストールと認証 (az login)

ホスティング NuGet パッケージをインストールします。

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease
  • Python 3.10 以降

プレリリース ホスティング パッケージ、Foundry クライアント、Azure認証パッケージをインストールします。

pip install --pre agent-framework-foundry agent-framework-foundry-hosting azure-identity

Foundry では、プラットフォームは呼び出し元のユーザー コンテキストと呼び出しコンテキストを提供します。ホスティング インフラストラクチャでは、それらを使用してユーザーごとの状態を分離し、要求コンテキストを Foundry サービスに転送します。 ローカル実行はそのプラットフォーム コンテキストを受け取らないので、アプリケーションは必要に応じて独自の ID と状態制御を提供する必要があります。

応答プロトコル

応答プロトコルは、ほとんどのエージェントに推奨される開始点です。 OpenAI と互換性のある /responses エンドポイントが公開され、プラットフォームによって会話履歴、ストリーミング、セッションのライフサイクルが自動的に管理されます。

ホストPythonエージェントの場合、早期に終了する応答はincomplete状態になります。 ストリーミング クライアントは終端 response.incomplete イベントを受信し、非ストリーミング クライアントは incomplete が status に設定された状態で受信します。 content_filter の完了理由は、content_filter が incomplete_details.reason に設定されている状態に対応し、length は max_output_tokens に対応します。 生成された出力または拒否コンテンツは、応答で引き続き使用できます。

using Azure.AI.AgentServer.Core;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry.Hosting;

var projectEndpoint = new Uri(Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
    ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set."));
var deployment = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "gpt-4o";

AIAgent agent = new AIProjectClient(projectEndpoint, new DefaultAzureCredential())
    .AsAIAgent(
        model: deployment,
        instructions: "You are a helpful AI assistant.",
        name: "my-agent");

var builder = AgentHost.CreateBuilder(args);
builder.Services.AddFoundryResponses(agent);
builder.RegisterProtocol("responses", endpoints => endpoints.MapFoundryResponses());

var app = builder.Build();
app.Run();

AgentHost.CreateBuilderは、Foundry ホスティング環境用に事前構成されたアプリケーション ホストを作成します。 AddFoundryResponsesはエージェントを Responses プロトコル ハンドラーに登録し、MapFoundryResponses/responses HTTP エンドポイントをマップします。

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import ResponsesHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a helpful AI assistant.",
)

server = ResponsesHostServer(agent)
server.run()

ResponsesHostServerはエージェントをラップし、Foundry Responses プロトコルを介して公開します。 呼び出し元の store フィールドは、外部応答とホスト管理セッションと承認状態を保存するかどうかを制御します。 history_source設定では、モデル履歴を提供するユーザーが個別に選択されます。

history_source モデル履歴の挙動
"agent_server" (既定値) ホストは、格納されている外部応答トランスクリプトを再構築し、ダウンストリーム サービス ストレージを無効にして、履歴の重複を防ぎます。
"service" ホストは現在の入力のみを送信し、格納しているモデル サービスの継続 ID をプライベートに保存します。 保存されたプロバイダーの会話は、以前の応答から分岐することはできません。
"agent" ホストは現在の入力のみを送信します。 エージェントの HistoryProvider またはダウンストリーム サービス ストレージのデフォルト設定が履歴を管理します。

"agent_server"または"service"を読み込み対応のHistoryProviderと組み合わせないでください。 既定のモードでは、 conversation_id、 previous_response_id、 conversationなどの固定ダウンストリーム継続オプションも拒否されます。 カスタムSupportsAgentRun実装にはhistory_source="agent"を使用します。

response_store コンストラクター パラメーターは、外部の応答永続化のバックエンドを選択します。 以前のコンストラクター パラメーター store は、 response_storeの非推奨のエイリアスです。どちらのパラメーターも、呼び出し元の要求ごとの store フィールドを設定しません。 store=falseを含む要求は一発です。ホストで管理された状態を保存せず、サポートされているダウンストリーム ストレージを無効にし、background=trueを使用できません。

ホストは指定されたエージェントを所有し、ホスティング固有のコンテキスト プロバイダーを追加する場合があります。 エージェントを別のホストで再利用したり、ホスト構築後に直接呼び出したりしないでください。

応答ホストは、ネイティブ コンピューターの呼び出し、スクリーンショット、および安全性チェックを保持します。 アプリケーションでは、要求されたアクションを実行し、安全性チェックを明示的に確認する必要があります。 完全なフローについては、「 ネイティブ コンピューターの使用」を参照してください。

エージェント インスタンスまたはファクトリを選択する

ResponsesHostServer と InvocationsHostServer はどちらも、agent パラメーターに、エージェント インスタンス、またはゼロ引数の同期もしくは非同期の呼び出し可能オブジェクトを指定できます。 ホストは、インスタンスを有効期間中再利用します。 呼び出し可能オブジェクトはリクエストごとに1回実行され、返されたエージェントはそのリクエストに属します。

エージェントが AgentSessionの外部で変更可能な状態を保持する場合は、呼び出し可能な状態を使用します。 特に、新しいワークフロー、エグゼキュータ、ラップ済みエージェントを構築するファクトリを使ってWorkflowAgentを作成します。

def create_workflow_agent():
    return build_workflow().as_agent(name="support-workflow")


server = ResponsesHostServer(agent=create_workflow_agent)

ワークフロー名と Executor ID は安定した状態に保ち、後で応答要求で保存されたチェックポイントを見つけることができます。 ResponsesHostServer は、セッション、チェックポイント、および関数承認ストアを通じてサポートされている状態を続行します。これは、要求スコープのエージェントに任意のフィールドを保持しません。 ワークフローと回復性の高い実行時間の長いワークフローサンプルを参照してください。

また、統合が要求 ID を持っているか、要求固有のリソースを所有している場合にもファクトリを使用します。 たとえば、現在のプラットフォーム呼び出しまたはユーザー コンテキストを使用する場合は、MCP 接続、ツールボックス、スキル プロバイダー、検索クライアント、メモリ プロバイダー、およびそれらの認証情報をファクトリ内で作成します。 プロセス全体の MCP 接続を再利用すると、その接続を開いた要求の ID を保持できます。

ホストは、各リクエストについて、ファクトリで作成されるエージェントに入り、そこから退出します。 Agent はコンテキスト管理クライアントと MCP ツールを管理しますが、ファクトリは、作成する他のプロバイダー、トランスポート、または資格情報を閉じる必要があります。 アプリケーションがファクトリの外部から提供した共有オブジェクトを閉じないでください。

Responses を使用してネイティブ ワークフローを構築する

Pythonは、workflow=を介して構築済みのワークフローを直接ホストできます。 ネイティブ ワークフローには、現在の Responses リクエストを型指定された開始入力または保留中の返信の完全なバッチにマップする parse_response コールバックが必要です。

from pydantic import BaseModel

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    HostedResponseRequest,
    ResponsesHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    text: str


def build_workflow(request: HostedResponseRequest):
    return build_fresh_workflow()


async def parse_response(request: HostedResponseRequest) -> WorkflowTurn[Ticket]:
    items = await request.get_input_items()
    if any(item.get("type") in ("function_call_output", "mcp_approval_response") for item in items):
        return WorkflowTurn(responses=await request.get_workflow_responses())

    text = await request.get_input_text()
    return WorkflowTurn(input=Ticket.model_validate_json(text or ""))


server = ResponsesHostServer(
    workflow=build_workflow,
    parse_response=parse_response,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[f"{Ticket.__module__}:{Ticket.__qualname__}"],
    ),
)

バックグラウンド作業を一時停止、続行、または復旧できるワークフローには、要求対応の同期または非同期ファクトリを使用します。 ファクトリは、新しく構築されたグラフと、新しく変更可能な Executor、エージェント、クライアント、プロバイダー、ツールを返す必要があります。 ホストが外部応答に関連付けられている正確なチェックポイントを復元できるように、ワークフロー名と Executor ID を安定した状態に保ちます。

信頼されたプラットフォーム ユーザーと Foundry サンドボックスによるネイティブ ワークフローの状態の分離 ホストは、応答権限を使用する前に、保留中の完全な応答バッチを検証します。 ワークフローを実行する前に、古い、部分的、重複、再生済み、クロスユーザー、クロスサンドボックスの応答は失敗として処理されます。 store=falseを含む要求はワークフローの状態を保存せず、再開可能な一時停止を返すことはできません。

list[Message]を受け入れるレガシ ワークフローの場合は、response_input_messages(request)を使用して現在の応答ターンのみを変換します。 事前の外部履歴を読み込んだり、保留中のワークフロー応答をデコードしたりすることはありません。 ホスティング agent=workflow.as_agent() は現在のベータ期間中も使用できますが、非推奨の警告が生成されます。 完全な例については、ネイティブ Responses ワークフローのサンプルを参照してください。

状態を保持し、長時間にわたる会話を処理する

ResponsesHostServer と InvocationsHostServer は、既定では永続セッション ストアを構成します。 AgentSessionStoreProvider は FoundryAgentSessionStoreを提供します。応答セッションでは agent_sessions 論理ストアが使用され、呼び出しセッションでは個別の invocation_sessions ストアが使用されます。 これらのストアは、ホストされている場合は Foundry State Store を使用し、ローカルで実行する場合は SDK のファイルベースのストレージを使用します。

応答ワークフロー エージェントの場合、 CheckpointStoreProvider は FoundryCheckpointStoreを提供します。 ネイティブ応答と呼び出しワークフローでは、同一の継続チェックポイントに同じプロバイダーを使用します。 FunctionApprovalStoreProvider は、保留中のエージェント ツール承認の FoundryFunctionApprovalStore を提供します。 ネイティブ ワークフロー要求と承認返信は、代わりにワークフロー チェックポイントにバインドされます。

Foundry で実行すると、既定のPythonは、プラットフォーム ユーザー ID と Foundry サンドボックス セッション ID によって名前空間の状態を格納します。 また、状態操作ごとにプラットフォーム呼び出し ID も必要です。 呼び出し ID は操作を承認し、関連付けます。これは会話 ID ではなく、ストレージ キーの一部ではありません。

応答の場合、プラットフォームで構成された FOUNDRY_AGENT_SESSION_ID はサンドボックスを識別し、別の呼び出し元が指定した agent_session_id は拒否されます。 呼び出しの場合、ホストは要求コンテキストに対してルーティングされた agent_session_id クエリ パラメーターを検証します。 FOUNDRY_AGENT_SESSION_IDが構成されていない場合は、クエリ パラメーターが存在し、空でない必要があり、要求コンテキストと一致している必要があります。 SDK フォールバック ID を使用する代わりに、欠損値、重複値、または競合する値は拒否されます。

これらの保証は、既定のホステッド ストアに適用されます。 カスタム ストア プロバイダーは、同等のユーザーとサンドボックスの分離を実装し、内部 AgentSession.session_id をホスト参照キーとは別に保持し、条件付き書き込みを使用して、古い要求が新しいスナップショットを上書きできないようにする必要があります。 新しいキーには、無条件のアップサートではなく、作成時のみの書き込みを使用する必要があります。 ETag で保護された書き込みと削除を使用した Cosmos DB 実装の カスタム ストレージ サンプル を参照してください。

history_source="agent"では、構成されたセッション ストアは、InMemoryHistoryProviderからのメッセージを含め、AgentSessionによって保持されるプロバイダーの状態を保持します。

両方のホストは、agent_session_store_providerを介してStoreProvider[SessionStore]を受け入れます。 セッション状態では、 AgentSession シリアル化をサポートする必要があります。 カスタム状態の種類のコーデックをregister_state_type()に登録します。復元された状態では、Pythonオブジェクト ID は保持されません。 新しい既定のストアでは、最後の書き込みから 30 日後にセッションが期限切れになります。 カスタムプロバイダーは、独自の保持期間を管理します。

スコープが設定された既定のストアでは、従来のスコープ外の agent_sessions、 invocation_sessions、チェックポイント、または関数承認データは読み取られません。 古い previous_response_id または会話 ID を再利用するのではなく、新しい応答会話を開始します。 呼び出しは、スコープ付きストア内の空の Agent Framework セッションで開始されます。

読み込まれた AgentSession レコードは ETag 条件を使用します。 別のリクエストが先に同じセッションの状態を進めた場合、古くなった書き込みは、より新しい状態を上書きするのではなく、失敗します。 このチェックでは、エージェントまたはツールの副作用に対するトランザクションや 1 回の実行は提供されないため、アプリケーションは重複する要求を調整する必要があります。

応答固有のストレージの場合は、function_approval_store_providerにStoreProviderを渡すか、checkpoint_store_providerにContextScopedStoreProviderを渡します。

外部バックグラウンド作業では、呼び出し元から参照できる response.id をポーリングに使用します。 既定の background_source="agent_server" は、ホストでのバックグラウンド実行を保持します。 background_source="provider"は、history_source="service"と保存機能を備えた再開可能な Responses クライアントでのみ設定してください。 ResponsesServerOptions(resilient_background=True)も設定されている場合、ホストは、プライベート継続トークンを保存した後にのみ、プロバイダーのポーリングを回復できます。 次のトークンが保存される前にクラッシュするとローカルツールの副作用が再実行される可能性があるため、それらを冪等にしてください。

azure.ai.agentserver.responsesからResponsesServerOptionsをインポートし、options パラメーターを使用してResponsesHostServerに渡します。 実行時間の長い会話オプションは、エージェントの種類によって異なります。

Capability エージェントの種類 要件と動作
ワークフロー チェックポイントのバックグラウンド復旧 ワークフローのみ ResponsesServerOptions(resilient_background=True)を設定します。 store=trueとbackground=trueを使用して応答要求を送信します。 再起動後、ホストは最新の永続的ワークフロー チェックポイントを再開するか、チェックポイントが存在しない場合は元の入力を再生します。 ホストによって管理されるため、ワークフローでチェックポイント ストレージを構成しないでください。 最後に永続化されたチェックポイント以降の処理は再実行される可能性があるため、外部副作用は冪等にしてください。
プロバイダーネイティブのバックグラウンド応答 保存機能付き Responses クライアントを使う非ワークフロー Agent history_source="service"とbackground_source="provider"を設定します。 保存されたプロバイダー継続トークンがホストの再起動後も存続する必要がある場合に resilient_background=True を設定します。
制御可能な会話 一時的にご利用いただけません steerable_conversations=True は設定しないでください。 ホストは、エージェント サーバー SDK が拒否されたステアリング ターンを安全に処理するまで、構築中に RuntimeError を発生させます。

完全な実装については、 カスタム ストレージ、 基本的な応答履歴と背景、 回復性の高い実行時間の長いワークフロー サンプルを参照してください。

ホストされているサンドボックスからファイルを読み取る

ホストされるサンドボックスの永続的な $HOME は、一般的なファイル システム境界としてではなく、要求ルーティング リソースとして扱います。 アプリケーションが明示的に専用ディレクトリにアップロードするファイルのみを受け入れ、現在のサンドボックス ID を検証し、絶対パス、トラバーサル、リンク、非規則ファイル、およびサイズ超過または無効なコンテンツを拒否します。

応答プロトコルの場合は、 agent_session_id 本文フィールドを使用して、ホストされたセッションに要求をルーティングします。 クエリ文字列セレクターは、呼び出し用です。 セッションアップロードとツールボックスコードインタープリターファイルは別々のリソースです。アップロードされたサンドボックス ファイルは、ツールボックス コンテナーに自動的にマウントされません。 境界付き UTF-8 読み取りとローカルおよびホスト型アップロードのガイダンスについては、 セッション ファイルのサンプル を参照してください。

制御要求オプション

ホストは、ネイティブ応答生成フィールドを Agent Framework の実行オプションにマップします。 たとえば、 max_output_tokens は max_tokens、parallel_tool_calls は allow_multiple_tool_calls になります。 extra_bodyからのフラット化された値は、変換されたネイティブ値をオーバーライドします。

同期または非同期の prepare_options(request, options) フックを使用して、通常のエージェントを実行する前に呼び出し元モデルのオプションを削除または置換します。 フックは、ホスト制御 ID、ストレージ、継続、またはトランスポート フィールドを設定できません。 ランタイム モデル オプションを受け入れられないカスタム SupportsAgentRun 実装の場合は、 unsupported_options を "warn" (既定)、 "ignore"、または "error"に設定します。

Foundry でホストされる MCP ツールでユーザーの同意が必要な場合、 ResponsesHostServer は oauth_consent_request 出力項目を含む不完全な応答を返します。 そのconsent_linkをユーザーに提示し、ユーザーが同意を完了した後、不完全な応答のIDをprevious_response_idとして続行します。 ホストは、この再試行のエージェント セッションを保持し、絶対 HTTPS 同意リンクのみを公開します。

ホストが予想される承認元を認識している場合は、同意リンクを allowed_oauth_consent_originsに制限します。

server = ResponsesHostServer(
    agent,
    allowed_oauth_consent_origins=[
        "https://logic-region.consent.azure-apihub.net",
        "https://auth.partner.example",
    ],
)

許可リストを省略すると、宛先の配信元を制限することなく、絶対 HTTPS 検証が保持されます。 空のリストを指定すると、すべての同意リンクが拒否されます。 正確な HTTPS 配信元のみを構成します。パス、クエリ、またはフラグメントを含むエントリは拒否されます。

呼び出しプロトコル

呼び出しプロトコルを使用すると、HTTP 要求と応答を完全に制御できます。 OpenAI と互換性のないカスタム ペイロード、非会話型処理、またはストリーミング プロトコルが必要な場合に使用します。

C# の呼び出しプロトコルでは、受信要求を処理するカスタム InvocationHandler を実装します。

using Azure.AI.AgentServer.Core;
using Azure.AI.AgentServer.Invocations;
using Microsoft.Agents.AI;

var builder = AgentHost.CreateBuilder(args);

builder.Services.AddSingleton<AIAgent, MyAgent>();
builder.Services.AddInvocationsServer();
builder.Services.AddScoped<InvocationHandler, MyInvocationHandler>();

builder.RegisterProtocol("invocations", endpoints => endpoints.MapInvocationsServer());

var app = builder.Build();
app.Run();

AddInvocationsServer メソッドは、Invocations プロトコル サービスを登録します。 InvocationHandlerを実装して、エージェントが各要求を処理する方法を定義します。

軽量セットアップの場合は、InvocationsHostServer パッケージのagent_framework_foundry_hostingを使用します。 エージェントは、 ResponsesHostServer と同様にラップされ、セッション管理が自動的に処理されます。

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import InvocationsHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

server = InvocationsHostServer(agent)
server.run()

InvocationsHostServer は、Responses ホストに記述されている同じインスタンスまたは要求スコープファクトリ フォームを受け入れます。 構成されたストアからシリアル化されたセッションが復元されるため、完了した会話はホストの再起動後も続行できます。 ストレージの動作、保持期間、およびカスタマイズについては、「状態を永続化し、長時間実行される会話を処理する」を参照してください。

ホスト環境では、Invocations は 状態を保持し、長時間にわたる会話を処理する で説明されている検証済みのリクエスト スコープを使用します。 AgentSession.session_idを 1 つの不透明な値として扱います。内部表現を解析したり依存したりしないでください。 ローカル実行では、既存のシングル ユーザー ストレージの動作が維持されます。

Invocations を使用してネイティブ ワークフローをホストする

ネイティブ ワークフローをホストするために、 workflow= と明示的な parse_request コールバックを渡します。 コールバックは、アプリケーション JSON スキーマを管理し、型指定された入力または完全な保留中の応答バッチを含む WorkflowTurn を返します。

from pydantic import BaseModel
from starlette.requests import Request

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    InvocationsHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    ticket_id: str
    question: str


class TicketDecision(BaseModel):
    approved: bool


def build_workflow(_request: Request):
    return build_fresh_workflow()


async def parse_request(request: Request) -> WorkflowTurn[Ticket]:
    payload = await request.json()
    stream = payload.get("stream", False)

    if "responses" in payload:
        decisions = {
            request_id: TicketDecision.model_validate(value)
            for request_id, value in payload["responses"].items()
        }
        return WorkflowTurn(responses=decisions, stream=stream)

    ticket = Ticket.model_validate(payload)
    return WorkflowTurn(input=ticket, stream=stream)


server = InvocationsHostServer(
    workflow=build_workflow,
    parse_request=parse_request,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[
            f"{Ticket.__module__}:{Ticket.__qualname__}",
            f"{TicketDecision.__module__}:{TicketDecision.__qualname__}",
        ],
    ),
)

ワークフローがチェックポイント プロバイダーの allowed_checkpoint_types リストに保存する、すべてのカスタム アプリケーションの種類を含めます。

ホストされるワークフローには、安定したワークフロー ID と Executor ID を備えた新しく構築されたグラフを返す要求対応ファクトリが必要です。 直接構築のワークフローは、一時停止しないローカルの 1 回限りの実行でのみ使用できます。

非ストリーミング ワークフロー応答では、output イベント リストを含む application JSON を使用します。 ストリーミングでは、フレーム化された output イベントと request_info イベントが出力され、その後、正確なワークフロー カーソルが保存された後にのみ done が生成されます。 ストリーミングされた出力は、done までは暫定的なものとして扱います。 ネイティブ ワークフローでは、legacy_wire_format=True機能はサポートされていません。

ホストは、信頼されたユーザーおよびサンドボックスのスコープ内の正確な保留中のチェックポイントに対する応答を検証します。 ワークフローに複数の保留中の要求がある場合は、バッチ全体に1ターンで返信します。 実行可能パーサー、型指定されたチケット ワークフロー、チェックポイントの種類の許可リスト、JSON/SSE の例については、 ネイティブ呼び出しワークフローのサンプルを参照してください。

呼び出しの要求と応答をカスタマイズする

既定では、 POST /invocations は、文字列 message、省略可能な options オブジェクト、および省略可能なブール stream 値を持つ JSON オブジェクトを受け入れます。 アプリケーション固有のペイロードを受け入れるには、InvocationRun(messages, options, stream)を返す同期または非同期のparse_requestコールバックを渡します。 エージェントを実行する前に、 prepare_options を使用して呼び出し元生成オプションのコピーをフィルター処理または置換します。

ホストはフック出力を検証し、プラットフォーム ID、ストレージ、継続、およびエージェント実行コントロールを拒否します。 ランタイム オプションを受け入れないエージェントの場合は、 unsupported_options を "warn" (既定)、 "ignore"、または "error"に設定します。 完全な実装については、 呼び出しパーサーのサンプル を参照してください。

ストリーミング以外の成功は、 {"response": "..."}形式で JSON を返します。 ストリーミングでは、サーバー送信イベント (1 つ以上の event: delta フレーム、成功した場合は event: done 、失敗した場合は event: error ) が使用されます。 ストリームはエラーの前にデルタを出力できるため、クライアントはデルタではなく doneを正常完了として扱う必要があります。 ホストは、応答ストリームを最終処理し、AgentSessionを永続化した後にのみ、doneを出力します。 その session_id は、シリアル化された AgentSession.session_idではなく、プラットフォーム サンドボックス ルート ID です。

legacy_wire_format=Trueは、以前のプレーンテキスト応答と生テキスト チャンク ストリームを必要とする既存のクライアントを移行する場合にのみ設定します。 この互換モードは非推奨であり、失敗を成功したテキストに変換しません。 ホストは、1 つのプロセス内でのみ同じセッション要求をシリアル化します。クロスプロセスの比較とスワップの競合は、外部ツールの効果の後でも発生する可能性があります。

呼び出しプロトコルは、保留中または中断されているワークフローの実行を再開しません。 別のワークフロー継続動作が必要な場合は、次のセクションのカスタム ハンドラー パターンを使用します。

要求処理を完全に制御するには、InvocationAgentServerHost パッケージのazure.ai.agentserver.invocationsを直接使用し、独自の呼び出しハンドラーを実装します。

import os
from collections.abc import AsyncGenerator

from agent_framework import Agent, AgentSession
from agent_framework.foundry import FoundryChatClient
from azure.ai.agentserver.invocations import InvocationAgentServerHost
from azure.identity import DefaultAzureCredential
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse

_sessions: dict[str, AgentSession] = {}

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

app = InvocationAgentServerHost()


@app.invoke_handler
async def handle_invoke(request: Request):
    """Handle streaming multi-turn chat."""
    data = await request.json()
    session_id = request.state.session_id
    stream = data.get("stream", False)
    user_message = data.get("message", None)

    if user_message is None:
        return Response(content="Missing 'message' in request", status_code=400)

    session = _sessions.setdefault(session_id, AgentSession(session_id=session_id))

    if stream:

        async def stream_response() -> AsyncGenerator[str]:
            async for update in agent.run(user_message, session=session, stream=True):
                yield update.text

        return StreamingResponse(
            stream_response(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
        )

    response = await agent.run([user_message], session=session, stream=stream)
    return JSONResponse({"response": response.text})


if __name__ == "__main__":
    app.run()

Warning

カスタム ハンドラーの例のメモリ内セッション ストアは、再起動時に失われます。 運用環境では永続ストレージ (Cosmos DB など) を使用します。

完全な呼び出しのデプロイについては、 Foundry でホストされる Telegram サンプルを参照してください。 ホストされたエージェント Webhook の前に API Management が配置され、永続的な会話履歴にマネージド ID、Key Vault、Cosmos DB が使用されます。

Note

Foundry でホストされるエージェントの Go サポートは近日公開予定です。 最新の状態については、 Agent Framework Go リポジトリ を参照してください。

Tip

ホストされるエージェント プロジェクトの例については、Pythonサンプルまたは C# サンプルを参照してください。 または、 azd ai agent init コマンドを使用して、新しいホステッド エージェント プロジェクトを最初からスキャフォールディングします。 詳細な手順については、この クイック スタート ガイド を参照してください。

ローカルでの実行

Azure Developer CLI (azd) は、ホストされたエージェントをローカルで実行してテストする最も簡単な方法を提供します。

プロジェクトを初期化する

新しいフォルダーを作成し、サンプル マニフェストから初期化します。

mkdir my-hosted-agent && cd my-hosted-agent
azd ai agent init -m <path-to-agent.manifest.yaml>

Tip

マニフェストには、ローカル YAML ファイルへのパスまたはリモート マニフェストへの URL を指定できます。

環境変数の設定

export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export AZURE_AI_MODEL_DEPLOYMENT_NAME="<your-model-deployment>"

エージェント ホストを実行する

azd ai agent run

エージェント ホストは、 http://localhost:8088で開始されます。

エージェントを呼び出す

azd ai agent invoke --local "Hello!"

または、 curlを使用します。

curl -X POST http://localhost:8088/responses \
  -H "Content-Type: application/json" \
  -d '{"input": "Hello!"}'

または、PowerShell で次の手順を実行します。

(Invoke-WebRequest -Uri http://localhost:8088/responses -Method POST -ContentType "application/json" -Body '{"input": "Hello!"}').Content

Foundry へのデプロイ

エージェントをローカルで検証したら、Microsoft Foundry にデプロイします。

  1. リソースをプロビジョニング する (Foundry プロジェクトがまだない場合):

    azd provision
    

    これにより、Foundry インスタンス、プロジェクト、モデル デプロイ、Application Insights、コンテナー レジストリを含むリソース グループが作成されます。

  2. エージェントをデプロイします。

    azd deploy
    

    これにより、エージェントがコンテナー イメージとしてパッケージ化され、Azure Container Registryにプッシュされ、Foundry Agent Service にデプロイされます。

Foundry ホスティング インフラストラクチャは、実行時に次の環境変数をエージェント コンテナーに自動的に挿入します。

Variable 説明
FOUNDRY_PROJECT_ENDPOINT Foundry プロジェクトのエンドポイント URL。
AZURE_AI_MODEL_DEPLOYMENT_NAME モデルの展開名(azd ai agent initで設定)。
APPLICATIONINSIGHTS_CONNECTION_STRING テレメトリ用の接続文字列としての Application Insights。

デプロイ後、エージェントは専用の Foundry エンドポイントを介してアクセスでき、Foundry ポータルからテストすることもできます。

次のステップ