A2A-Agenten selbst hosten

Verwenden Sie die .NET A2A-Hostingpakete, um einen Agent Framework-Agent über ASP.NET Core verfügbar zu machen. Siehe A2A-Integration für paketsetup und ein vollständiges Serverbeispiel.

Verwenden Sie das Go-Paket provider/a2aprovider mit den offiziellen A2A Go-Serverhandlern. Ein vollständiges Serverbeispiel finden Sie unter A2A-Integration .

Agent Framework stellt zwei Python Pakete für Hosting-Agents und Workflows über das offizielle A2A SDK bereit:

Package Integrationsmodell Verwenden Sie ihn, wenn
agent-framework-a2a Eine klar strukturierte A2AExecutor, die Anfragen konvertiert, einen Agenten ausführt und A2A-Aufgabenereignisse sowie Artefakte veröffentlicht. Sie möchten das standardmäßige Agent Framework-to-A2A-Verhalten und müssen nur den A2A SDK-Server zusammenstellen.
agent-framework-hosting-a2a Modulare Bausteine für einen app-eigenen Executor. Beginnen Sie mit den grundlegenden Agent- oder Workflowkonvertern und verwenden Sie optional AgentA2AAdapter oder WorkflowA2AAdapter, die auf diesen Konvertern aufbauen, um die native Kartengenerierung und Modusvalidierung hinzuzufügen. Ihre Anwendung muss eine eigene Sitzungszuordnung, Aufgabenübergänge, Ereignisübermittlung, Artefaktgrenzen, Ausgabekonvertierung oder ein Multiprotokollhost besitzen.

Beide Pakete verwenden systemeigene A2A SDK-Typen und Serverkomponenten. Ihre Anwendung stellt den Anforderungshandler, den Aufgabenspeicher, die Routen oder den SDK-Anwendungs-Generator, die Authentifizierung und die Bereitstellung bereit. Mit agent-framework-hosting-a2a kann die Anwendung die Agentenkarte direkt erstellen oder von einem Adapter generieren lassen.

Verwenden Sie den A2A-Executor mit starken Vorgaben

Installieren Sie agent-framework-a2a, falls der integrierte Serveradapter zu Ihrem Lebenszyklus passt:

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

A2AExecutor implementiert das AgentExecutor des A2A SDK. Er liest die Benutzereingabe aus dem A2A-Anforderungskontext, erstellt eine Agent Framework-Sitzung aus der A2A-Kontext-ID, führt den Agent im Streaming- oder Nicht-Streaming-Modus aus, konvertiert unterstützte Ausgabeinhalte und veröffentlicht Aufgabenstatus- und Artefaktereignisse über das SDK TaskUpdater.

Erstellen Sie es mit dem DefaultRequestHandler A2A SDK, dem Task-Store, der Agent-Card und der Starlette-Anwendung oder einer anderen unterstützten Serverintegration. Konfigurieren Sie Streaming mit A2AExecutor(agent, stream=True), übergeben Sie stabile Ausführungsoptionen für Agent-Läufe über run_kwargs, oder leiten Sie eine Unterklasse von A2AExecutor ab und überschreiben Sie handle_events, wenn Sie eine andere Ausgabezuordnung benötigen.

A2AExecutor ist auf einen A2A-Endpunkt begrenzt und verwaltet seine A2A-Ausführung sowie die Zuordnung von Sitzungen direkt. Verwenden Sie die Hostingpakete, wenn derselbe Agent über mehrere Protokolle in einer Anwendung verfügbar sein muss.

Die vollständige Servereinrichtung finden Sie unter Verfügbarmachen eines Agent Framework-Agents über A2A.

Verwenden eines Adapters in einem App-eigenen Executor

Installieren Sie das Hostingpaket, wenn Ihre Anwendung den nativen A2A-Executor enthält, Agent Framework jedoch die öffentliche Card generieren und Konvertierungen validieren soll:

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

AgentA2AAdapter akzeptiert einen Agenten oder AgentState. Die asynchrone get_card Methode leitet den öffentlichen Namen und die Beschreibung ab, verwendet standardmäßig konservative Textmodi und kann systemeigene A2A-Fähigkeiten von Agent Framework-Instanzen SkillsProvider ableiten. Serverfunktionen und unterstützte Schnittstellen bleiben explizit, da sie den Anwendungsendpunkt und nicht die Methode des run Agents beschreiben.

Der Adapter stellt die Methoden a2a_to_run und a2a_from_run bereit, die Werte standardmäßig anhand der konfigurierten Kartenmodi validieren. Die Anwendung ist weiterhin für den A2A-Executor, den Aufgabenlebenszyklus, die Ereigniswarteschlange, die Grenzen von Artefakten, die Sitzungsrichtlinie, die Authentifizierung, die Routen und die Bereitstellung verantwortlich.

Dieser Executor verwendet einen Adapter für eingehende Konvertierung, Agentstatus und ausgehende Konvertierung:

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

Das Server-Setup erstellt den Adapter, generiert sein natives AgentCard und kombiniert den anwendungseigenen Executor mit dem A2A SDK-Request-Handler:

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

Erstellen eines app-eigenen A2A-Executors

Verwenden Sie die eigenständigen Hostinghilfsprogramme, wenn Ihre Anwendung auch die direkte Kontrolle über die Kartenerstellung benötigt:

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

Die Helfer sind frameworkneutral:

  • a2a_to_run wandelt ein A2A Message in Ausführungsargumente des Agent Framework um.
  • a2a_from_run konvertiert Agent Framework-Antworten und Streamingupdates in A2A-Werte Part .

Ihr Executor wählt Sitzungsschlüssel aus und ist verantwortlich für Aufgabenübergänge, Ereigniswarteschlangen, Artefakt-IDs, Nachrichtenabgrenzungen und die ausgehende Zustellung. a2a_from_run gibt eine flache Teilliste zurück, sodass die Anwendung diese Teile in A2A-Nachrichten oder Artefakte gruppieren und Metadaten auf Nachrichtenebene anwenden kann.

Das Hostingsetup unterstützt auch Multiprotokollanwendungen. Nutzen Sie dasselbe Agentenziel und AgentState dieselbe Infrastruktur gemeinsam für A2A-, OpenAI-Responses-, Telegram- und MCP-Routen, wobei jeder Protokollendpunkt seine eigene Umwandlungs-, Autorisierungs- und Richtlinie für Sitzungsschlüssel beibehält. Auf diese Weise können Clients einen Agent über verschiedene Protokolle gleichzeitig erreichen, ohne für jeden Endpunkt eine separate Agent-Bereitstellung zu erstellen.

Verfassen Sie die Hilfsprogramme in einem systemeigenen A2A SDK-Executor. In diesem Beispiel werden A2A-Aufgaben erstellt und aktualisiert, eine eingehende Nachricht in einen Agent-Framework-Durchlauf umgewandelt, die aktualisierte AgentState Sitzung gespeichert, nachdem der Stream beendet ist, und zurückgegebene Teile als Artefakte veröffentlicht.

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

Das Beispiel verwendet Starlette und Uvicorn, aber die Helfer sind nicht an beide gebunden. Verwenden Sie Ihr Anwendungsframework oder einen A2A SDK-Anwendungs-Generator, um die A2A-Agent-Karte und JSON-RPC Routen zu 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, "/"),
    ]
)

Hosten eines Workflows mit einem Adapter

WorkflowA2AAdapter bietet dieselbe Grenze für die Kartenerstellung und Konvertierung für einen Workflow oder WorkflowState. Er leitet konservative Eingabe- und Ausgabemodi aus den deklarierten Typen des Workflows ab, oder Sie können explizite Modi für eine anwendungsspezifische Darstellung bereitstellen.

Die eigenständigen a2a_to_workflow_run und a2a_from_workflow_run Hilfsfunktionen bieten die typisierte Konvertierung von Workflow-Ein- und -Ausgaben. Der Adapter macht sie als asynchrone a2a_to_run und synchrone a2a_from_run Methoden verfügbar, die anhand der effektiven Kartenmodi überprüft werden. Die Eingabekonvertierung akzeptiert für den einzelnen Start-Executor-Eingabetyp des Workflows einen A2A-Text-, Rohdaten- oder Datenteil, und die Ausgabekonvertierung ordnet die abgeschlossenen öffentlichen Ausgaben des Workflows nativen A2A-Teilen zu. Rufen Sie get_card vor der Konvertierung validierter Ausgaben auf, wenn der Adapter Ausgabemodi ableiten muss.

Die Anwendung bleibt verantwortlich für den nativen A2A-Executor sowie für das Streamen von Fortschritt, Aufgabenstatus, Artefakten, Prüfpunkten und die Fortsetzung mit Human-in-the-Loop. Ausstehende Anfragen, die menschliche Eingaben erfordern, werden nicht automatisch umgewandelt, sodass der Host eine eigene Richtlinie für die Fortsetzung implementieren muss.

Sichere Sitzungen und Aufgabenstatus

A2AExecutor verwendet die A2A-Kontext-ID als Agent Framework-Sitzungs-ID. Die adapterbasierten und helperbasierten Beispiele führen den A2A-Mandanten und die Kontext-ID zusammen, um eine von der Anwendung ausgewählte Zuordnung zu veranschaulichen. Bei jedem Ansatz muss ein Produktionshost den Aufrufer authentifizieren, bevor der Aufrufer den A2A-Anforderungs-Handler erreicht, den Mandanten und das Subjekt aus dieser vertrauenswürdigen Identität ableiten und alle Aufgaben-, Kontext-, Fortsetzungs- und Abbruch-IDs autorisieren.

Important

Die standardmäßigen Aufgaben- und Speicher für Push-Konfigurationen des A2A SDK werden im Arbeitsspeicher gehalten und ordnen den Besitz nach Benutzernamen zu. Verwenden Sie für einen Mandantendienst ein owner_resolver, das die Besitzzuordnung von demselben vertrauenswürdigen Mandanten und Subjekt ableitet, und verwenden Sie dauerhafte Speicher für Aufgaben und Sitzungen, wenn Replikate neu gestartet oder horizontal skaliert werden können.

Vollständige Beispiele für einen hilfsmodulbasierten Server und mehrere Agents finden Sie in den A2A-Hostingbeispielen. Informationen zu A2A-Clients und Protokollfunktionen finden Sie unter A2A-Integration.

Nächste Schritte

Informationen zur A2A-Integration

Gehen Sie tiefer:

  • Last updated on