A2A-agents zelf hosten

Gebruik de .NET A2A-hostingpakketten om een Agent Framework-agent beschikbaar te maken via ASP.NET Core. Zie A2A-integratie voor pakketinstallatie en een volledig servervoorbeeld.

Gebruik het Go-pakket provider/a2aprovider met de officiële A2A Go-serverhandlers. Zie A2A-integratie voor een volledig servervoorbeeld.

Agent Framework biedt twee Python pakketten voor het hosten van agents en werkstromen via de officiële A2A SDK:

Beide pakketten maken gebruik van systeemeigen A2A SDK-typen en serveronderdelen. Uw toepassing levert de aanvraaghandler, taakopslag, routes of SDK-toepassingsbouwer, verificatie en implementatie. Met agent-framework-hosting-a2a kan de toepassing de agentkaart direct samenstellen of deze door een adapter laten genereren.

Gebruik de eigenzinnige A2A-executor

Installeren agent-framework-a2a wanneer de ingebouwde serveradapter overeenkomt met uw levenscyclus:

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

A2AExecutor implementeert de A2A SDK's AgentExecutor. Het leest de gebruikersinvoer uit de A2A-aanvraagcontext, maakt een Agent Framework-sessie op basis van de A2A-context-id, voert de agent uit in de streaming- of niet-streamingmodus, converteert ondersteunde uitvoerinhoud en publiceert taakstatus- en artefactgebeurtenissen via de SDK TaskUpdater.

Stel deze samen met de A2A SDK's DefaultRequestHandler, taakopslag, agentkaart en Starlette-toepassing of een andere ondersteunde serverintegratie. Configureer streaming met A2AExecutor(agent, stream=True), geef stabiele uitvoeringsopties voor de agent door via run_kwargs, of maak een subklasse van A2AExecutor en overschrijf handle_events wanneer u een andere mapping van de uitvoer nodig hebt.

A2AExecutor is gericht op een A2A-eindpunt en beheert de A2A-uitvoering en sessietoewijzing rechtstreeks. Gebruik de hostingpakketten wanneer dezelfde agent beschikbaar moet zijn via verschillende protocollen in één toepassing.

Zie Een Agent Framework-agent beschikbaar maken via A2A voor het instellen van de volledige server.

Gebruik een adapter in een executorservice van de app

Installeer het hostingpakket wanneer uw toepassing eigenaar is van de systeemeigen A2A-executor, maar wil dat Agent Framework de openbare kaart genereert en conversies valideert:

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

AgentA2AAdapter ondersteunt een agent of AgentState. De asynchrone get_card methode leidt de openbare naam en beschrijving af, maakt standaard gebruik van conservatieve tekstmodi en kan systeemeigen A2A-vaardigheden afleiden van Agent Framework-exemplaren SkillsProvider . Servermogelijkheden en ondersteunde interfaces blijven expliciet omdat ze het toepassingseindpunt beschrijven in plaats van de methode van run de agent.

De adapter biedt standaard de methoden a2a_to_run en a2a_from_run, die waarden valideren aan de hand van de geconfigureerde kaartmodi. De toepassing heeft nog steeds de regie over de A2A-executor, de levenscyclus van taken, de eventwachtrij, artefactgrenzen, het sessiebeleid, authenticatie, routes en de uitrol.

Deze uitvoerder maakt gebruik van één adapter voor binnenkomende conversie, agentstatus en uitgaande conversie:

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

De serverinstallatie maakt de adapter, genereert de systeemeigen AgentCardadapter en stelt de uitvoerder van de app samen met de A2A SDK-aanvraaghandler:

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,
    )

Een app-eigen A2A-uitvoerder maken

Gebruik de zelfstandige hostinghelpers wanneer uw toepassing ook directe controle nodig heeft over het maken van kaarten:

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

De helpers zijn frameworkneutraal:

• a2a_to_run converteert een A2A Message naar Agent Framework-uitvoeringsargumenten.
• a2a_from_run converteert Agent Framework-antwoorden en streaming-updates naar A2A-waarden Part .

Uw uitvoerder selecteert sessiesleutels en is eigenaar van taakovergangen, gebeurteniswachtrijen, artefact-id's, berichtgrenzen en uitgaande bezorging. a2a_from_run retourneert een lijst met platte onderdelen, zodat de toepassing deze onderdelen kan groeperen in A2A-berichten of artefacten en metagegevens op berichtniveau kan toepassen.

De hostinginstallatie ondersteunt ook toepassingen met meerdere protocollen. Deel hetzelfde agentdoel en AgentState dezelfde infrastructuur in A2A-, OpenAI-antwoorden, Telegram- en MCP-routes, terwijl elk protocoleindpunt zijn eigen conversie-, autorisatie- en sessiesleutelbeleid behoudt. Hierdoor kunnen clients één agent bereiken via verschillende protocollen tegelijk zonder dat er voor elk eindpunt een afzonderlijke agentimplementatie hoeft te worden gemaakt.

Stel de helpers samen in een systeemeigen A2A SDK-uitvoerprogramma. In dit voorbeeld worden A2A-taken gemaakt en bijgewerkt, wordt het binnenkomende bericht geconverteerd naar een Agent Framework-uitvoering, wordt de bijgewerkte AgentState sessie behouden nadat de stream is voltooid en worden geretourneerde onderdelen gepubliceerd als artefacten.

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

Het voorbeeld maakt gebruik van Starlette en Uvicorn, maar de helpers zijn ook niet gebonden aan. Gebruik uw toepassingsframework of een A2A SDK-toepassingsbouwer om de A2A-agentkaart en JSON-RPC routes te bedienen:

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

Een werkstroom hosten met een adapter

WorkflowA2AAdapter biedt dezelfde afbakening voor het genereren en converteren van kaarten voor een workflow of WorkflowState. Hiermee worden conservatieve invoer- en uitvoermodi afgeleid van de gedeclareerde typen van de werkstroom of kunt u expliciete modi opgeven voor een toepassingsspecifieke weergave.

De zelfstandige a2a_to_workflow_run en a2a_from_workflow_run helpers bieden getypte werkstroominvoer en uitvoerconversie. De adapter maakt ze beschikbaar als asynchrone a2a_to_run en synchrone a2a_from_run methoden die worden gevalideerd op basis van de effectieve kaartmodi. Invoerconversie accepteert één A2A-tekstdeel, rawdeel of gegevensdeel voor het enige invoertype van de start-executor van de workflow, en uitvoerconversie zet de uitvoer van voltooide openbare workflows om in native A2A-delen. Roep get_card aan voordat de uitvoerconversie is gevalideerd wanneer de adapter uitvoermodi moet afleiden.

De toepassing blijft verantwoordelijk voor de native A2A-executor en voor het streamen van voortgang, taakstatus, artefacten, controlepunten en human-in-the-loop-voortzetting. Aanvragen voor menselijke invoer worden niet automatisch geconverteerd, dus moet de host een eigen vervolgbeleid implementeren.

Beveiligde sessies en taakstatus

A2AExecutor gebruikt de A2A-context-id als de sessie-id van agentframework. De adapter- en helpervoorbeelden combineren de A2A-tenant en context-ID om een door de toepassing geselecteerde toewijzing te laten zien. Bij elke benadering moet een productiehost de aanvrager authenticeren voordat deze de A2A-aanvraaghandler bereikt, de tenant en het subject op basis van die vertrouwde identiteit bepalen en alle taak-, context-, voortzettings- en annulering-ID's autoriseren.

Zie de voorbeelden van A2A-hosting voor een volledige helper- en multiagent-server. Zie A2A-integratie voor A2A-clients en protocolmogelijkheden.

Volgende stappen 

Ga dieper in:

• Overzicht van selfhosting
• OpenAI-antwoorden
• Telegram