Microsoft Agent Framework、Python でチャネル機能追加

エージェントやワークフローが真に有用となるのは、人々や他のシステムが普段使っているインターフェースやチャネルを通じてアクセスできる場合に限られます。具体的には、OpenAI Responses クライアント、Telegram、A2A を利用する他社製エージェント、あるいは MCP クライアントなどが該当します。

エージェントやワークフローを構築する開発者にとって重要なのは、どのチャネルを公開するか、そしてそれらがどのように振る舞うかを自分で制御できることです。これらを支援するため、Microsoft Agent Framework for agents and workflows in Python に「チャネル」機能を新たに導入しました。

アプリケーションに必要なチャネルを選択しましょう

新しいパッケージ群は、小さな共有ホスティングコアと、各チャネル固有の統合モジュールから構成されています。

• agent-framework-hosting: エージェント、ワークフロー、セッション状態を管理するための共通ヘルパーを提供します。
• agent-framework-hosting-responses: OpenAI Responses のリクエストと結果を変換します。
• agent-framework-hosting-telegram: Telegram からの更新情報や外部への操作変換を行います。
• agent-framework-hosting-a2a: エージェントやワークフローをネイティブな A2A SDK タイプに接続します。
• agent-framework-hosting-mcp: エージェントやワークフローをネイティブな MCP ツールとして公開します。

各チャネルパッケージは、そのプロトコルの境界線に特化して設計されています。アプリケーション側では、選択した Web フレームワークまたはネイティブ SDK を引き続き使用し、ルーティング、認証、認可、ストレージ、バックグラウンド処理、デプロイメントなどの管理責任を保持します。

つまり、エージェントやワークフローを新しいランタイム環境に埋め込む必要も、複数の SDK で重複して維持する必要もありません。必要なチャネルを 1 つ追加するだけで済むのです。

共通のターゲットと状態モデル

「agent-framework-hosting」パッケージは、すべてのチャネルで利用される共通の基盤を提供します。AgentState はエージェントのターゲットとそのセッションストアを保持し、WorkflowState はインスタンス、ファクトリ、またはビルダーからワークフローのターゲットを解決します。

from agent_framework_hosting import AgentState, WorkflowState

agent_state = AgentState(agent)

workflow_state = WorkflowState(workflow_builder, cache_target=False)

両方の状態タイプは、チャネルハンドラやアダプターがターゲットを一貫した方法で解決するための共通の手段を提供します。ターゲットは直接作成するか、同期または非同期の設定コードを通じて供給することもでき、各リクエストごとにキャッシュしたり再作成したりできます。

エージェントの場合、AgentState はアプリケーションが選択したセッション ID を AgentSession 値にマッピングします。ただし、Responses の呼び出し元、Telegram ユーザー、A2A コンテキスト、または MCP クライアントがどのようにしてセッション ID に変換されるかについては決定しません。このマッピングはアプリケーション側で構築する必要があります。以下の resolve_session_id は例示用のアプリケーションコードであり、Agent Framework から提供される関数ではありません。

アプリケーション定義のアイデンティティとセッションマッピング。

session_id = resolve_session_id(authenticated_user, channel_identity)

session = await agent_state.get_or_create_session(session_id)

このマッピングにより、1 つの会話を複数のチャネルにわたって継続して実行することが可能になります。認証済みユーザーが、Responses と Telegram の両方から同じ正規化セッション ID に解決される場合、両方のチャネルで同一の AgentSession が読み込まれて更新されます。逆に、異なる ID に解決された場合は、それぞれの履歴は別々に保持されます。共有セッションを安全に保つためのアイデンティティリンク、権限管理、競合制御は、アプリケーション側が責任を持って実装する必要があります。

ワークフローもまた、WorkflowState を通じて同じターゲット解決モデルを利用しますが、チェックポイントの保存やチャネル継続 ID からチェックポイントへのマッピングは、それぞれアプリケーションが管理します。両方のターゲットタイプにおいて、共有ホスティング層によってエージェントまたはワークフローの定義を周囲のチャネルから独立させることが可能であり、各チャネルパッケージがプロトコル固有の入出力処理を担当します。

OpenAI Responses

Responses ヘルパーは、受信したリクエストを Agent Framework の実行値に変換し、完了またはストリーミングされた結果を再び Responses 形式に戻す役割を果たします。

from agent_framework_hosting_responses import (

create_response_id,

responses_from_run,

responses_session_id,

responses_to_run,

)

run = responses_to_run(body)

session_id, is_conversation = responses_session_id(body)

response_id = create_response_id()

session = await agent_state.get_or_create_session(session_id or response_id)

result = await (await agent_state.get_target()).run(

run["messages"],

session=session,

options=run["options"],

)

await agent_state.set_session(session_id if is_conversation else response_id, session)

response = responses_from_run(

result,

response_id=response_id,

conversation_id=session_id if is_conversation else None,

)

このアプリケーションは、レスポンス ID と会話 ID をセッションにどのようにマッピングするかを決定します。また、呼び出し元が制御可能なリクエストオプションの範囲やストリーミングの実行タイミング、継続 ID の認証方法、そしてセッション状態の保存場所についても定義されます。

この実行可能（Runnable）な Responses エージェントのサンプルでは、ネイティブの FastAPI ルーティング、ストリーミング機能、セッションの継続処理、およびアプリケーションが管理するオプションポリシーが実装されています。

ワークフローでも同じ Responses インターフェースを利用できます。AgentSession を保存するのではなく、応答 ID をワークフローのチェックポイントにマッピングします。完全な実装については、Responses ワークフローのサンプルを参照してください。

Telegram

Telegram のヘルパーは、ネイティブな更新情報を Agent Framework への入力に変換し、ストリーミング実行を Telegram 上の操作へと変換します。アプリケーションは、これらの操作を実行するために直接 HTTP コールを行うか、aiogram や python-telegram-bot といった Telegram SDK を利用します。

コマンド、ウェブフック認証、ポーリング、メディア処理、編集の制限、配信ポリシーなどは、通常のアプリケーションコードとして扱われます。また、コンテキストプロバイダーを通じて手動でチャンネル固有の指示を追加することも可能です。例えば、Telegram 向けのレスポンス形式を調整する際などに役立ちます。

Telegram のサンプルには、aiogram を使用した完全なポーリングおよびウェブフックアプリケーションが含まれています。これにより、ストリーミング編集、/new などのコマンド、メディア変換、チャットごとの順序制御、セッションの継続性がデモンストレーションされています。

A2A と MCP

A2A の場合、AgentA2AAdapter と WorkflowA2AAdapter がネイティブなエージェントカードを生成し、広告される入力・出力モードが変換ヘルパーと整合するように保ちます。アプリケーション側は、ネイティブな A2A エグゼキューター、タスクライフサイクル、イベントキュー、ルーティング、タスクリストの管理を引き続き担当します。

A2A ホーティングのサンプルでは、Agent Framework のエージェントをネイティブな A2A サーバーを通じて公開する方法が示されています。

MCP については、AgentMCPTool と WorkflowMCPTool が Agent Framework のターゲットからネイティブツールを派生させます。アプリケーションは、FastMCP などのサーバーや直接登録された MCP ハンドラーと組み合わせて、より低レベルの変換関数を利用することも可能です。

MCP ホーティングのサンプルには、手動変換、FastMCP、生成されたエージェントツール、セッション認識型エージェント、ワークフロー派生ツールの解説が含まれています。

独自のチャネルミックスを構築する

同じエージェントやワークフローは、複数のチャネルで同時に活用できます。各チャネルには固有の認証ルールと表示指示を設定できながら、基盤となるターゲットやアプリケーションインフラは共有可能です。

開発者として、チャネル固有のアイデンティティをセッション ID にマッピングするリゾルバーを定義します。統一されたリゾルバーを使えば、ユーザーは Responses クライアントで会話を開始し、その後 Telegram へ移行したり、サポートされている他のあらゆるチャネル間を自由に行き来できます。会話の再スタートを必要とせず、シームレスに移動できるのが特徴です。ただし、必要な場合は各チャネルのアイデンティティを別々の名前空間に保持する設定も引き続き利用可能です。

この柔軟性はアプリケーションフレームワークにも適用されます。現在の HTTP サンプルでは FastAPI を使用していますが、ヘルパー機能はプロトコル変換と実行状態の境界で動作します。Django や Flask といった他の Python ウェブフレームワーク、既存のサービス、あるいはネイティブプロトコル SDK とも統合可能です。

今後の展望

今後はさらに機能を拡張し、対応チャネルを増やし、ヘルパーやその他の機能を追加して、エージェントのホストをより容易にする予定です。しかし、ユーザーが本当に必要とするものを作るためには、何がうまくいき、何が課題で、エンドツーエンドのシナリオを完結させるために何が必要かという声を聞きたいと考えています。ぜひ、現在運用しているアプリケーションとインフラの中で、ユーザーが必要なプロトコルやチャネルに、Agent Framework のエージェントやワークフローを接続するためにこれらのパッケージを活用してください。

上記のドキュメントと実行可能なサンプルを基に、Agent Framework を活用し、ユーザーに適したチャネル構成を組み立ててください。また、どのような機能が効果的だったか、次に必要な統合は何かについて、Python の Issue #6265 でぜひお知らせください。