Nota
O acesso a esta página requer autorização. Pode tentar iniciar sessão ou alterar os diretórios.
O acesso a esta página requer autorização. Pode tentar alterar os diretórios.
Use os pacotes de alojamento .NET A2A para expor um agente do Agent Framework através do ASP.NET Core. Veja a integração A2A para configuração de pacotes e um exemplo completo do servidor.
Use o pacote Go provider/a2aprovider com os gestores oficiais de servidores A2A Go. Veja a integração A2A para um exemplo completo de servidor.
O Agent Framework fornece dois pacotes Python para alojar agentes e fluxos de trabalho através do SDK oficial A2A:
| Package | Modelo de integração | Use-o quando |
|---|---|---|
agent-framework-a2a |
Um A2AExecutor predefinido que converte solicitações, executa um agente e publica eventos e artefactos de tarefas A2A. |
Queres o comportamento padrão do Agent Framework para A2A e só precisas de montar o servidor SDK A2A. |
agent-framework-hosting-a2a |
Blocos de construção incrementais para um executor controlado pela aplicação. Comece com o agente base ou os conversores de fluxo de trabalho e, opcionalmente, utilize AgentA2AAdapter ou WorkflowA2AAdapter, que se baseiam nesses conversores para adicionar a geração nativa de cartões e a validação de modo. |
A sua aplicação precisa de possuir o mapeamento de sessões, transições de tarefas, entrega de eventos, limites de artefactos, conversão de saída ou um host multi-protocolo. |
Ambos os pacotes utilizam tipos nativos de SDK A2A e componentes de servidor. A sua aplicação fornece o gestor de pedidos, armazenamento de tarefas, rotas ou construtor de aplicações SDK, autenticação e implementação. Com agent-framework-hosting-a2a, a aplicação pode construir a placa agente diretamente ou deixar um adaptador gerá-la.
Utilize o executor A2A predefinido
Instale agent-framework-a2a quando o adaptador de servidor integrado se adequa ao seu modelo de ciclo de vida:
pip install --pre agent-framework-a2a starlette uvicorn
A2AExecutor implementa o SDK AgentExecutordo A2A . Lê a entrada do utilizador a partir do contexto do pedido A2A, cria uma sessão do Agent Framework a partir do ID de contexto A2A, executa o agente em modo de streaming ou não streaming, converte o conteúdo de saída suportado e publica eventos de estado da tarefa e de artefactos através do TaskUpdater do SDK.
Crie-o com o SDK A2A DefaultRequestHandler, o armazenamento de tarefas, o cartão de agente e a aplicação Starlette ou outra integração de servidor suportada. Configura a transmissão em fluxo com A2AExecutor(agent, stream=True), passa as opções de execução estáveis do agente através de run_kwargs, ou cria uma subclasse de A2AExecutor e sobrescreve handle_events quando precisares de um mapeamento de saída diferente.
A2AExecutor é direcionado para um endpoint A2A e gere diretamente a execução A2A e o mapeamento de sessões. Use os pacotes de alojamento quando o mesmo agente deve estar disponível através de vários protocolos numa aplicação.
Para a configuração completa do servidor, veja Expor um agente do Agent Framework sobre A2A.
Utilize um adaptador num executor da aplicação
Instale o pacote de alojamento quando a sua aplicação possuir o executor nativo A2A mas quiser que o Agent Framework gere o cartão público e valide as conversões:
pip install --pre agent-framework-hosting-a2a starlette uvicorn
AgentA2AAdapter aceita um agente ou AgentState. O seu método assíncrono get_card determina o nome público e a descrição, utiliza modos de texto conservadores por predefinição e pode inferir capacidades nativas de A2A a partir de instâncias de Agent Framework SkillsProvider. As capacidades do servidor e as interfaces suportadas mantêm-se explícitas porque descrevem o endpoint da aplicação em vez do método do run agente.
O adaptador expõe os métodos a2a_to_run e a2a_from_run, que validam valores em relação aos modos de cartão configurados por defeito. A aplicação continua a controlar o executor A2A, o ciclo de vida da tarefa, a fila de eventos, as fronteiras dos artefactos, a política de sessão, a autenticação, as rotas e a implantação.
Este executor utiliza um adaptador para conversão de entrada, estado do agente e conversão de saída:
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.")]),
)
A configuração do servidor cria o adaptador, gera o respetivo AgentCard nativo e compõe o executor pertencente à aplicação com o processador de pedidos do SDK A2A:
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,
)
Crie um executor A2A pertencente à aplicação
Use os auxiliares de alojamento independentes quando a sua aplicação também precisar de controlo direto sobre a criação do cartão:
pip install --pre agent-framework-hosting-a2a starlette uvicorn
Os auxiliares são neutros em termos de estrutura:
-
a2a_to_runconverte um A2AMessageem argumentos executados pelo Agent Framework. -
a2a_from_runconverte respostas do Agent Framework e atualizações em streaming para valores A2APart.
O seu executor seleciona as chaves de sessão e controla as transições de tarefas, filas de eventos, IDs de artefactos, limites de mensagens e entrega de saída.
a2a_from_run retorna uma lista plana de partes para que a aplicação possa agrupar essas partes em mensagens A2A ou artefactos e aplicar metadados ao nível da mensagem.
A configuração de alojamento também suporta aplicações multi-protocolo. Partilham o mesmo destino do agente e a mesma AgentState infraestrutura entre as rotas A2A, OpenAI Responses, Telegram e MCP, ao passo que cada endpoint de protocolo mantém a sua própria política de conversão, autorização e chave de sessão. Isto permite que os clientes cheguem a um agente através de protocolos diferentes ao mesmo tempo, sem criar uma implementação de agente separada para cada endpoint.
Compor os auxiliares no executor nativo do SDK A2A. Este exemplo cria e atualiza tarefas A2A, converte a mensagem de entrada numa execução do Agent Framework, mantém a sessão atualizada AgentState após o fim do fluxo e publica as partes devolvidas como artefactos.
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.")]),
)
A amostra usa Starlette e Uvicorn, mas os ajudantes não estão ligados a nenhum dos dois. Utilize o framework da sua aplicação ou um criador de aplicações do SDK A2A para disponibilizar o cartão do agente A2A e as rotas 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, "/"),
]
)
Hospedar um fluxo de trabalho com um adaptador
WorkflowA2AAdapter fornece o mesmo limite de geração e conversão de cartões para um fluxo de trabalho ou WorkflowState. Infere modos conservativos de entrada e saída a partir dos tipos declarados pelo fluxo de trabalho, ou pode fornecer modos explícitos para uma representação específica da aplicação.
Os auxiliares autónomos a2a_to_workflow_run e a2a_from_workflow_run fornecem conversão tipada de entrada e saída de fluxo de trabalho. O adaptador disponibiliza-os como métodos assíncronos a2a_to_run e síncronos a2a_from_run, que são validados em função dos modos efetivos do cartão. A conversão de entrada aceita uma parte A2A do tipo texto, raw ou dados para o único tipo de entrada start-executor do fluxo de trabalho, e a conversão de saída mapeia as saídas públicas concluídas do fluxo de trabalho para partes A2A nativas. Chamar get_card antes da conversão de saída validada quando o adaptador tiver de inferir modos de saída.
A aplicação continua a ser responsável pelo executor nativo A2A e pela transmissão em fluxo do progresso, do estado da tarefa, dos artefactos, dos pontos de controlo e da continuação com intervenção humana. As solicitações pendentes de intervenção humana não são convertidas automaticamente, pelo que o anfitrião tem de implementar a sua própria política de continuação.
Sessões seguras e estado da tarefa
A2AExecutor utiliza o ID de contexto A2A como ID de sessão do Agent Framework. Os exemplos baseados em adaptadores e auxiliares combinam o tenant A2A e o identificador de contexto para demonstrar um mapeamento escolhido pela aplicação. Em cada abordagem, um host de produção tem de autenticar o chamador antes de este atingir o processador de pedidos A2A, derivar o tenant e o identificador do sujeito a partir dessa identidade fidedigna e autorizar todos os identificadores de tarefa, contexto, continuação e cancelamento.
Importante
Os armazenamentos predefinidos de tarefas e de configuração push do SDK A2A são propriedade em memória e de âmbito por nome de utilizador. Para um serviço multicliente, utilize um owner_resolver que obtenha a propriedade a partir do mesmo cliente e sujeito fidedignos e utilize armazenamentos persistentes para tarefas e sessões quando as réplicas puderem reiniciar ou aumentar horizontalmente.
Para obter exemplos completos de servidor baseado em helpers e de vários agentes, consulte os exemplos de alojamento do A2A. Para clientes A2A e capacidades de protocolo, veja integração A2A.
Passos seguintes
Vai mais fundo: