使用 .NET A2A 主機套件，透過 ASP.NET Core 公開 Agent Framework 代理程式。 請參閱 A2A 整合 以了解套件設定及完整伺服器範例。
使用 Go provider/a2aprovider 套件搭配官方的 A2A Go 伺服器處理器。 完整伺服器範例請參見 A2A 整合 。
Agent Framework 提供兩個 Python 套件，透過官方 A2A SDK 來託管代理程式與工作流程：
| Package | 整合模型 | 在下列情況下使用 |
|---|---|---|
agent-framework-a2a |
一個帶有既定設計取向的 A2AExecutor，可轉換請求、執行代理程式，並發布 A2A 任務事件和工件。 |
你想要的是標準的代理框架轉 A2A 行為，只需要組裝 A2A SDK 伺服器。 |
agent-framework-hosting-a2a |
用於由應用程式擁有的執行器的漸進式建構模組。 從基礎代理程式或工作流程轉換器開始，並可選擇使用 AgentA2AAdapter 或 WorkflowA2AAdapter；這兩者皆以這些轉換器為基礎，新增原生卡片生成與模式驗證功能。 |
你的應用程式需要擁有會話映射、任務轉換、事件傳遞、工件邊界、輸出轉換或多協定主機。 |
這兩個套件都使用原生的 A2A SDK 類型和伺服器元件。 你的應用程式提供請求處理程式、任務儲存、路由或 SDK 應用程式建構器、認證與部署。 有了 agent-framework-hosting-a2a，應用程式可以直接構建代理卡，或讓轉接器自行產生。
使用有主見的A2A執行人
當內建的伺服器配接器符合你的生命週期需求時，請安裝 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 執行與工作階段對應。 當同一代理必須透過多個協定在同一應用程式中同時提供時，請使用主機套件。
完整伺服器設定請參見 Expose an Agent Framework agent over A2A。
在應用程式擁有的執行器中使用轉接器
當您的應用程式擁有原生 A2A 執行器，但希望 Agent Framework 產生公開卡片並驗證轉換結果時，請安裝託管套件：
pip install --pre agent-framework-hosting-a2a starlette uvicorn
AgentA2AAdapter 接受代理程式或 AgentState。 其非同步 get_card 方法會產生公開名稱與描述，預設使用保守的文字模式，並能從代理框架 SkillsProvider 實例推斷原生的 A2A 技能。 伺服器能力與支援介面保持明確，因為它們描述的是應用程式端點，而非代理程式的方法 run 。
介面卡提供 a2a_to_run 和 a2a_from_run 方法，預設會根據已設定的卡片模式驗證值。 應用程式仍擁有 A2A 執行器、任務生命週期、事件佇列、工件邊界、會話政策、認證、路由及部署。
此執行器使用一個轉接器進行入站轉換、代理狀態及出站轉換：
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 請求處理程序組合應用程式自有執行器：
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 執行器
當你的應用程式也需要直接控制卡片建立時，請使用獨立主機輔助工具：
pip install --pre agent-framework-hosting-a2a starlette uvicorn
這些輔助工具與框架無關：
-
a2a_to_run將 A2AMessage轉換成 Agent Framework 執行參數。 -
a2a_from_run將代理框架的回應與串流更新轉換為 A2APart值。
你的執行者負責選擇會話金鑰，並擁有任務轉換、事件佇列、工件 ID、訊息邊界以及外發傳遞。
a2a_from_run 回傳一個平面零件清單，讓應用程式能將這些零件分組成 A2A 訊息或產物，並套用訊息層級的元資料。
主機架構也支援多協定應用程式。 在 A2A、OpenAI 回應、Telegram 及 MCP 路由間共享相同的代理目標與 AgentState 基礎設施，且每個協定端點保留自己的轉換、授權及會話金鑰政策。 這讓用戶端能同時透過不同協定連接同一個代理，而無需為每個端點建立獨立的代理部署。
在原生的 A2A SDK 執行器中組合輔助工具。 此範例會建立並更新 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 方法，並可依其有效卡片模式進行驗證。 輸入轉換接受一個 A2A 文字、原始碼或資料部分作為工作流程的單一啟動執行器輸入類型，輸出轉換則將完成的公開工作流程輸出映射為原生 A2A 部分。 當配接器必須推斷輸出模式時，請在經驗證的輸出轉換之前呼叫 get_card。
應用程式仍負責原生 A2A 執行器，以及串流進度、任務狀態、工件、檢查點及人工在迴圈中的持續執行。 尚待處理的人工輸入請求不會自動轉為接續作業，因此主機必須自行實作其接續原則。
安全會話與任務狀態
A2AExecutor 使用 A2A 上下文 ID 作為代理框架會話 ID。 基於介面卡與輔助器的範例結合了 A2A 租戶與上下文識別碼，以展示應用程式選擇的映射。 在每種方法中，生產主機必須在呼叫者抵達 A2A 請求處理程序前驗證該呼叫者，從該受信任身份中推導出租戶與主體，並授權所有任務、上下文、繼續及取消 ID。
Important
A2A SDK 的預設任務與推送配置儲存皆為記憶體內，且範圍擁有權依使用者名稱。 對於多租戶服務，請使用由同一受信任租戶和主體衍生出所有權的服務 owner_resolver ，並在複本可能重啟或擴展時使用持久任務與會話儲存。
欲了解完整的基於輔助伺服器及多代理的範例，請參閱 A2A 主機範例。 關於 A2A 用戶端及協定功能，請參見 A2A 整合。
下一步
深入探討：