From 46e048ac6f169d32e191da8eb7191aef70ca38e2 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 10 Jul 2026 15:17:18 +0200 Subject: [PATCH 1/3] Python: Add A2A hosting helpers Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 --- docs/specs/002-python-hosting-channels.md | 24 +++ python/AGENTS.md | 1 + python/PACKAGE_STATUS.md | 1 + python/packages/hosting-a2a/AGENTS.md | 18 ++ python/packages/hosting-a2a/LICENSE | 21 +++ python/packages/hosting-a2a/README.md | 33 ++++ .../agent_framework_hosting_a2a/__init__.py | 18 ++ .../_conversion.py | 157 ++++++++++++++++++ .../agent_framework_hosting_a2a/py.typed | 0 python/packages/hosting-a2a/pyproject.toml | 80 +++++++++ .../tests/hosting_a2a/test_conversion.py | 111 +++++++++++++ python/pyproject.toml | 1 + python/samples/04-hosting/a2a/README.md | 10 ++ python/samples/04-hosting/a2a/a2a_server.py | 75 ++++++++- .../04-hosting/a2a/agent_framework_to_a2a.py | 89 +++++++++- .../samples/04-hosting/a2a/requirements.txt | 5 +- python/uv.lock | 18 ++ 17 files changed, 651 insertions(+), 11 deletions(-) create mode 100644 python/packages/hosting-a2a/AGENTS.md create mode 100644 python/packages/hosting-a2a/LICENSE create mode 100644 python/packages/hosting-a2a/README.md create mode 100644 python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py create mode 100644 python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py create mode 100644 python/packages/hosting-a2a/agent_framework_hosting_a2a/py.typed create mode 100644 python/packages/hosting-a2a/pyproject.toml create mode 100644 python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md index 190a192fdff..533f7ef7a1a 100644 --- a/docs/specs/002-python-hosting-channels.md +++ b/docs/specs/002-python-hosting-channels.md @@ -66,6 +66,7 @@ must be aligned with the helper-first model before implementation. Old vocabular | Package | Import surface | v1 helper-first contents | |---|---|---| | `agent-framework-hosting` | `agent_framework_hosting` | `AgentState`, `WorkflowState`, `SessionStore`, and run-argument `TypedDict`s. | +| `agent-framework-hosting-a2a` | `agent_framework_hosting_a2a` | A2A `Message` to run conversion and Agent Framework output to A2A `Part` conversion. | | `agent-framework-hosting-responses` | `agent_framework_hosting_responses` | Responses helpers: request parsing, session id extraction, response id creation, response rendering, streaming rendering. | | Future protocol packages | e.g. `agent_framework_hosting_telegram` | Protocol-specific helpers such as `telegram_to_run(...)`, `telegram_from_run(...)`, `telegram_session_id(...)`, and command/media helpers when useful. | @@ -90,6 +91,7 @@ Examples: - `responses_to_run(...)`, `responses_from_run(...)`, `responses_from_streaming_run(...)`, `responses_session_id(...)`; +- `a2a_to_run(...)`, `a2a_from_run(...)`; - `telegram_to_run(...)`, `telegram_from_run(...)`, `telegram_from_streaming_run(...)`, `telegram_session_id(...)`, `telegram_command(...)`; - `activity_to_run(...)`, `activity_from_run(...)`, `activity_session_id(...)`, `activity_command(...)`; @@ -244,6 +246,28 @@ text deltas, and a completed event. The final completed payload is produced thro also preserves the model id observed on streaming updates when the finalized `AgentResponse` no longer carries raw model metadata. +## `agent-framework-hosting-a2a` + +The A2A package provides only the conversion seam between the native A2A SDK +and Agent Framework: + +- `a2a_to_run(message, *, stream=False) -> AgentRunArgs` +- `a2a_from_run(result) -> list[a2a.types.Part]` + +`a2a_to_run(...)` accepts a native A2A `Message` and converts its text, URL, +raw-byte, and structured-data parts into one Agent Framework user message. + +`a2a_from_run(...)` accepts an `AgentResponse`, `Message`, or +`AgentResponseUpdate` and converts supported text, URI, and data content into +native A2A `Part` values. This one helper is usable for both completed and +streaming runs. + +The package does not provide an A2A `AgentExecutor`, application, route, +request handler, task store, event queue, `TaskUpdater`, task-state policy, +artifact-id policy, or session-key policy. Application code composes the two +helpers with those native A2A SDK constructs and may use any server framework +supported by the SDK. + ## Security responsibilities Protocol helper packages parse and render. They do not authenticate callers, authorize access to state, or decide which diff --git a/python/AGENTS.md b/python/AGENTS.md index 0453efdfdea..720db83b018 100644 --- a/python/AGENTS.md +++ b/python/AGENTS.md @@ -94,6 +94,7 @@ python/ ### Protocols & UI - [a2a](packages/a2a/AGENTS.md) - Agent-to-Agent protocol +- [hosting-a2a](packages/hosting-a2a/AGENTS.md) - A2A hosting conversion helpers - [ag-ui](packages/ag-ui/AGENTS.md) - AG-UI protocol - [chatkit](packages/chatkit/AGENTS.md) - OpenAI ChatKit integration - [devui](packages/devui/AGENTS.md) - Developer UI for testing diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md index c0080f0cbbf..dc34c7c4c2f 100644 --- a/python/PACKAGE_STATUS.md +++ b/python/PACKAGE_STATUS.md @@ -35,6 +35,7 @@ Status is grouped into these buckets: | `agent-framework-gemini` | `python/packages/gemini` | `alpha` | | `agent-framework-github-copilot` | `python/packages/github_copilot` | `rc` | | `agent-framework-hosting` | `python/packages/hosting` | `alpha` | +| `agent-framework-hosting-a2a` | `python/packages/hosting-a2a` | `alpha` | | `agent-framework-hosting-responses` | `python/packages/hosting-responses` | `alpha` | | `agent-framework-hyperlight` | `python/packages/hyperlight` | `beta` | | `agent-framework-lab` | `python/packages/lab` | `beta` | diff --git a/python/packages/hosting-a2a/AGENTS.md b/python/packages/hosting-a2a/AGENTS.md new file mode 100644 index 00000000000..c7c0af25928 --- /dev/null +++ b/python/packages/hosting-a2a/AGENTS.md @@ -0,0 +1,18 @@ +# A2A Hosting Helpers (`agent-framework-hosting-a2a`) + +Side-effect-free conversion helpers for hosting Agent Framework agents through +the native A2A SDK. + +## Public API + +- `a2a_to_run(message, *, stream=False)` converts an A2A `Message` to + `AgentRunArgs`. +- `a2a_from_run(result)` converts an Agent Framework response, message, or + streaming update to A2A `Part` values. + +## Boundary + +This package does not provide an `AgentExecutor`, routes, a web application, +task stores, event queues, task state policy, artifact ID policy, or outbound +delivery. Applications compose the conversion helpers with native A2A SDK +constructs. diff --git a/python/packages/hosting-a2a/LICENSE b/python/packages/hosting-a2a/LICENSE new file mode 100644 index 00000000000..9e841e7a26e --- /dev/null +++ b/python/packages/hosting-a2a/LICENSE @@ -0,0 +1,21 @@ + MIT License + + Copyright (c) Microsoft Corporation. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE diff --git a/python/packages/hosting-a2a/README.md b/python/packages/hosting-a2a/README.md new file mode 100644 index 00000000000..374383fae44 --- /dev/null +++ b/python/packages/hosting-a2a/README.md @@ -0,0 +1,33 @@ +# agent-framework-hosting-a2a + +A2A conversion helpers for app-owned Agent Framework hosting. + +The package deliberately does not choose a web framework or wrap the A2A SDK +server lifecycle. It provides two conversion functions: + +- `a2a_to_run(...)` converts a native A2A `Message` into Agent Framework run + arguments. +- `a2a_from_run(...)` converts an `AgentResponse`, `Message`, or streaming + `AgentResponseUpdate` into native A2A `Part` values. + +Application code keeps ownership of the A2A SDK's `AgentExecutor`, +`RequestContext`, `TaskUpdater`, event queue, task store, routes, task state, +artifact IDs, authentication, and deployment. + +```python +run = a2a_to_run(context.message) +session_id = f"a2a:{context.tenant}:{context.context_id}" +session = await state.get_or_create_session(session_id) +result = await agent.run( + run["messages"], + session=session, + options=run["options"], +) +await state.set_session(session_id, session) +parts = a2a_from_run(result) + +# Native A2A SDK application code publishes `parts` with TaskUpdater. +``` + +The surrounding A2A application may use Starlette, FastAPI, another ASGI +framework, or the SDK's own application builders. These helpers do not care. diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py new file mode 100644 index 00000000000..70e446d97da --- /dev/null +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py @@ -0,0 +1,18 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""A2A conversion helpers for app-owned Agent Framework hosting.""" + +import importlib.metadata + +from ._conversion import a2a_from_run, a2a_to_run + +try: + __version__ = importlib.metadata.version(__name__) +except importlib.metadata.PackageNotFoundError: + __version__ = "0.0.0" + +__all__ = [ + "__version__", + "a2a_from_run", + "a2a_to_run", +] diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py new file mode 100644 index 00000000000..e8805db8e17 --- /dev/null +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py @@ -0,0 +1,157 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Conversion between native A2A values and Agent Framework run values.""" + +from __future__ import annotations + +import base64 +import json +import logging +from collections.abc import Sequence +from typing import Any, cast + +from a2a.types import Message as A2AMessage +from a2a.types import Part +from agent_framework import AgentResponse, AgentResponseUpdate, ChatOptions, Content, Message +from agent_framework_hosting import AgentRunArgs +from google.protobuf.json_format import MessageToDict + +logger = logging.getLogger("agent_framework.hosting.a2a") + + +def a2a_to_run(message: A2AMessage, *, stream: bool = False) -> AgentRunArgs: + """Convert an A2A message into Agent Framework run arguments. + + A2A text, URL, raw-byte, and structured-data parts become Agent Framework + content. The helper does not create sessions, inspect task stores, or + interact with an A2A request handler. + + Args: + message: Native A2A message to convert. + + Keyword Args: + stream: Whether the caller intends to run the agent in streaming mode. + + Returns: + Arguments corresponding to ``Agent.run(...)``. + + Raises: + ValueError: If the message has no supported content parts. + """ + contents: list[Content] = [] + for part in message.parts: + metadata = MessageToDict(part.metadata) if part.metadata else None + match part.WhichOneof("content"): + case "text": + contents.append( + Content.from_text( + text=part.text, + additional_properties=metadata, + raw_representation=part, + ) + ) + case "url": + contents.append( + Content.from_uri( + uri=part.url, + media_type=part.media_type or None, + additional_properties=metadata, + raw_representation=part, + ) + ) + case "raw": + contents.append( + Content.from_data( + data=part.raw, + media_type=part.media_type or "application/octet-stream", + additional_properties=metadata, + raw_representation=part, + ) + ) + case "data": + contents.append( + Content.from_text( + text=json.dumps(MessageToDict(part.data), separators=(",", ":"), sort_keys=True), + additional_properties=metadata, + raw_representation=part, + ) + ) + case unsupported: + logger.warning("A2A message part type %s is not supported and was omitted.", unsupported) + + if not contents: + raise ValueError("A2A message has no supported text, URL, raw, or data parts to convert to a run.") + + return AgentRunArgs( + messages=[ + Message( + "user", + contents, + message_id=message.message_id or None, + additional_properties={"a2a_metadata": MessageToDict(message.metadata)} if message.metadata else None, + raw_representation=message, + ) + ], + options=cast("ChatOptions[Any]", {}), + stream=stream, + ) + + +def a2a_from_run(result: AgentResponse[Any] | Message | AgentResponseUpdate) -> list[Part]: + """Convert Agent Framework output into native A2A parts. + + ``AgentResponse`` values are flattened in message order. User-role + messages are omitted. Text, external URI, and inline data content become + the corresponding native A2A part types. The caller remains responsible + for creating A2A messages or artifacts and publishing them through + ``TaskUpdater`` or an event queue. + + Args: + result: A completed response, response message, or streaming update. + + Returns: + Native A2A parts ready for an A2A SDK message or artifact. + + Raises: + ValueError: If Agent Framework data content contains an invalid data URI. + """ + items: Sequence[Message | AgentResponseUpdate] = result.messages if isinstance(result, AgentResponse) else [result] + + parts: list[Part] = [] + for item in items: + if item.role == "user": + continue + for content in item.contents: + metadata = content.additional_properties or {} + match content.type: + case "text" if content.text is not None: + parts.append(Part(text=content.text, metadata=metadata)) + case "uri" if content.uri is not None: + parts.append( + Part( + url=content.uri, + media_type=content.media_type or "", + metadata=metadata, + ) + ) + case "data" if content.uri is not None: + prefix, separator, encoded = content.uri.partition(",") + if not separator or not prefix.startswith("data:") or ";base64" not in prefix: + raise ValueError("Agent Framework data content must contain a base64 data URI.") + try: + raw = base64.b64decode(encoded, validate=True) + except ValueError as exc: + raise ValueError("Agent Framework data content contains invalid base64 data.") from exc + parts.append( + Part( + raw=raw, + media_type=content.media_type or "", + metadata=metadata, + ) + ) + case _: + logger.warning( + "Agent Framework content type %s is not supported by A2A and was omitted.", + content.type, + ) + return parts diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/py.typed b/python/packages/hosting-a2a/agent_framework_hosting_a2a/py.typed new file mode 100644 index 00000000000..e69de29bb2d diff --git a/python/packages/hosting-a2a/pyproject.toml b/python/packages/hosting-a2a/pyproject.toml new file mode 100644 index 00000000000..18cc956cab6 --- /dev/null +++ b/python/packages/hosting-a2a/pyproject.toml @@ -0,0 +1,80 @@ +[project] +name = "agent-framework-hosting-a2a" +description = "A2A conversion helpers for app-owned Agent Framework hosting." +authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}] +readme = "README.md" +requires-python = ">=3.10" +version = "1.0.0a260710" +license-files = ["LICENSE"] +urls.homepage = "https://aka.ms/agent-framework" +urls.source = "https://github.com/microsoft/agent-framework/tree/main/python" +urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true" +urls.issues = "https://github.com/microsoft/agent-framework/issues" +classifiers = [ + "License :: OSI Approved :: MIT License", + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Typing :: Typed", +] +dependencies = [ + "a2a-sdk>=1.0.0,<2", + "agent-framework-core>=1.11.0,<2", + "agent-framework-hosting==1.0.0a260709", +] + +[tool.uv] +prerelease = "if-necessary-or-explicit" +environments = [ + "sys_platform == 'darwin'", + "sys_platform == 'linux'", + "sys_platform == 'win32'" +] + +[tool.uv-dynamic-versioning] +fallback-version = "0.0.0" + +[tool.pytest.ini_options] +testpaths = 'tests' +addopts = "-ra -q -r fEX" +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +filterwarnings = [] +timeout = 120 +markers = [ + "integration: marks tests as integration tests that require external services", +] + +[tool.ruff] +extend = "../../pyproject.toml" + +[tool.coverage.run] +omit = [ + "**/__init__.py" +] + +[tool.pyright] +extends = "../../pyproject.toml" +include = ["agent_framework_hosting_a2a"] +exclude = ['tests'] + +[tool.bandit] +targets = ["agent_framework_hosting_a2a"] +exclude_dirs = ["tests"] + +[tool.poe] +executor.type = "uv" +include = "../../shared_tasks.toml" + +[tool.poe.tasks.test] +help = "Run the default unit test suite for this package." +cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_a2a --cov-report=term-missing:skip-covered tests' + +[build-system] +requires = ["flit-core >= 3.11,<4.0"] +build-backend = "flit_core.buildapi" diff --git a/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py new file mode 100644 index 00000000000..941f1961987 --- /dev/null +++ b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py @@ -0,0 +1,111 @@ +# Copyright (c) Microsoft. All rights reserved. + +from a2a.types import Message as A2AMessage +from a2a.types import Part, Role +from agent_framework import AgentResponse, AgentResponseUpdate, Content, Message +from google.protobuf.json_format import MessageToDict +from pytest import raises + +from agent_framework_hosting_a2a import a2a_from_run, a2a_to_run + + +def test_a2a_to_run_converts_supported_parts() -> None: + data_part = Part() + data_part.data.string_value = "structured" + message = A2AMessage( + message_id="message-1", + role=Role.ROLE_USER, + parts=[ + Part(text="hello", metadata={"source": "text"}), + Part(url="https://example.com/image.png", media_type="image/png"), + Part(raw=b"audio", media_type="audio/wav"), + data_part, + ], + metadata={"tenant_value": "kept"}, + ) + + run = a2a_to_run(message, stream=True) + + messages = run["messages"] + assert isinstance(messages, list) + converted = messages[0] + assert isinstance(converted, Message) + assert run["stream"] is True + assert run["options"] == {} + assert converted.message_id == "message-1" + assert converted.additional_properties == {"a2a_metadata": {"tenant_value": "kept"}} + assert [content.type for content in converted.contents] == ["text", "uri", "data", "text"] + assert converted.contents[0].additional_properties == {"source": "text"} + assert converted.contents[1].uri == "https://example.com/image.png" + assert converted.contents[2].uri == "data:audio/wav;base64,YXVkaW8=" + assert converted.contents[3].text == '"structured"' + + +def test_a2a_to_run_rejects_empty_message() -> None: + with raises(ValueError, match="no supported"): + a2a_to_run(A2AMessage(message_id="message-1", role=Role.ROLE_USER)) + + +def test_a2a_from_run_converts_final_response() -> None: + response = AgentResponse( + messages=[ + Message("user", ["omit me"]), + Message( + "assistant", + [ + Content.from_text("hello", additional_properties={"source": "agent"}), + Content.from_uri("https://example.com/image.png", media_type="image/png"), + Content.from_data(b"audio", "audio/wav"), + ], + ), + ] + ) + + parts = a2a_from_run(response) + + assert len(parts) == 3 + assert parts[0].text == "hello" + assert MessageToDict(parts[0].metadata) == {"source": "agent"} + assert parts[1].url == "https://example.com/image.png" + assert parts[1].media_type == "image/png" + assert parts[2].raw == b"audio" + assert parts[2].media_type == "audio/wav" + + +def test_a2a_from_run_converts_streaming_update() -> None: + update = AgentResponseUpdate( + role="assistant", + contents=[Content.from_text("chunk")], + message_id="message-1", + ) + + parts = a2a_from_run(update) + + assert len(parts) == 1 + assert parts[0].text == "chunk" + + +def test_a2a_from_run_preserves_empty_text_part() -> None: + parts = a2a_from_run(Message("assistant", [Content.from_text("")])) + + assert len(parts) == 1 + assert parts[0].WhichOneof("content") == "text" + assert parts[0].text == "" + + +def test_a2a_from_run_rejects_invalid_data_uri() -> None: + content = Content("data", uri="not-a-data-uri", media_type="application/octet-stream") + + with raises(ValueError, match="base64 data URI"): + a2a_from_run(Message("assistant", [content])) + + +def test_a2a_from_run_rejects_invalid_base64_data() -> None: + content = Content( + "data", + uri="data:application/octet-stream;base64,not valid base64", + media_type="application/octet-stream", + ) + + with raises(ValueError, match="invalid base64"): + a2a_from_run(Message("assistant", [content])) diff --git a/python/pyproject.toml b/python/pyproject.toml index b7575995643..d07654a83a9 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -92,6 +92,7 @@ agent-framework-foundry-local = { workspace = true } agent-framework-gemini = { workspace = true } agent-framework-github-copilot = { workspace = true } agent-framework-hosting = { workspace = true } +agent-framework-hosting-a2a = { workspace = true } agent-framework-hosting-responses = { workspace = true } agent-framework-hyperlight = { workspace = true } agent-framework-lab = { workspace = true } diff --git a/python/samples/04-hosting/a2a/README.md b/python/samples/04-hosting/a2a/README.md index c187e55b61d..7e079b84ea1 100644 --- a/python/samples/04-hosting/a2a/README.md +++ b/python/samples/04-hosting/a2a/README.md @@ -2,6 +2,11 @@ This sample demonstrates how to **host** Agent Framework agents as A2A-compliant servers using the [A2A (Agent2Agent) protocol](https://a2a-protocol.org/latest/). +`agent-framework-hosting-a2a` only converts between native A2A values and +Agent Framework run values. The sample deliberately keeps the A2A SDK's +`AgentExecutor`, task lifecycle, event queue, task store, and Starlette routes +in application code. The helper package does not choose a web framework. + > **Looking for client samples?** See [`samples/02-agents/a2a/`](../../02-agents/a2a/) for consuming remote A2A agents. ## Server Samples @@ -92,3 +97,8 @@ def resolve_tenant_user_scope(context): return f"{context.tenant}:{context.user.user_name}" task_store = InMemoryTaskStore(owner_resolver=resolve_tenant_user_scope) ``` + +The sample also includes the authenticated A2A tenant and protocol context id +in its Agent Framework session key. Production applications must derive that +tenant from trusted authentication context and use a durable session store when +running multiple replicas or transient workers. diff --git a/python/samples/04-hosting/a2a/a2a_server.py b/python/samples/04-hosting/a2a/a2a_server.py index a03eea2a288..eb5e1b4428f 100644 --- a/python/samples/04-hosting/a2a/a2a_server.py +++ b/python/samples/04-hosting/a2a/a2a_server.py @@ -1,22 +1,35 @@ # Copyright (c) Microsoft. All rights reserved. import argparse +import logging import os import sys +import uuid +from asyncio import CancelledError +from typing import Generic, TypeVar import uvicorn +from a2a.helpers import new_task_from_user_message +from a2a.server.agent_execution import AgentExecutor, RequestContext +from a2a.server.events import EventQueue from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes -from a2a.server.tasks import InMemoryTaskStore +from a2a.server.tasks import InMemoryTaskStore, TaskUpdater +from a2a.types import Part, TaskState from agent_definitions import AGENT_CARD_FACTORIES, AGENT_FACTORIES # pyrefly: ignore[missing-import] -from agent_framework.a2a import A2AExecutor +from agent_framework import SupportsAgentRun from agent_framework.foundry import FoundryChatClient +from agent_framework_hosting import AgentState +from agent_framework_hosting_a2a import a2a_from_run, a2a_to_run from azure.identity import AzureCliCredential from dotenv import load_dotenv from starlette.applications import Starlette # Load environment variables from .env file load_dotenv() +logger = logging.getLogger(__name__) + +AgentT = TypeVar("AgentT", bound=SupportsAgentRun) """ A2A Server Sample — Host an Agent Framework agent as an A2A endpoint @@ -41,6 +54,61 @@ """ +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() + session_id = f"a2a:{context.tenant}:{context.context_id}" + session = await self.state.get_or_create_session(session_id) + stream = agent.run(run["messages"], session=session, stream=True) + 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) + await stream.get_final_response() + await self.state.set_session(session_id, session) + await updater.complete() + except CancelledError: + await updater.update_status(state=TaskState.TASK_STATE_CANCELED) + except Exception as exc: + logger.exception("A2A agent execution failed.") + await updater.update_status( + state=TaskState.TASK_STATE_FAILED, + message=updater.new_agent_message([Part(text=str(exc))]), + ) + + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="A2A Agent Server") parser.add_argument( @@ -88,11 +156,12 @@ def main() -> None: # 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 = A2AExecutor(agent, stream=True) + executor = AppAgentExecutor(state) task_store = InMemoryTaskStore() request_handler = DefaultRequestHandler( agent_executor=executor, diff --git a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py index c29f18ed6de..bb8bea6939b 100644 --- a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py +++ b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py @@ -1,23 +1,94 @@ # Copyright (c) Microsoft. All rights reserved. +import logging +import uuid +from asyncio import CancelledError +from typing import Generic, TypeVar + import uvicorn +from a2a.helpers import new_task_from_user_message +from a2a.server.agent_execution import AgentExecutor, RequestContext +from a2a.server.events import EventQueue from a2a.server.request_handlers import DefaultRequestHandler from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes -from a2a.server.tasks import InMemoryTaskStore +from a2a.server.tasks import InMemoryTaskStore, TaskUpdater from a2a.types import ( AgentCapabilities, AgentCard, AgentInterface, AgentSkill, + Part, + TaskState, ) -from agent_framework import Agent -from agent_framework.a2a import A2AExecutor +from agent_framework import Agent, SupportsAgentRun from agent_framework.openai import OpenAIChatClient +from agent_framework_hosting import AgentState +from agent_framework_hosting_a2a import a2a_from_run, a2a_to_run from dotenv import load_dotenv from starlette.applications import Starlette load_dotenv() +logger = logging.getLogger(__name__) + +AgentT = TypeVar("AgentT", bound=SupportsAgentRun) + + +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() + session_id = f"a2a:{context.tenant}:{context.context_id}" + session = await self.state.get_or_create_session(session_id) + stream = agent.run(run["messages"], session=session, stream=True) + 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) + await stream.get_final_response() + await self.state.set_session(session_id, session) + await updater.complete() + except CancelledError: + await updater.update_status(state=TaskState.TASK_STATE_CANCELED) + except Exception as exc: + logger.exception("A2A agent execution failed.") + await updater.update_status( + state=TaskState.TASK_STATE_FAILED, + message=updater.new_agent_message([Part(text=str(exc))]), + ) + + if __name__ == "__main__": # --8<-- [start:AgentSkill] flight_skill = AgentSkill( @@ -40,7 +111,9 @@ # This will be the public-facing agent card public_agent_card = AgentCard( name="Europe Travel Agent", - description="A helpful Europe Travel Agent that can help users search and book flights and hotels across Europe.", + description=( + "A helpful Europe Travel Agent that can help users search and book flights and hotels across Europe." + ), version="1.0.0", default_input_modes=["text"], default_output_modes=["text"], @@ -53,11 +126,15 @@ agent = Agent( client=OpenAIChatClient(), name="Europe Travel Agent", - instructions="You are a helpful Europe Travel Agent. You can help users search and book flights and hotels across Europe.", + instructions=( + "You are a helpful Europe Travel Agent. " + "You can help users search and book flights and hotels across Europe." + ), ) + state = AgentState(agent) request_handler = DefaultRequestHandler( - agent_executor=A2AExecutor(agent), + agent_executor=AppAgentExecutor(state), task_store=InMemoryTaskStore(), agent_card=public_agent_card, ) diff --git a/python/samples/04-hosting/a2a/requirements.txt b/python/samples/04-hosting/a2a/requirements.txt index fe6e43a1383..6bc698a2482 100644 --- a/python/samples/04-hosting/a2a/requirements.txt +++ b/python/samples/04-hosting/a2a/requirements.txt @@ -1,6 +1,6 @@ # Agent Framework packages # To use the deployed version, uncomment the lines below and comment out the local installation lines -# agent-framework-a2a +# agent-framework-hosting-a2a # agent-framework-foundry # Local installation (for development and testing) @@ -8,7 +8,8 @@ # Without explicit entries, pip would fetch transitive dependencies from PyPI instead of local source. -e ../../../packages/core # Core framework - base dependency for all packages -e ../../../packages/foundry # Foundry support - dependency for FoundryChatClient in a2a_server.py --e ../../../packages/a2a # A2A integration - provides A2AAgent and a2a-sdk +-e ../../../packages/hosting # Shared AgentState and SessionStore helpers +-e ../../../packages/hosting-a2a # A2A-to-Agent Framework conversion helpers # Azure authentication azure-identity diff --git a/python/uv.lock b/python/uv.lock index b15b290e312..c8efcfe20c7 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -51,6 +51,7 @@ members = [ "agent-framework-gemini", "agent-framework-github-copilot", "agent-framework-hosting", + "agent-framework-hosting-a2a", "agent-framework-hosting-responses", "agent-framework-hyperlight", "agent-framework-lab", @@ -654,6 +655,23 @@ dependencies = [ [package.metadata] requires-dist = [{ name = "agent-framework-core", editable = "packages/core" }] +[[package]] +name = "agent-framework-hosting-a2a" +version = "1.0.0a260710" +source = { editable = "packages/hosting-a2a" } +dependencies = [ + { name = "a2a-sdk", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, +] + +[package.metadata] +requires-dist = [ + { name = "a2a-sdk", specifier = ">=1.0.0,<2" }, + { name = "agent-framework-core", editable = "packages/core" }, + { name = "agent-framework-hosting", editable = "packages/hosting" }, +] + [[package]] name = "agent-framework-hosting-responses" version = "1.0.0a260709" From cec551bd1abb4f4d26a22e61530b6660a5dc1bf2 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 10 Jul 2026 15:42:18 +0200 Subject: [PATCH 2/3] Python: Preserve final A2A streaming output Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 --- python/samples/04-hosting/a2a/a2a_server.py | 18 ++++++++++++++++-- .../04-hosting/a2a/agent_framework_to_a2a.py | 18 ++++++++++++++++-- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/python/samples/04-hosting/a2a/a2a_server.py b/python/samples/04-hosting/a2a/a2a_server.py index eb5e1b4428f..9f93e9752f8 100644 --- a/python/samples/04-hosting/a2a/a2a_server.py +++ b/python/samples/04-hosting/a2a/a2a_server.py @@ -83,7 +83,14 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non agent = await self.state.get_target() session_id = f"a2a:{context.tenant}:{context.context_id}" session = await self.state.get_or_create_session(session_id) - stream = agent.run(run["messages"], session=session, stream=True) + 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: @@ -96,7 +103,14 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non append=True if artifact_id in streamed_artifact_ids else None, ) streamed_artifact_ids.add(artifact_id) - await stream.get_final_response() + 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: diff --git a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py index bb8bea6939b..7f640fe9d37 100644 --- a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py +++ b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py @@ -63,7 +63,14 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non agent = await self.state.get_target() session_id = f"a2a:{context.tenant}:{context.context_id}" session = await self.state.get_or_create_session(session_id) - stream = agent.run(run["messages"], session=session, stream=True) + 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: @@ -76,7 +83,14 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non append=True if artifact_id in streamed_artifact_ids else None, ) streamed_artifact_ids.add(artifact_id) - await stream.get_final_response() + 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: From eab0bfca91ade960f935ffe87e10c36be3cd3f70 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Fri, 10 Jul 2026 16:23:53 +0200 Subject: [PATCH 3/3] Python: Clarify A2A conversion boundary Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5d6987cd-1b67-4ba1-8b54-3c50da6e7607 --- python/packages/hosting-a2a/AGENTS.md | 4 +++ python/packages/hosting-a2a/README.md | 5 +++ .../_conversion.py | 7 ++-- .../tests/hosting_a2a/test_conversion.py | 35 +++++++++++++++++++ python/samples/04-hosting/a2a/a2a_server.py | 4 +-- .../04-hosting/a2a/agent_framework_to_a2a.py | 4 +-- 6 files changed, 52 insertions(+), 7 deletions(-) diff --git a/python/packages/hosting-a2a/AGENTS.md b/python/packages/hosting-a2a/AGENTS.md index c7c0af25928..2d00985854d 100644 --- a/python/packages/hosting-a2a/AGENTS.md +++ b/python/packages/hosting-a2a/AGENTS.md @@ -16,3 +16,7 @@ This package does not provide an `AgentExecutor`, routes, a web application, task stores, event queues, task state policy, artifact ID policy, or outbound delivery. Applications compose the conversion helpers with native A2A SDK constructs. + +`a2a_from_run(...)` intentionally returns a flat part list. It preserves +content-level metadata, while applications own A2A message and artifact +boundaries plus message-level metadata. diff --git a/python/packages/hosting-a2a/README.md b/python/packages/hosting-a2a/README.md index 374383fae44..0a5082a64a1 100644 --- a/python/packages/hosting-a2a/README.md +++ b/python/packages/hosting-a2a/README.md @@ -14,6 +14,11 @@ Application code keeps ownership of the A2A SDK's `AgentExecutor`, `RequestContext`, `TaskUpdater`, event queue, task store, routes, task state, artifact IDs, authentication, and deployment. +`a2a_from_run(...)` preserves content-level metadata on each returned part and +flattens completed responses in message order. The application decides how to +group those parts into A2A messages or artifacts and owns their message-level +metadata and boundaries. + ```python run = a2a_to_run(context.message) session_id = f"a2a:{context.tenant}:{context.context_id}" diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py index e8805db8e17..e9f81663bce 100644 --- a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py @@ -102,9 +102,10 @@ def a2a_from_run(result: AgentResponse[Any] | Message | AgentResponseUpdate) -> ``AgentResponse`` values are flattened in message order. User-role messages are omitted. Text, external URI, and inline data content become - the corresponding native A2A part types. The caller remains responsible - for creating A2A messages or artifacts and publishing them through - ``TaskUpdater`` or an event queue. + the corresponding native A2A part types. Content-level metadata is + preserved on each part. The caller remains responsible for grouping parts + into A2A messages or artifacts, including message boundaries and metadata, + and publishing them through ``TaskUpdater`` or an event queue. Args: result: A completed response, response message, or streaming update. diff --git a/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py index 941f1961987..47dd94a3a49 100644 --- a/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py +++ b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py @@ -46,6 +46,22 @@ def test_a2a_to_run_rejects_empty_message() -> None: a2a_to_run(A2AMessage(message_id="message-1", role=Role.ROLE_USER)) +def test_a2a_to_run_omits_unsupported_parts() -> None: + run = a2a_to_run( + A2AMessage( + message_id="message-1", + role=Role.ROLE_USER, + parts=[Part(), Part(text="hello")], + ) + ) + + messages = run["messages"] + assert isinstance(messages, list) + converted = messages[0] + assert isinstance(converted, Message) + assert converted.text == "hello" + + def test_a2a_from_run_converts_final_response() -> None: response = AgentResponse( messages=[ @@ -93,6 +109,25 @@ def test_a2a_from_run_preserves_empty_text_part() -> None: assert parts[0].text == "" +def test_a2a_from_run_omits_unsupported_content() -> None: + parts = a2a_from_run( + Message( + "assistant", + [ + Content(type="function_call", call_id="call-1", name="get_weather", arguments="{}"), + Content.from_text("hello"), + ], + ) + ) + + assert len(parts) == 1 + assert parts[0].text == "hello" + + +def test_a2a_from_run_omits_user_messages() -> None: + assert a2a_from_run(AgentResponse(messages=[Message("user", ["omit me"])])) == [] + + def test_a2a_from_run_rejects_invalid_data_uri() -> None: content = Content("data", uri="not-a-data-uri", media_type="application/octet-stream") diff --git a/python/samples/04-hosting/a2a/a2a_server.py b/python/samples/04-hosting/a2a/a2a_server.py index 9f93e9752f8..c5409d5896f 100644 --- a/python/samples/04-hosting/a2a/a2a_server.py +++ b/python/samples/04-hosting/a2a/a2a_server.py @@ -115,11 +115,11 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non await updater.complete() except CancelledError: await updater.update_status(state=TaskState.TASK_STATE_CANCELED) - except Exception as exc: + except Exception: logger.exception("A2A agent execution failed.") await updater.update_status( state=TaskState.TASK_STATE_FAILED, - message=updater.new_agent_message([Part(text=str(exc))]), + message=updater.new_agent_message([Part(text="Agent execution failed.")]), ) diff --git a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py index 7f640fe9d37..1eda0aa80cb 100644 --- a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py +++ b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py @@ -95,11 +95,11 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non await updater.complete() except CancelledError: await updater.update_status(state=TaskState.TASK_STATE_CANCELED) - except Exception as exc: + except Exception: logger.exception("A2A agent execution failed.") await updater.update_status( state=TaskState.TASK_STATE_FAILED, - message=updater.new_agent_message([Part(text=str(exc))]), + message=updater.new_agent_message([Part(text="Agent execution failed.")]), )