Hostujte A2A agenty sami

Pomocí balíčků hostování .NET A2A zpřístupňte agenta Agent Framework prostřednictvím ASP.NET Core. Kompletní průvodce vícejazyčným serverem najdete v tématu Agenti hostitelů s A2A .

Balíček Go provider/a2aprovider použijte s oficiálními obslužnými rutinami serveru A2A Go. Kompletní příklad serveru najdete v tématu Agenti hostitelů s A2A .

Agent Framework poskytuje dva balíčky Python pro hostování agentů a pracovních postupů prostřednictvím oficiální sady A2A SDK:

Package	Integrační model	Použijte ji, když
agent-framework-a2a	Předdefinované A2AExecutor, které převádí požadavky, spouští agenta a publikuje události a artefakty úloh A2A.	Chcete standardní chování Agent Framework-to-A2A a stačí sestavit pouze server A2A SDK.
agent-framework-hosting-a2a	Inkrementální stavební bloky pro executor vlastněný aplikací. Začněte se základním agentem nebo převaděči pracovních postupů a volitelně použijte AgentA2AAdapter nebo WorkflowA2AAdapter, který staví na těchto převaděčích pro přidání nativního generování karet a ověřování režimu.	Vaše aplikace musí sama zajišťovat mapování relací, přechody mezi úlohami, předávání událostí, vymezení artefaktů, převod výstupu nebo fungovat jako víceprotokolový hostitel.

Oba balíčky používají nativní typy A2A SDK a součásti serveru. Vaše aplikace poskytuje obsluhu požadavků, úložiště úloh, směrování nebo tvůrce aplikace SDK, ověřování a nasazení. Pomocí agent-framework-hosting-a2a může aplikace kartu agenta vytvořit přímo nebo ji nechat vygenerovat adaptérem.

Použití názorného exekutoru A2A

Nainstalujte agent-framework-a2a, pokud integrovaný serverový adaptér vyhovuje vašemu životnímu cyklu:

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

A2AExecutor implementuje sadu A2A SDK AgentExecutor. Načte uživatelský vstup z kontextu požadavku A2A, vytvoří relaci rozhraní Agent Framework z ID kontextu A2A, spustí agenta v režimu streamování nebo bez streamování, převede podporovaný výstupní obsah a publikuje stav úlohy a události artefaktů prostřednictvím sady SDK TaskUpdater.

Sestavte ji pomocí sady SDK A2A DefaultRequestHandler, úložiště úloh, karty agenta a aplikace Starlette nebo jiné podporované serverové integrace. Nakonfigurujte streamování pomocí A2AExecutor(agent, stream=True), předejte možnosti spuštění stabilního agenta prostřednictvím run_kwargs nebo vytvořte podtřídu A2AExecutor a přepište handle_events, pokud potřebujete odlišné mapování výstupu.

A2AExecutor je omezen na koncový bod A2A a přímo spravuje své spouštění A2A a mapování relací. Hostitelské balíčky použijte, pokud musí být stejný agent dostupný prostřednictvím několika protokolů v jedné aplikaci.

Kompletní nastavení serveru najdete v tématu Zveřejnění agenta rozhraní Agent Framework přes A2A.

Použijte adaptér ve spouštěči úloh patřícím aplikaci

Balíček hostování nainstalujte, pokud vaše aplikace vlastní nativní exekutor A2A, ale chce, aby rozhraní Agent Framework vygenerovalo veřejnou kartu a ověřilo převody:

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

AgentA2AAdapter přijímá agenta nebo AgentState. Jeho asynchronní get_card metoda odvozuje veřejný název a popis, používá ve výchozím nastavení režimy konzervativního textu a dokáže odvodit nativní dovednosti A2A z instancí agent Framework SkillsProvider . Možnosti serveru a podporovaná rozhraní zůstávají explicitní, protože místo metody agenta run popisují koncový bod aplikace.

Adaptér zveřejňuje a2a_to_run a a2a_from_run metody, které ve výchozím nastavení ověřují hodnoty v nakonfigurovaných režimech karet. Aplikace stále spravuje exekutor A2A, životní cyklus úloh, frontu událostí, hranice artefaktů, zásady relací, ověřování, směrování a nasazení.

Tento executor používá jeden adaptér pro vstupní převod, stav agenta a výstupní převod:

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

Nastavení serveru vytvoří adaptér, vygeneruje jeho nativní AgentCard a sestaví aplikací vlastněný exekutor s obslužnou rutinou požadavků A2A SDK:

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

Vytvořit executor A2A vlastněný aplikací

Použijte samostatné pomocné funkce pro hostování, pokud vaše aplikace také potřebuje přímou kontrolu nad vytvářením karet:

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

Pomocné funkce jsou nezávislé na frameworku:

• a2a_to_run převede A2A Message na argumenty spuštění pro Agent Framework.
• a2a_from_run převádí odpovědi z rozhraní Agent Framework a průběžné aktualizace na hodnoty A2A Part.

Váš executor vybírá klíče relace a spravuje přechody úloh, fronty událostí, ID artefaktů, hranice zpráv a odchozí doručení. a2a_from_run vrátí plochý seznam částí, aby aplikace mohly tyto části seskupit do zpráv nebo artefaktů A2A a použít metadata na úrovni zpráv.

Nastavení hostování také podporuje aplikace s více protokoly. Sdílet stejný cíl agenta a infrastrukturu AgentState napříč trasami A2A, OpenAI Responses, Telegramu a MCP, přičemž každý koncový bod protokolu má vlastní zásady pro převod, autorizaci a klíče relace. Díky tomu se klienti dostanou k jednomu agentu prostřednictvím různých protokolů současně, aniž by pro každý koncový bod vytvořili samostatné nasazení agenta.

Vytvořte pomocné rutiny v nativním exekutoru sady A2A SDK. Tato ukázka vytváří a aktualizuje úlohy A2A, převádí příchozí zprávu na běh v rozhraní Agent Framework, po dokončení datového proudu uloží aktualizovanou relaci AgentState a publikuje vrácené části jako artefakty.

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

Ukázka používá Starlette a Uvicorn, ale pomocné funkce nejsou vázané ani na jeden z nich. Použijte architekturu aplikace nebo tvůrce aplikací A2A SDK k obsluhování karty agenta A2A a JSON-RPC tras:

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

Hostování pracovního postupu pomocí adaptéru

WorkflowA2AAdapter poskytuje stejnou hranici generování a převodu karet pro pracovní postup nebo WorkflowState. Odvodí konzervativní režimy vstupu a výstupu z deklarovaných typů pracovního postupu nebo můžete zadat explicitní režimy pro reprezentaci specifickou pro aplikaci.

Samostatné pomocné funkce a2a_to_workflow_run a a2a_from_workflow_run poskytují typovanou konverzi vstupu a výstupu pracovního postupu. Adaptér je zpřístupňuje prostřednictvím asynchronních a2a_to_run a synchronních a2a_from_run metod, které je ověřují vůči jeho efektivním režimům karty. Převod vstupu přijímá jednu textovou, raw nebo datovou část A2A pro jediný vstupní typ spouštěcího exekutoru workflow a převod výstupu mapuje dokončené veřejné výstupy workflow na nativní části A2A. Volání get_card před ověřeným převodem výstupu, když adaptér musí odvodit výstupní režimy.

Aplikace nadále odpovídá za nativní vykonavatel A2A a za streamování průběhu, stav úloh, artefakty, kontrolní body a pokračování se zapojením člověka. Požadavky vyžadující lidský vstup nejsou automaticky převáděny, takže hostitelská aplikace musí implementovat vlastní strategii pokračování.

Zabezpečení relací a stav úlohy

A2AExecutor používá identifikátor kontextu A2A jako identifikátor relace Agent Framework. Ukázky založené na adaptérech a pomocných rutin kombinují tenanta A2A a ID kontextu, aby bylo možné demonstrovat mapování vybrané aplikací. V každém případě musí produkční hostitel ověřit volajícího dříve, než se požadavek dostane k obsluze požadavků A2A, odvodit tenanta a subjekt z této důvěryhodné identity a autorizovat všechna ID úloh, kontextu, pokračování a zrušení.

Important

Výchozí úložiště úloh a konfigurací push v sadě A2A SDK jsou paměťová a vlastnictví vymezují podle uživatelského jména. Pro službu s více tenanty použijte owner_resolver, které odvozuje vlastnictví ze stejného důvěryhodného tenanta a subjektu, a použijte odolná úložiště úloh a relací, pokud se repliky můžou restartovat nebo horizontálně škálovat.

Úplné ukázky serveru založeného na pomocnících a víceagentových scénářů najdete v ukázkách hostování A2A. Informace o možnostech klientů a protokolů A2A najdete ve službě agenta A2A.

Další kroky

Hostování agentů s A2A

Jděte hlouběji:

• Přehled samoobslužného hostování
• Odpovědi OpenAI
• Telegram