Egenhostade A2A-agenter

Använd .NET A2A-värdpaket för att exponera en Agent Framework-agent via ASP.NET Core. Se A2A-integrering för paketkonfiguration och ett komplett serverexempel.

Använd Go-paketet provider/a2aprovider med de officiella A2A Go-serverhantarna. Se A2A-integrering för ett komplett serverexempel.

Agent Framework tillhandahåller två Python paket för värdagenter och arbetsflöden via den officiella A2A SDK:t:

Package	Integrationsmodell	Använd den när
`agent-framework-a2a`	En åsikt `A2AExecutor` som konverterar begäranden, kör en agent och publicerar A2A-aktivitetshändelser och artefakter.	Du vill ha standardbeteendet Agent Framework-to-A2A och behöver bara montera A2A SDK-servern.
`agent-framework-hosting-a2a`	Inkrementella byggstenar för en appägd exekutor. Börja med den grundläggande agenten eller arbetsflödeskonverterarna och använd `AgentA2AAdapter` eller `WorkflowA2AAdapter`, som bygger på dessa konverterare för att lägga till inbyggd kortgenerering och lägesverifiering.	Ditt program måste äga sessionsmappning, aktivitetsövergångar, händelseleverans, artefaktgränser, utdatakonvertering eller en värd för flera protokoll.

Båda paketen använder interna A2A SDK-typer och serverkomponenter. Din applikation tillhandahåller förfrågningshanteraren, uppgiftslagret, rutter eller SDK-byggaren för applikationer, autentisering och driftsättning. Med agent-framework-hosting-a2a kan applikationen konstruera agentkortet direkt eller låta en adapter generera det.

Använd den förkonfigurerade A2A-exekveraren

Installera agent-framework-a2a när den inbyggda serveradaptern passar din livscykel:

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

A2AExecutor implementerar A2A SDK:s AgentExecutor. Den läser användarindata från A2A-begärandekontexten, skapar en Agent Framework-session från A2A-kontext-ID:t, kör agenten i strömnings- eller icke-direktuppspelningsläge, konverterar utdatainnehåll som stöds och publicerar aktivitetsstatus och artefakthändelser via SDK:ns TaskUpdater.

Bygg det med A2A SDK:s DefaultRequestHandler, uppgiftslagring, agentkort och Starlette-applikation eller en annan serverintegrering som stöds. Konfigurera strömning med A2AExecutor(agent, stream=True), skicka stabila agentkörningsalternativ via run_kwargseller underklass A2AExecutor och åsidosätt handle_events när du behöver en annan utdatamappning.

A2AExecutor är begränsad till en A2A-slutpunkt och hanterar dess A2A-körning och sessionsmappning direkt. Använd värdpaketen när samma agent måste vara tillgänglig via flera protokoll i ett program.

Fullständig serverkonfiguration finns i Exponera en Agent Framework-agent över A2A.

Använd en adapter i en appägd executor

Installera värdpaketet när ditt program äger den interna A2A-kören men vill att Agent Framework ska generera det offentliga kortet och verifiera konverteringar:

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

AgentA2AAdapter accepterar en agent eller AgentState. Dess asynkrona get_card metod härleder det offentliga namnet och beskrivningen, använder konservativa textlägen som standard och kan härleda interna A2A-kunskaper från Agent Framework-instanser SkillsProvider . Serverfunktioner och gränssnitt som stöds förblir explicita eftersom de beskriver programslutpunkten i stället för agentens run metod.

Adaptern exponerar a2a_to_run och a2a_from_run metoder som verifierar värden mot de konfigurerade kortlägena som standard. Applikationen äger fortfarande A2A-köraren, uppgiftens livscykel, händelsekön, artefaktgränser, sessionspolicy, autentisering, rutter och driftsättning.

Den här exekveraren använder en adapter för inkommande konvertering, agenttillstånd och utgående konvertering:

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

Serverkonfigurationen skapar adaptern, genererar dess egna AgentCard och sätter samman den appägda exekveraren med A2A SDK:s begärandehanterare:

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

Skapa en A2A-exekverare som ägs av appen

Använd de fristående värdhjälparna när ditt program också behöver direkt kontroll över skapande av kort:

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

Hjälparna är ramverksneutrala:

a2a_to_run konverterar en A2A Message till Agent Framework-körningsargument.
a2a_from_run konverterar Agent Framework-svar och direktuppspelningsuppdateringar till A2A-värden Part .

Din exekutor väljer sessionsnycklar och ansvarar för uppgiftsövergångar, händelseköer, artefakt-ID:n, meddelandegränser och utgående leverans. a2a_from_run returnerar en platt dellista så att programmet kan gruppera dessa delar i A2A-meddelanden eller artefakter och tillämpa metadata på meddelandenivå.

Värdinstallationen stöder även program med flera protokoll. Dela samma agentmål och AgentState infrastruktur över A2A-, OpenAI-svars-, Telegram- och MCP-vägar, medan varje protokollslutpunkt behåller sin egen princip för konvertering, auktorisering och sessionsnyckel. På så sätt kan klienter nå en agent via olika protokoll samtidigt utan att skapa en separat agentdistribution för varje slutpunkt.

Sammanställ hjälpfunktionerna i en inbyggd A2A SDK-exekverare. Det här exemplet skapar och uppdaterar A2A-uppgifter, konverterar det inkommande meddelandet till en Agent Framework-körning, bevarar den uppdaterade AgentState sessionen när strömmen är klar och publicerar returnerade delar som artefakter.

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

Exemplet använder Starlette och Uvicorn, men hjälpfunktionerna är inte knutna till någon av dem. Använd ditt programramverk eller en A2A SDK-programbyggare för att hantera A2A-agentkortet och JSON-RPC vägar:

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

Värd för ett arbetsflöde med en adapter

WorkflowA2AAdapter ger samma kortgenererings- och konverteringsgräns för ett arbetsflöde eller WorkflowState. Den härleder konservativa indata- och utdatalägen från arbetsflödets deklarerade typer, eller så kan du ange explicita lägen för en programspecifik representation.

De fristående hjälpfunktionerna a2a_to_workflow_run och a2a_from_workflow_run erbjuder konvertering av in- och utdata för typade arbetsflöden. Adaptern exponerar dem som asynkrona a2a_to_run och synkrona a2a_from_run metoder som validerar mot dess effektiva kortlägen. Indatakonvertering accepterar en A2A-text, rå eller datadel för arbetsflödets enda start-executor-indatatyp och utdatakonverteringskartor slutförde offentliga arbetsflödesutdata till interna A2A-delar. Anropa get_card före verifierad utdatakonvertering när adaptern måste härleda utdatalägen.

Applikationen är fortfarande ansvarig för den inbyggda A2A-exekveraren och för strömningsförlopp, uppgiftsstatus, artefakter, kontrollpunkter och fortsättning med mänsklig medverkan. Väntande begäranden om inmatning från människa konverteras inte automatiskt, så värdprogrammet måste implementera en egen policy för hur processen ska fortsätta.

Säkra sessioner och aktivitetstillstånd

A2AExecutor använder A2A-kontext-ID:t som Agent Framework-sessions-ID. De adapterbaserade och hjälparbaserade exemplen kombinerar A2A-klientorganisation och kontext-ID för att demonstrera en programstyrd mappning. I alla tillvägagångssätt måste en produktionsvärd autentisera anroparen innan den når A2A-hanteraren för begäranden, härleda klientorganisationen och subjektet från den betrodda identiteten och auktorisera alla uppgifts-, kontext-, fortsättnings- och annullerings-ID:n.

Important

A2A SDK:s standardlager för uppgifter och pushkonfiguration lagras i minnet och har användarnamn som avgränsning för ägarskap. För en flerklienttjänst använder du en owner_resolver som härleder ägandet från samma betrodda klientorganisation och subject, och använder beständiga uppgifts- och sessionslager när repliker kan starta om eller skala ut.

En fullständig hjälpbaserad server och flera agentexempel finns i A2A-värdexemplen. Information om A2A-klienter och protokollfunktioner finns i A2A-integrering.

Nästa steg

Läs mer om A2A-integrering

Gå djupare:

Feedback

Var den här sidan till hjälp?

Last updated on 2026-07-23