.NET A2A ホスティング パッケージを使用して、ASP.NET Coreを介して Agent Framework エージェントを公開します。 パッケージのセットアップと完全なサーバーの例については、 A2A 統合 を参照してください。
Go provider/a2aprovider パッケージは、公式の A2A Go サーバー ハンドラーと共に使用します。 完全なサーバーの例については、 A2A 統合 を参照してください。
Agent Framework には、公式の A2A SDK を介してエージェントとワークフローをホストするための 2 つのPython パッケージが用意されています。
| Package | 統合モデル | 次の場合に使用します。 |
|---|---|---|
agent-framework-a2a |
リクエストを変換し、エージェントを実行し、A2Aタスクのイベントと成果物を発行する、独自の設計思想に基づくA2AExecutor。 |
標準の Agent Framework から A2A への動作が必要であり、A2A SDK サーバーをアセンブルするだけで済みます。 |
agent-framework-hosting-a2a |
アプリ管理のエグゼキューターのための段階的な構成要素。 基本的なエージェントまたはワークフロー コンバーターから開始し、必要に応じて AgentA2AAdapter または WorkflowA2AAdapter を使用します。これらのコンバーターを基にしてネイティブ カード生成とモード検証を追加します。 |
アプリケーションは、セッション マッピング、タスク遷移、イベント配信、成果物の境界、出力変換、またはマルチプロトコル ホストを所有する必要があります。 |
どちらのパッケージも、ネイティブ A2A SDK の種類とサーバー コンポーネントを使用します。 アプリケーションは、要求ハンドラー、タスク ストア、ルート、または SDK アプリケーション ビルダー、認証、デプロイを提供します。
agent-framework-hosting-a2aを使用すると、アプリケーションはエージェント カードを直接構築することも、アダプターで生成することもできます。
設計方針が明確な A2A Executor を使用する
組み込みのサーバー アダプターがライフサイクルと一致する場合は、 agent-framework-a2a をインストールします。
pip install --pre agent-framework-a2a starlette uvicorn
A2AExecutor は、A2A SDK の AgentExecutorを実装します。 A2A 要求コンテキストからユーザー入力を読み取り、A2A コンテキスト ID からエージェント フレームワーク セッションを作成し、ストリーミング モードまたは非ストリーミング モードでエージェントを実行し、サポートされている出力コンテンツを変換して、SDK の TaskUpdaterを介してタスクの状態と成果物イベントを発行します。
A2A SDK の DefaultRequestHandler、タスク ストア、エージェント カード、Starlette アプリケーション、またはサポートされている別のサーバー統合を使用して作成します。
A2AExecutor(agent, stream=True)を使用してストリーミングを構成し、run_kwargsを介して安定したエージェント実行オプションを渡すか、A2AExecutorをサブクラス化して、別の出力マッピングが必要な場合はhandle_eventsをオーバーライドします。
A2AExecutor は A2A エンドポイントにスコープを設定し、A2A の実行とセッション マッピングを直接管理します。 1 つのアプリケーションで複数のプロトコルを使用して同じエージェントを使用できる必要がある場合は、ホスティング パッケージを使用します。
サーバーの完全なセットアップについては、「 A2A 経由でエージェント フレームワーク エージェントを公開する」を参照してください。
アプリ所有の Executor でアダプターを使用する
アプリケーションがネイティブ A2A Executor を所有しているが、Agent Framework でパブリック カードを生成して変換を検証する場合は、ホスティング パッケージをインストールします。
pip install --pre agent-framework-hosting-a2a starlette uvicorn
AgentA2AAdapter はエージェントまたは AgentStateを受け入れます。 その非同期 get_card メソッドは、パブリック名と説明を派生させ、既定では保守的なテキスト モードを使用し、Agent Framework SkillsProvider インスタンスからネイティブ A2A スキルを推論できます。 サーバーの機能とサポートされているインターフェイスは、エージェントの run メソッドではなくアプリケーション エンドポイントを記述するため、明示的なままです。
アダプターは、既定で構成されたカード モードに対して値を検証する a2a_to_run メソッドと a2a_from_run メソッドを公開します。 アプリケーションは引き続き A2A Executor、タスク ライフサイクル、イベント キュー、成果物の境界、セッション ポリシー、認証、ルート、デプロイを所有しています。
この Executor は、受信変換、エージェントの状態、および送信変換に 1 つのアダプターを使用します。
class AppAgentExecutor(AgentExecutor):
"""Native A2A SDK executor composed with Agent Framework conversion helpers."""
def __init__(self, adapter: AgentA2AAdapter[Any]) -> None:
self.adapter = adapter
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.context_id is None:
raise ValueError("A2A context id is required")
updater = TaskUpdater(event_queue, context.task_id or "", context.context_id)
await updater.cancel()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.message is None or context.context_id is None:
raise ValueError("A2A message and context id are required")
task = context.current_task
if task is None:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, context.context_id)
await updater.submit()
try:
await updater.start_work()
run = self.adapter.a2a_to_run(context.message, stream=True)
agent = await self.adapter.state.get_target()
# Demo-only key: the outer server must authenticate and authorize these protocol IDs for multi-user use.
session_id = f"a2a:{context.tenant}:{context.context_id}"
session = await self.adapter.state.get_or_create_session(session_id)
if not run["stream"]:
raise RuntimeError("This executor requires streaming run arguments.")
stream = agent.run( # pyright: ignore[reportCallIssue]
run["messages"],
session=session,
options=run["options"],
stream=run["stream"],
)
default_artifact_id = uuid.uuid4().hex
streamed_artifact_ids: set[str] = set()
async for update in stream:
parts = self.adapter.a2a_from_run(update)
if parts:
artifact_id = update.message_id or default_artifact_id
await updater.add_artifact(
parts=parts,
artifact_id=artifact_id,
append=True if artifact_id in streamed_artifact_ids else None,
)
streamed_artifact_ids.add(artifact_id)
final_response = await stream.get_final_response()
if not streamed_artifact_ids:
parts = self.adapter.a2a_from_run(final_response)
if parts:
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=updater.new_agent_message(parts),
)
await self.adapter.state.set_session(session_id, session)
await updater.complete()
except asyncio.CancelledError:
await updater.update_status(state=TaskState.TASK_STATE_CANCELED)
except Exception:
logger.exception("A2A agent execution failed.")
await updater.update_status(
state=TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text="Agent execution failed.")]),
)
サーバーセットアップでは、アダプターを作成し、ネイティブ AgentCardを生成し、A2A SDK 要求ハンドラーを使用してアプリ所有の Executor を作成します。
if __name__ == "__main__":
flight_skill = InlineSkill(
frontmatter=SkillFrontmatter(
name="flight-booking",
description="Search and book flights across Europe.",
),
instructions="Help users search and book flights across Europe.",
)
hotel_skill = InlineSkill(
frontmatter=SkillFrontmatter(
name="hotel-booking",
description="Search and book hotels across Europe.",
),
instructions="Help users search and book hotels across Europe.",
)
agent = Agent(
client=OpenAIChatClient(),
name="Europe Travel Agent",
description="Helps users search and book flights and hotels across Europe.",
instructions="You are a helpful Europe Travel Agent.",
context_providers=[SkillsProvider([flight_skill, hotel_skill])],
)
state = AgentState(agent)
adapter = AgentA2AAdapter(
state,
version="1.0.0",
capabilities=AgentCapabilities(streaming=True),
supported_interfaces=[AgentInterface(url="http://localhost:9999/", protocol_binding="JSONRPC")],
)
public_agent_card = asyncio.run(adapter.get_card())
request_handler = DefaultRequestHandler(
agent_executor=AppAgentExecutor(adapter),
task_store=InMemoryTaskStore(),
agent_card=public_agent_card,
)
アプリ所有の A2A Executor をビルドする
アプリケーションでカードの作成を直接制御する必要がある場合は、スタンドアロン ホスティング ヘルパーを使用します。
pip install --pre agent-framework-hosting-a2a starlette uvicorn
ヘルパーはフレームワークに依存しません。
-
a2a_to_runは、A2AMessageを Agent Framework の実行引数に変換します。 -
a2a_from_runは、Agent Framework の応答とストリーミング更新を A2APart値に変換します。
Executor はセッション キーを選択し、タスクの遷移、イベント キュー、成果物 ID、メッセージ境界、および送信配信を所有します。
a2a_from_run はフラット パーツ リストを返します。そのため、アプリケーションはこれらのパーツを A2A メッセージまたは成果物にグループ化し、メッセージ レベルのメタデータを適用できます。
ホスティング セットアップでは、マルチプロトコル アプリケーションもサポートされます。 A2A、OpenAI Responses、Telegram、MCP ルート間で同じエージェント ターゲットと AgentState インフラストラクチャを共有し、各プロトコル エンドポイントは独自の変換、承認、およびセッション キー ポリシーを保持します。 これにより、クライアントは、エンドポイントごとに個別のエージェントデプロイを作成することなく、異なるプロトコルを使用して 1 つのエージェントに同時にアクセスできます。
ネイティブ A2A SDK Executor でヘルパーを作成します。 このサンプルでは、A2A タスクを作成して更新し、受信メッセージを Agent Framework の実行に変換し、更新された AgentState セッションをストリームの完了後に保持し、返された部分を成果物として発行します。
class AppAgentExecutor(AgentExecutor, Generic[AgentT]):
"""Native A2A SDK executor composed with Agent Framework conversion helpers."""
def __init__(self, state: AgentState[AgentT]) -> None:
self.state = state
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.context_id is None:
raise ValueError("A2A context id is required")
updater = TaskUpdater(event_queue, context.task_id or "", context.context_id)
await updater.cancel()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.message is None or context.context_id is None:
raise ValueError("A2A message and context id are required")
task = context.current_task
if task is None:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, context.context_id)
await updater.submit()
try:
await updater.start_work()
run = a2a_to_run(context.message, stream=True)
agent = await self.state.get_target()
# Demo-only key: the outer server must authenticate and authorize these protocol IDs for multi-user use.
session_id = f"a2a:{context.tenant}:{context.context_id}"
session = await self.state.get_or_create_session(session_id)
if not run["stream"]:
raise RuntimeError("This executor requires streaming run arguments.")
stream = agent.run( # pyright: ignore[reportCallIssue]
run["messages"],
session=session,
options=run["options"],
stream=run["stream"],
)
default_artifact_id = uuid.uuid4().hex
streamed_artifact_ids: set[str] = set()
async for update in stream:
parts = a2a_from_run(update)
if parts:
artifact_id = update.message_id or default_artifact_id
await updater.add_artifact(
parts=parts,
artifact_id=artifact_id,
append=True if artifact_id in streamed_artifact_ids else None,
)
streamed_artifact_ids.add(artifact_id)
final_response = await stream.get_final_response()
if not streamed_artifact_ids:
parts = a2a_from_run(final_response)
if parts:
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=updater.new_agent_message(parts),
)
await self.state.set_session(session_id, session)
await updater.complete()
except CancelledError:
await updater.update_status(state=TaskState.TASK_STATE_CANCELED)
except Exception:
logger.exception("A2A agent execution failed.")
await updater.update_status(
state=TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text="Agent execution failed.")]),
)
このサンプルでは Starlette と Uvicorn を使用していますが、ヘルパーはどちらにも関連付けされていません。 アプリケーション フレームワークまたは A2A SDK アプリケーション ビルダーを使用して、A2A エージェント カードと JSON-RPC ルートを提供します。
# Create the Agent Framework agent for the chosen type
agent_factory = AGENT_FACTORIES[args.agent_type]
agent = agent_factory(client)
state = AgentState(agent)
# Build the A2A server components
url = f"http://{args.host}:{args.port}/"
agent_card = AGENT_CARD_FACTORIES[args.agent_type](url)
executor = AppAgentExecutor(state)
task_store = InMemoryTaskStore()
request_handler = DefaultRequestHandler(
agent_executor=executor,
task_store=task_store,
agent_card=agent_card,
)
app = Starlette(
routes=[
*create_agent_card_routes(agent_card),
*create_jsonrpc_routes(request_handler, "/"),
]
)
アダプターを使用してワークフローをホストする
WorkflowA2AAdapter は、ワークフローまたは WorkflowStateに対して同じカード生成と変換の境界を提供します。 これは、ワークフローの宣言された型から保守的な入力モードと出力モードを推論するか、アプリケーション固有の表現に明示的なモードを指定できます。
スタンドアロンの a2a_to_workflow_run および a2a_from_workflow_run ヘルパーは、型指定されたワークフローの入力と出力の変換を提供します。 アダプターは、有効なカード モードに対して検証する非同期 a2a_to_run および同期 a2a_from_run メソッドとして公開します。 入力変換では、ワークフローの単一の start-executor 入力タイプに対して、1 つの A2A テキスト、raw、またはデータ パートを受け付け、出力変換では、完了済みの公開ワークフロー出力をネイティブ A2A パートに対応付けます。 アダプターが出力モードを推論する必要がある場合は、検証済みの出力変換の前に get_card を呼び出します。
アプリケーションは、ネイティブ A2A Executor とストリーミングの進行状況、タスクの状態、成果物、チェックポイント、および人間のループ内継続に対して引き続き責任を負います。 保留中の人間入力要求は自動的に変換されないため、ホストは独自の継続ポリシーを実装する必要があります。
セッションとタスクの状態をセキュリティで保護する
A2AExecutor では、エージェント フレームワークのセッション ID として A2A コンテキスト ID が使用されます。 アダプター ベースのサンプルとヘルパー ベースのサンプルでは、A2A テナントとコンテキスト ID を組み合わせて、アプリケーションで選択したマッピングを示します。 どの方法でも、運用ホストは、A2A 要求ハンドラーに到達する前に呼び出し元を認証し、その信頼された ID からテナントとサブジェクトを派生させ、すべてのタスク、コンテキスト、継続、および取り消し ID を承認する必要があります。
Important
A2A SDK の既定のタスク ストアとプッシュ構成ストアはインメモリで、所有権のスコープはユーザー名ごとに管理されます。 マルチテナント サービスの場合は、同じ信頼されたテナントとサブジェクトから所有権を取得する owner_resolver を使用し、レプリカが再起動またはスケールアウトできる場合は永続的なタスクとセッション ストアを使用します。
ヘルパー ベースのサーバーとマルチエージェントの完全な例については、 A2A ホスティングのサンプルを参照してください。 A2A クライアントとプロトコルの機能については、 A2A の統合に関するセクションを参照してください。
次のステップ
より深く進む: