Ospita autonomamente gli agenti A2A

Usare i pacchetti di hosting .NET A2A per esporre un agente di Agent Framework tramite ASP.NET Core. Vedere Integrazione A2A per l'installazione dei pacchetti e un esempio completo del server.

Usare il pacchetto Go provider/a2aprovider con i gestori server A2A Go ufficiali. Per un esempio completo del server, vedere Integrazione A2A .

Agent Framework offre due pacchetti Python per l'hosting di agenti e flussi di lavoro tramite L'SDK A2A ufficiale:

Entrambi i pacchetti usano tipi di SDK A2A nativi e componenti server. L'applicazione fornisce il gestore delle richieste, l'archivio attività, le route o il generatore di applicazioni SDK, l'autenticazione e la distribuzione. Con agent-framework-hosting-a2a, l'applicazione può costruire direttamente la scheda dell'agente o consentire a un adattatore di generarlo.

Usa l'executor A2A con convenzioni predefinite

Installa agent-framework-a2a quando l'adattatore server integrato è adatto al tuo ciclo di vita:

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

A2AExecutor implementa l'SDK AgentExecutordi A2A. Legge l'input dell'utente dal contesto della richiesta A2A, crea una sessione di Agent Framework dall'ID del contesto A2A, esegue l'agente in modalità streaming o non streaming, converte i contenuti di output supportati e pubblica gli eventi di stato dell'attività e degli artefatti tramite TaskUpdater dell'SDK.

Componilo con il DefaultRequestHandlertask store, l'agent card e l'applicazione Starlette dell'SDK A2A oppure con un'altra integrazione del server supportata. Configurare lo streaming con A2AExecutor(agent, stream=True), passare le opzioni di esecuzione dell'agente stabile tramite run_kwargso sottoclasse A2AExecutor ed eseguire l'override handle_events quando è necessario un mapping di output diverso.

A2AExecutor ha come ambito un endpoint A2A e gestisce direttamente la mappatura A2A di esecuzione e delle sessioni. Usare i pacchetti di hosting quando lo stesso agente deve essere disponibile tramite diversi protocolli in un'unica applicazione.

Per la configurazione completa del server, vedere Esporre un agente di Agent Framework su A2A.

Usare un adattatore in un executor gestito dall'app

Installa il pacchetto di hosting quando l'applicazione include l'esecutore A2A nativo ma vuoi che Agent Framework generi la scheda pubblica e convalidi le conversioni:

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

AgentA2AAdapter accetta un agente o AgentState. Il metodo asincrono get_card ricava il nome e la descrizione pubblici, usa per impostazione predefinita modalità testo conservative e può dedurre funzionalità A2A native dalle istanze di Agent Framework SkillsProvider. Le funzionalità del server e le interfacce supportate rimangono esplicite perché descrivono l'endpoint dell'applicazione anziché il metodo dell'agente run .

L'adattatore espone i metodi a2a_to_run e a2a_from_run, che per impostazione predefinita convalidano i valori rispetto alle modalità della scheda configurate. L'applicazione è ancora proprietaria dell'executor A2A, del ciclo di vita delle attività, della coda di eventi, dei limiti degli artefatti, dei criteri di sessione, dell'autenticazione, delle route e della distribuzione.

Questo esecutore utilizza un adattatore per la conversione in entrata, lo stato dell'agente e la conversione in uscita:

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

La configurazione del server crea l'adattatore, genera il relativo elemento nativo AgentCard e compone l'executor di proprietà dell'app con il gestore delle richieste dell'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,
    )

Crea un esecutore A2A gestito dall'app

Usare gli helper di hosting autonomi quando l'applicazione richiede anche il controllo diretto sulla creazione di schede:

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

Gli helper sono indipendenti dal framework:

• a2a_to_run converte un A2A Message negli argomenti di esecuzione di Agent Framework.
• a2a_from_run converte le risposte di Agent Framework e gli aggiornamenti in streaming in valori A2A Part .

L'esecutore seleziona le chiavi di sessione ed è responsabile delle transizioni delle attività, delle code di eventi, degli ID degli artefatti, dei confini dei messaggi e della consegna in uscita. a2a_from_run restituisce un elenco di parti flat in modo che l'applicazione possa raggruppare tali parti in messaggi o artefatti A2A e applicare metadati a livello di messaggio.

La configurazione di hosting supporta anche applicazioni multiprotocollo. Condividono lo stesso target dell'agente e la stessa infrastruttura AgentState tra i percorsi A2A, OpenAI Responses, Telegram e MCP, mentre ogni endpoint di protocollo mantiene i propri criteri specifici di conversione, autorizzazione e chiave di sessione. In questo modo i client raggiungono un agente tramite protocolli diversi contemporaneamente senza creare una distribuzione separata dell'agente per ogni endpoint.

Combina le funzioni helper in un esecutore nativo dell'SDK A2A. Questo esempio crea e aggiorna le attività A2A, converte il messaggio in ingresso in un'esecuzione di Agent Framework, salva in modo permanente la sessione aggiornata AgentState al termine del flusso e pubblica le parti restituite come artefatti.

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

L'esempio usa Starlette e Uvicorn, ma gli helper non sono associati a nessuno dei due. Usare il framework applicazione o un generatore di applicazioni A2A SDK per gestire la scheda dell'agente A2A e le route 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, "/"),
    ]
)

Ospitare un flusso di lavoro con un adattatore

WorkflowA2AAdapter fornisce lo stesso limite di generazione e conversione di schede per un flusso di lavoro o WorkflowState. Deduce le modalità di input e output conservativi dai tipi dichiarati del flusso di lavoro oppure è possibile fornire modalità esplicite per una rappresentazione specifica dell'applicazione.

Gli helper indipendenti a2a_to_workflow_run e a2a_from_workflow_run forniscono la conversione tipizzata dell'input e dell'output del flusso di lavoro. L'adattatore li espone come metodi asincroni a2a_to_run e sincroni a2a_from_run che verificano le modalità effettive della scheda. La conversione dell'input accetta una singola parte A2A di tipo testo, raw o dati per il singolo tipo di input start-executor del flusso di lavoro, mentre la conversione dell'output associa gli output pubblici completati del flusso di lavoro a parti A2A native. Chiamare get_card prima della conversione dell'output convalidata quando l'adapter deve dedurre le modalità di output.

L'applicazione rimane responsabile dell'executor A2A nativo e della trasmissione in streaming dell'avanzamento, dello stato del task, degli artefatti, dei checkpoint e della continuazione con intervento umano. Le richieste di input umano in sospeso non vengono convertite automaticamente, quindi l'host deve implementare i propri criteri di continuazione.

Proteggere le sessioni e lo stato delle attività

A2AExecutor usa l'ID del contesto A2A come ID sessione di Agent Framework. Gli esempi basati su adattatori e helper combinano il tenant A2A e l'ID di contesto per dimostrare un'associazione selezionata dall'applicazione. In ogni approccio, un host di produzione deve autenticare il chiamante prima che raggiunga il gestore delle richieste A2A, ricavare il tenant e il soggetto da tale identità attendibile e autorizzare tutti gli ID di attività, contesto, continuazione e annullamento.

Per esempi completi di server e multi-agente basati su helper, vedere gli esempi di hosting A2A. Per le funzionalità del protocollo e dei client A2A, vedere Integrazione A2A.

Passaggi successivi

Approfondimento:

• Panoramica dell'hosting in proprio
• Risposte OpenAI
• Telegramma