Saját üzemeltetésű A2A-ügynökök

Az .NET A2A-üzemeltetési csomagok használatával elérhetővé tehet egy Agent Framework-ügynököt az ASP.NET Core-on keresztül. A teljes, többnyelvű szerverútmutatóért lásd: Hostügynökök A2A-val.

Használja a Go provider/a2aprovider csomagot a hivatalos A2A Go kiszolgálókezelőkkel. A teljes kiszolgálópéldáért lásd: A2A-t használó gazdagépügynökök.

Az Agent Framework két Python csomagot biztosít az ügynökök és munkafolyamatok üzemeltetéséhez a hivatalos A2A SDK-val:

Csomag Integrációs modell Akkor használja, ha
agent-framework-a2a Véleményezett A2AExecutor , amely átalakítja a kéréseket, futtat egy ügynököt, és közzéteszi az A2A-tevékenységeseményeket és -összetevőket. A standard Agent Framework-to-A2A viselkedést szeretné használni, és csak az A2A SDK-kiszolgálót kell összeállítania.
agent-framework-hosting-a2a Növekményes építőelemek egy alkalmazás tulajdonában lévő végrehajtóhoz. Kezdje az alapügynökkel vagy munkafolyamat-konverterekkel, és opcionálisan használja AgentA2AAdapter vagy WorkflowA2AAdapter, amelyek ezekre a konverterekre épülve natív kártyagenerálást és módérvényesítést adnak hozzá. Az alkalmazásnak saját munkamenet-leképezést, feladatáttűnést, eseménykézbesítést, összetevő-határokat, kimeneti átalakítást vagy többprotokollos gazdagépet kell tartalmaznia.

Mindkét csomag natív A2A SDK-típusokat és kiszolgálóösszetevőket használ. Az alkalmazás biztosítja a kéréskezelőt, a feladattárolót, az útvonalakat vagy az SDK-alkalmazáskészítőt, a hitelesítést és az üzembe helyezést. Ezzel agent-framework-hosting-a2aaz alkalmazás közvetlenül létrehozhatja az ügynökkártyát, vagy lehetővé teheti, hogy egy adapter hozza létre.

A véleményezett A2A-végrehajtó használata

Telepítse a agent-framework-a2a elemet, ha a beépített kiszolgálóadapter illeszkedik az életciklusához:

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

A2AExecutor implementálja az A2A SDK-t AgentExecutor. Beolvassa a felhasználói bemenetet az A2A-kérelemkörnyezetből, létrehoz egy Ügynök-keretrendszer-munkamenetet az A2A környezeti azonosítójából, streamelési vagy nem streamelési módban futtatja az ügynököt, konvertálja a támogatott kimeneti tartalmat, és közzéteszi a feladat állapotát és az összetevő eseményeit az SDK-on TaskUpdaterkeresztül.

Állítsa össze az A2A SDK-jával, DefaultRequestHandler, a feladattárolóval, az ügynökkártyával és a Starlette alkalmazással vagy egy másik támogatott szerverintegrációval. Konfigurálja a streamelést a(z) A2AExecutor(agent, stream=True) használatával, adja át a stabil ügynökfuttatási beállításokat a(z) run_kwargs segítségével, vagy származtasson alosztályt a(z) A2AExecutor osztályból, és írja felül a(z) handle_events metódust, ha eltérő kimeneti leképezésre van szüksége.

A2AExecutor hatóköre egy A2A-végpontra terjed ki, és közvetlenül kezeli az A2A-végrehajtást és a munkamenet-leképezést. Használja az üzemeltetési csomagokat, ha ugyanazt az ügynököt több protokollon keresztül is elérhetővé kell tenni egy alkalmazásban.

A szerver teljes beállításához lásd: Egy Agent Framework-ügynök elérhetővé tétele A2A-n keresztül.

Adapter használata egy alkalmazás tulajdonában lévő végrehajtóban

Telepítse a hosztolási csomagot, ha az alkalmazás a natív A2A-végrehajtót használja, de azt szeretné, hogy az Agent Framework hozza létre a publikus kártyát, és ellenőrizze a konverziókat:

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

AgentA2AAdapter elfogad egy ügynököt vagy AgentState. Aszinkron get_card metódusa levezeti a publikus nevet és leírást, alapértelmezés szerint konzervatív szöveges módokat használ, és képes natív A2A-képességeket kikövetkeztetni az Agent Framework SkillsProvider példányaiból. A kiszolgálói képességek és a támogatott felületek explicitek maradnak, mivel az ügynök metódusa run helyett az alkalmazásvégpontot írják le.

Az adapter alapértelmezés szerint elérhetővé teszi a a2a_to_run és a2a_from_run metódusokat, amelyek a konfigurált kártyamódok alapján ellenőrzik az értékeket. Az alkalmazás továbbra is az A2A-végrehajtót, a feladat életciklusát, az eseménysort, az összetevők határait, a munkamenet-szabályzatot, a hitelesítést, az útvonalakat és az üzembe helyezést birtokolja.

Ez a végrehajtó egy adaptert használ a bejövő átalakításhoz, az ügynökállapothoz és a kimenő konverzióhoz:

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 kiszolgálóbeállítás létrehozza az adaptert, létrehozza a natív AgentCardfájlt, és az alkalmazás tulajdonában lévő végrehajtót az A2A SDK kéréskezelőjével állítja össze:

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

Alkalmazás tulajdonában lévő A2A-végrehajtó létrehozása

Használja az önálló üzemeltetési segítőket, ha az alkalmazásnak közvetlen vezérlésre van szüksége a kártya létrehozása felett:

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

A segítők keretrendszersemlegesek:

  • a2a_to_run az A2A-t Message Agent Framework futtatási argumentumaivá alakítja.
  • a2a_from_run Az Ügynök-keretrendszer válaszait és a streamelési frissítéseket A2A-értékekké Part alakítja.

A végrehajtó kiválasztja a munkamenetkulcsokat, és a feladatáttűnéseket, az eseménysorokat, az összetevő azonosítóit, az üzenethatárokat és a kimenő kézbesítést birtokolja. a2a_from_run Egy lapos részlistát ad vissza, így az alkalmazás ezeket az elemeket A2A-üzenetekbe vagy összetevőkbe csoportosíthatja, és üzenetszintű metaadatokat alkalmazhat.

Az üzemeltetési beállítás támogatja a többprotokollos alkalmazásokat is. Ugyanazt az ügynökcélt és AgentState infrastruktúrát oszthatja meg az A2A, az OpenAI-válaszok, a Telegram és az MCP-útvonalak között, míg minden protokollvégpont megtartja a saját konverziós, engedélyezési és munkamenetkulcs-szabályzatát. Ez lehetővé teszi, hogy az ügyfelek egyszerre különböző protokollokon keresztül érjék el az egyik ügynököt anélkül, hogy minden végponthoz külön ügynöktelepítést hoznak létre.

Állítsa össze a segédfüggvényeket egy natív A2A SDK-végrehajtóban. Ez a minta létrehozza és frissíti az A2A-feladatokat, átalakítja a bejövő üzenetet ügynök-keretrendszer-futtatássá, megőrzi a frissített AgentState munkamenetet a stream befejezése után, és a visszaadott részeket összetevőként közzéteszi.

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 minta Starlette-t és Uvicorn-t használ, de a segítők egyikhez sem kötődnek. Az A2A-ügynökkártya és az JSON-RPC útvonalak kiszolgálásához használja az alkalmazás-keretrendszert vagy az A2A SDK-alkalmazásszerkesztőt:

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

Munkafolyamat üzemeltetése adapterrel

WorkflowA2AAdapter ugyanazt a kártyagenerálási és konverziós határt biztosítja egy munkafolyamathoz vagy WorkflowState. A munkafolyamat deklarált típusaiból konzervatív bemeneti és kimeneti módokat, vagy explicit módokat is megadhat egy alkalmazásspecifikus megjelenítéshez.

Az önálló a2a_to_workflow_run és a2a_from_workflow_run segítők beírt munkafolyamat-bemenetet és kimeneti átalakítást biztosítanak. Az adapter aszinkron a2a_to_run és szinkron a2a_from_run metódusokon keresztül teszi ezeket elérhetővé, amelyek az adapter tényleges kártyamódjai alapján végeznek ellenőrzést. A bemeneti átalakítás egy A2A szöveges, nyers vagy adatrészt fogad el a munkafolyamat egyetlen kezdő-végrehajtó bemeneti típusához, a kimeneti konverzió pedig a befejezett nyilvános munkafolyamat-kimeneteket natív A2A-részre képezi le. Hívás get_card az érvényesített kimeneti átalakítás előtt, ha az adapternek kimeneti módokra kell következtetnie.

Az alkalmazás továbbra is felelős a natív A2A-végrehajtóért, valamint az előrehaladás, a feladat állapota, az artefaktumok, az ellenőrzőpontok és az emberi közreműködéssel történő folytatás folyamatos közvetítéséért. A felhasználói bevitelre váró kérések nem konvertálódnak automatikusan, ezért a gazdaalkalmazásnak saját folytatási szabályzatot kell megvalósítania.

Biztonságos munkamenetek és feladatállapot

A2AExecutor Az A2A környezetazonosítót használja az Ügynök-keretrendszer munkamenet-azonosítójaként. Az adapteralapú és segédalapú minták az A2A-bérlőt és a környezetazonosítót kombinálják az alkalmazás által kiválasztott leképezés bemutatásához. Minden megközelítés esetében az éles környezetben futó gazdagépnek hitelesítenie kell a hívót, mielőtt az elérné az A2A-kéréskezelőt, meg kell határoznia a bérlőt és az alanyt ebből a megbízható identitásból, és engedélyeznie kell az összes feladat-, környezeti-, folytatási- és megszakítási azonosítót.

Important

Az A2A SDK alapértelmezett feladat- és push-konfigurációs tárolói memóriabeliek, és a tulajdonlás hatóköre felhasználónév szerint van meghatározva. Több-bérlős szolgáltatás esetén használjon egy olyan owner_resolver elemet, amely ugyanabból a megbízható bérlőből és alanyból származtatja a tulajdonosi hozzárendelést, és használjon tartós feladat- és munkamenet-tárolókat, ha a replikák újraindulhatnak vagy horizontálisan skálázódhatnak.

A teljes segédalapú kiszolgáló- és többügynök-példákért tekintse meg az A2A üzemeltetési mintáit. Az A2A-ügyfelek és a protokoll képességeiről lásd az A2A-ügynökszolgáltatást.

Következő lépések

Hosztügynökök A2A-val

Mélyedjen el:

  • Last updated on