Edit

Self-host A2A agents

Use the .NET A2A hosting packages to expose an Agent Framework agent through ASP.NET Core. See A2A integration for package setup and a complete server example.

Use the Go provider/a2aprovider package with the official A2A Go server handlers. See A2A integration for a complete server example.

Agent Framework provides two Python packages for hosting agents and workflows through the official A2A SDK:

Package Integration model Use it when
agent-framework-a2a An opinionated A2AExecutor that converts requests, runs an agent, and publishes A2A task events and artifacts. You want the standard Agent Framework-to-A2A behavior and only need to assemble the A2A SDK server.
agent-framework-hosting-a2a Incremental building blocks for an app-owned executor. Start with the foundational agent or workflow converters, and optionally use AgentA2AAdapter or WorkflowA2AAdapter, which build on those converters to add native card generation and mode validation. Your application needs to own session mapping, task transitions, event delivery, artifact boundaries, output conversion, or a multi-protocol host.

Both packages use native A2A SDK types and server components. Your application supplies the request handler, task store, routes or SDK application builder, authentication, and deployment. With agent-framework-hosting-a2a, the application can construct the agent card directly or let an adapter generate it.

Use the opinionated A2A executor

Install agent-framework-a2a when the built-in server adapter matches your lifecycle:

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

A2AExecutor implements the A2A SDK's AgentExecutor. It reads the user input from the A2A request context, creates an Agent Framework session from the A2A context ID, runs the agent in streaming or non-streaming mode, converts supported output content, and publishes task status and artifact events through the SDK's TaskUpdater.

Compose it with the A2A SDK's DefaultRequestHandler, task store, agent card, and Starlette application or another supported server integration. Configure streaming with A2AExecutor(agent, stream=True), pass stable agent run options through run_kwargs, or subclass A2AExecutor and override handle_events when you need a different output mapping.

A2AExecutor is scoped to an A2A endpoint and manages its A2A execution and session mapping directly. Use the hosting packages when the same agent must be available through several protocols in one application.

For the complete server setup, see Expose an Agent Framework agent over A2A.

Use an adapter in an app-owned executor

Install the hosting package when your application owns the native A2A executor but wants Agent Framework to generate the public card and validate conversions:

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

AgentA2AAdapter accepts an agent or AgentState. Its asynchronous get_card method derives the public name and description, uses conservative text modes by default, and can infer native A2A skills from Agent Framework SkillsProvider instances. Server capabilities and supported interfaces remain explicit because they describe the application endpoint rather than the agent's run method.

The adapter exposes a2a_to_run and a2a_from_run methods that validate values against the configured card modes by default. The application still owns the A2A executor, task lifecycle, event queue, artifact boundaries, session policy, authentication, routes, and deployment.

This executor uses one adapter for inbound conversion, agent state, and outbound conversion:

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

The server setup creates the adapter, generates its native AgentCard, and composes the app-owned executor with the 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,
    )

Build an app-owned A2A executor

Use the standalone hosting helpers when your application also needs direct control over card creation:

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

The helpers are framework-neutral:

  • a2a_to_run converts an A2A Message to Agent Framework run arguments.
  • a2a_from_run converts Agent Framework responses and streaming updates to A2A Part values.

Your executor selects session keys and owns task transitions, event queues, artifact IDs, message boundaries, and outbound delivery. a2a_from_run returns a flat part list so the application can group those parts into A2A messages or artifacts and apply message-level metadata.

The hosting setup also supports multi-protocol applications. Share the same agent target and AgentState infrastructure across A2A, OpenAI Responses, Telegram, and MCP routes, while each protocol endpoint keeps its own conversion, authorization, and session-key policy. This lets clients reach one agent through different protocols at the same time without creating a separate agent deployment for each endpoint.

Compose the helpers in a native A2A SDK executor. This sample creates and updates A2A tasks, converts the inbound message into an Agent Framework run, persists the updated AgentState session after the stream finishes, and publishes returned parts as artifacts.

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

The sample uses Starlette and Uvicorn, but the helpers are not tied to either. Use your application framework or an A2A SDK application builder to serve the A2A agent card and JSON-RPC routes:

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

Host a workflow with an adapter

WorkflowA2AAdapter provides the same card-generation and conversion boundary for a workflow or WorkflowState. It infers conservative input and output modes from the workflow's declared types, or you can supply explicit modes for an application-specific representation.

The standalone a2a_to_workflow_run and a2a_from_workflow_run helpers provide typed workflow input and output conversion. The adapter exposes them as asynchronous a2a_to_run and synchronous a2a_from_run methods that validate against its effective card modes. Input conversion accepts one A2A text, raw, or data part for the workflow's single start-executor input type, and output conversion maps completed public workflow outputs to native A2A parts. Call get_card before validated output conversion when the adapter must infer output modes.

The application remains responsible for the native A2A executor and for streaming progress, task status, artifacts, checkpoints, and human-in-the-loop continuation. Pending human-input requests aren't converted automatically, so the host must implement its own continuation policy.

Secure sessions and task state

A2AExecutor uses the A2A context ID as the Agent Framework session ID. The adapter-based and helper-based samples combine the A2A tenant and context ID to demonstrate an application-selected mapping. In every approach, a production host must authenticate the caller before it reaches the A2A request handler, derive the tenant and subject from that trusted identity, and authorize all task, context, continuation, and cancellation IDs.

Important

The A2A SDK's default task and push-configuration stores are in-memory and scope ownership by user name. For a multi-tenant service, use an owner_resolver that derives ownership from the same trusted tenant and subject, and use durable task and session stores when replicas can restart or scale out.

For a complete helper-based server and multi-agent examples, see the A2A hosting samples. For A2A clients and protocol capabilities, see A2A integration.

Next steps

Learn about A2A integration

Go deeper:

  • Last updated on