Hospede você mesmo agentes A2A

Use os pacotes de hospedagem A2A do .NET para expor um agente do Agent Framework por meio do ASP.NET Core. Consulte a integração A2A para a instalação do pacote e um exemplo de servidor completo.

Use o pacote Go provider/a2aprovider com os manipuladores de servidor A2A Go oficiais. Consulte a integração A2A para obter um exemplo de servidor completo.

O Agent Framework fornece dois pacotes Python para hospedar agentes e fluxos de trabalho por meio do SDK oficial do A2A:

Ambos os pacotes usam tipos de SDK A2A nativos e componentes de servidor. Seu aplicativo fornece o manipulador de solicitações, o repositório de tarefas, as rotas ou o construtor de aplicativos do SDK, a autenticação e a implantação. Com agent-framework-hosting-a2a, o aplicativo pode construir o cartão do agente diretamente ou permitir que um adaptador o gere.

Usar o executor A2A opinativo

Instale agent-framework-a2a quando o adaptador de servidor integrado se adequar ao seu ciclo de vida:

pip install --pre agent-framework-a2a starlette uvicorn

A2AExecutor implementa o AgentExecutor do SDK A2A. Ele lê a entrada do usuário do contexto da solicitação A2A, cria uma sessão do Agent Framework a partir do ID de contexto A2A, executa o agente em modo de streaming ou sem streaming, converte o conteúdo de saída compatível e publica eventos de status da tarefa e de artefato por meio de TaskUpdater do SDK.

Compile-o com o SDK do A2A DefaultRequestHandler, o armazenamento de tarefas, o cartão de agente e o aplicativo Starlette ou outra integração de servidor compatível. Configure a transmissão em fluxo com A2AExecutor(agent, stream=True), passe opções estáveis de execução do agente por meio de run_kwargs, ou crie uma subclasse de A2AExecutor e substitua handle_events quando precisar de um mapeamento de saída diferente.

A2AExecutor é restrito a um endpoint A2A e gerencia diretamente sua execução A2A e o mapeamento de sessão. Use os pacotes de hospedagem quando o mesmo agente precisar estar disponível por meio de vários protocolos em um aplicativo.

Para obter a configuração completa do servidor, consulte Expor um agente do Agent Framework por A2A.

Use um adaptador em um executor do aplicativo

Instale o pacote de hospedagem quando seu aplicativo possui o executor A2A nativo, mas deseja 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. Seu método assíncrono get_card deriva o nome público e a descrição, usa modos de texto conservadores por padrão e pode inferir habilidades A2A nativas de instâncias do Agent Framework SkillsProvider . As capacidades do servidor e as interfaces suportadas permanecem explícitas porque descrevem o endpoint do aplicativo, e não o método do agente run.

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 padrão. O aplicativo ainda possui o executor A2A, o ciclo de vida da tarefa, a fila de eventos, os limites do artefato, a política de sessão, a autenticação, as rotas e a implantação.

Esse executor usa 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 seu AgentCard nativo e compõe o executor do próprio aplicativo com o manipulador de requisições 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,
    )

Criar um executor A2A pertencente ao aplicativo

Use os auxiliares de hospedagem autônomos quando seu aplicativo também precisar de controle direto sobre a criação de cartão:

pip install --pre agent-framework-hosting-a2a starlette uvicorn

Os auxiliares são neutros em termos de estrutura:

• a2a_to_run converte um A2A Message em argumentos de execução do Agent Framework.
• a2a_from_run converte respostas do Agent Framework e atualizações de streaming em valores A2A Part .

Seu executor seleciona chaves de sessão e é responsável por transições de tarefas, filas de eventos, IDs de artefatos, limites das mensagens e entrega de saída. a2a_from_run retorna uma lista de partes simples para que o aplicativo possa agrupar essas partes em mensagens A2A ou artefatos e aplicar metadados no nível da mensagem.

A configuração de hospedagem também dá suporte a aplicativos de vários protocolos. Compartilhe o mesmo destino do agente e a infraestrutura AgentState entre as rotas A2A, OpenAI Responses, Telegram e MCP, enquanto cada endpoint de protocolo mantém sua própria política de conversão, autorização e chaves de sessão. Isso permite que os clientes acessem um agente por diferentes protocolos ao mesmo tempo, sem criar uma implantação separada do agente para cada ponto de extremidade.

Compor os auxiliares em um executor nativo do SDK A2A. Este exemplo cria e atualiza tarefas A2A, converte a mensagem de entrada em uma execução do Agent Framework, persiste a sessão atualizada AgentState após a conclusão do fluxo e publica partes retornadas como artefatos.

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.")]),
            )

O exemplo usa Starlette e Uvicorn, mas os auxiliares também não estão vinculados. Use seu framework de aplicativos ou um criador de aplicativos com o 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ão para um fluxo de trabalho ou WorkflowState. Ele infere modos de entrada e saída conservadores dos tipos declarados do fluxo de trabalho ou você pode fornecer modos explícitos para uma representação específica do aplicativo.

Os auxiliares autônomos a2a_to_workflow_run e a2a_from_workflow_run fornecem conversão tipada de entrada e saída do fluxo de trabalho. O adaptador os disponibiliza como métodos assíncronos a2a_to_run e síncronos a2a_from_run que validam com base nos modos efetivos do cartão. A conversão de entrada aceita uma parte A2A de texto, bruta ou de dados para o único tipo de entrada do executor inicial 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. Chame get_card antes da conversão de saída validada quando o adaptador deve inferir modos de saída.

A aplicação continua responsável pelo executor A2A nativo e pela transmissão de progresso, pelo status da tarefa, pelos artefatos, pelos pontos de verificação e pela continuação com intervenção humana. As solicitações de entrada humana pendentes não são convertidas automaticamente, portanto, o host deve implementar sua própria política de continuação.

Sessões seguras e estado da tarefa

A2AExecutor usa a ID de contexto A2A como a ID da sessão do Agent Framework. Os exemplos baseados em adaptador e em helper combinam o tenant A2A e a ID do contexto para demonstrar um mapeamento selecionado pela aplicação. Em cada abordagem, um host de produção deve autenticar o chamador antes que a solicitação chegue ao manipulador de solicitações A2A, determinar o tenant e o sujeito com base nessa identidade confiável e autorizar todos os IDs de tarefa, contexto, continuação e cancelamento.

Para obter um servidor baseado em auxiliar completo e exemplos de vários agentes, consulte os exemplos de hospedagem A2A. Para clientes A2A e recursos de protocolo, consulte a integração A2A.

Próximas Etapas 

Vá mais fundo:

• Visão geral da hospedagem própria
• Respostas OpenAI
• Telegrama