diff --git a/python/packages/hosting-a2a/AGENTS.md b/python/packages/hosting-a2a/AGENTS.md index 2d00985854..77a3465216 100644 --- a/python/packages/hosting-a2a/AGENTS.md +++ b/python/packages/hosting-a2a/AGENTS.md @@ -1,14 +1,29 @@ # A2A Hosting Helpers (`agent-framework-hosting-a2a`) -Side-effect-free conversion helpers for hosting Agent Framework agents through -the native A2A SDK. +Conversion and native card-generation helpers for hosting Agent Framework +agents and workflows through an application-owned A2A server. ## Public API - `a2a_to_run(message, *, stream=False)` converts an A2A `Message` to - `AgentRunArgs`. + `AgentRunArgs`; pass `input_modes` for optional advertised-mode validation. - `a2a_from_run(result)` converts an Agent Framework response, message, or - streaming update to A2A `Part` values. + streaming update to A2A `Part` values; pass `output_modes` for optional + advertised-mode validation. +- `await AgentA2AAdapter(target, ...).get_card()` creates a native `AgentCard` + from an agent or `AgentState`, target metadata, and explicit A2A discovery + policy. The adapter also re-exposes `a2a_to_run(...)` and + `a2a_from_run(...)`, validating against configured card modes by default. +- `a2a_to_workflow_run(message, workflow)` validates one text, raw, or data + part against the workflow's single start-executor input type. +- `a2a_from_workflow_run(result)` converts completed public workflow outputs + to native A2A parts and rejects pending external-input requests. +- `await WorkflowA2AAdapter(target, ...).get_card()` creates a native `AgentCard` + from a workflow or `WorkflowState` and infers defensible modes from declared + workflow types. The adapter also re-exposes the workflow conversion helpers + as `await a2a_to_run(...)` and `a2a_from_run(...)`, validating against + effective card modes by default. Inferred workflow output modes are resolved + by `get_card()` before validated output conversion. ## Boundary @@ -20,3 +35,30 @@ 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. + +Card builders return native A2A protobuf values; do not create a parallel card +model or subclass. `AgentA2AAdapter` infers built-in Agent Framework skills from +the resolved agent's `SkillsProvider` instances by default; `infer_skills=False` +disables this. The `skills` argument accepts both Agent Framework `Skill` +values and native A2A `AgentSkill` values. Do not infer A2A skills from function +tools. Capabilities such as streaming and push notifications remain explicit +because they describe the application server. Skill discovery runs with a +`SkillsSourceContext` containing the resolved agent and no session. + +`supported_interfaces` contains one native `AgentInterface` per public +protocol endpoint. The URL is where the application mounted the corresponding +A2A routes, and the binding must match the protocol actually served there +(commonly `JSONRPC`, `HTTP+JSON`, or `GRPC`). + +A2A mode strings are extensible, but automatic parsing is intentionally +limited to `text`, `application/json`, `application/octet-stream`, and +pass-through concrete media types (with wildcard validation such as +`image/*`). JSON-only output parses Agent Framework JSON text into native A2A +data parts; structured workflow output becomes a data part, or JSON text when +only `text` is advertised. Custom modes are accepted when a native part already +carries that media type, and otherwise raise instead of guessing a serializer. + +Workflow mode inference is conservative: string schemas map to `text`, binary +strings to `application/octet-stream`, and JSON-compatible schemas to +`application/json`. Unknown application-specific representations require +explicit card modes and custom application conversion. diff --git a/python/packages/hosting-a2a/README.md b/python/packages/hosting-a2a/README.md index 0a5082a64a..432094d4f7 100644 --- a/python/packages/hosting-a2a/README.md +++ b/python/packages/hosting-a2a/README.md @@ -1,32 +1,39 @@ # agent-framework-hosting-a2a -A2A conversion helpers for app-owned Agent Framework hosting. +Helpers for composing Agent Framework agents and workflows with an +application-owned native A2A server. -The package deliberately does not choose a web framework or wrap the A2A SDK -server lifecycle. It provides two conversion functions: +The package converts protocol values and can generate the common discovery +fields for a native `AgentCard`. It does not provide an `AgentExecutor`, task +lifecycle, event queue, task store, routes, session policy, authentication, or +deployment. -- `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. +## Choose the level of help -Application code keeps ownership of the A2A SDK's `AgentExecutor`, -`RequestContext`, `TaskUpdater`, event queue, task store, routes, task state, -artifact IDs, authentication, and deployment. +| API | Adds | +| --- | --- | +| `a2a_to_run`, `a2a_from_run` | Native A2A-to-agent value conversion | +| `AgentA2AAdapter` | Agent card generation plus the agent conversion helpers | +| `a2a_to_workflow_run`, `a2a_from_workflow_run` | Typed workflow input and output conversion | +| `WorkflowA2AAdapter` | Workflow card generation plus the workflow conversion helpers | -`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. +Each level is optional. Applications keep using native A2A SDK objects and can +construct an `AgentCard` directly when they need discovery fields beyond the +common generated surface. + +## Agent conversions + +The core helpers work with any native A2A `AgentExecutor`: ```python -run = a2a_to_run(context.message) +run = a2a_to_run(context.message, stream=False) 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"], + stream=run["stream"], ) await state.set_session(session_id, session) parts = a2a_from_run(result) @@ -34,5 +41,145 @@ 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. +`a2a_from_run(...)` returns a flat part list and preserves content-level +metadata. The application decides how to group those parts into A2A messages +or artifacts and owns their message-level metadata and boundaries. + +Standalone conversions are permissive by default. Pass `input_modes` or +`output_modes` to validate the converted parts against an advertised contract: + +```python +run = a2a_to_run(message, input_modes=["text", "image/*"]) +parts = a2a_from_run(result, output_modes=["text"]) +``` + +### Mode parsing + +A2A mode strings are extensible; there is no exhaustive protocol-wide list. +The helpers have an exhaustive set of built-in parsing behaviors: + +| Mode | Automatic behavior | +| --- | --- | +| `text` | Uses A2A text parts and Agent Framework text content | +| `application/json` | Parses JSON text into an A2A data part and parses A2A data into typed workflow input | +| `application/octet-stream` | Uses raw byte parts | +| Concrete media types such as `image/png` or `audio/wav` | Preserves matching raw or URL parts | +| Wildcards such as `image/*` | Validates matching concrete media types | + +Custom mode strings may still be advertised. They pass validation when the +native part already carries that exact media type, but conversion raises when +it would need to synthesize that representation without a built-in parser. +Configured mode values must be non-empty strings. + +## Supported interfaces + +`supported_interfaces` tells an A2A client where and how it can call the +application. Add one `AgentInterface` for each protocol binding the server +actually exposes: + +```python +supported_interfaces = [ + AgentInterface( + url="https://example.com/a2a", + protocol_binding="JSONRPC", + ) +] +``` + +The `url` is the public base URL where the matching A2A routes are mounted; +include a path such as `/a2a` when the application mounts them below the +domain root. `protocol_binding` identifies the wire protocol implemented at +that URL, commonly `JSONRPC`, `HTTP+JSON`, or `GRPC`. Advertise only bindings +that the application has configured. `protocol_version` and `tenant` are +optional native A2A interface fields for deployments that use them. + +## Generate an agent card + +`AgentA2AAdapter` infers the public name and description from the agent and uses +conservative text input/output modes. Pass either an agent or an existing +`AgentState`; `get_card()` is async so factory-backed states can resolve their +target. + +By default, the card also discovers Agent Framework `Skill` values from +`SkillsProvider` instances on the agent. The guaranteed skill frontmatter +name and description become a native A2A `AgentSkill`, using the card's input +and output modes. Set `infer_skills=False` to disable discovery. The `skills` +parameter also accepts explicit Agent Framework `Skill` values or fully +specified native A2A `AgentSkill` values when tags, examples, security, or +skill-specific modes need to be controlled directly. Card discovery happens +outside an agent run, so context-aware skill sources receive no session; use +explicit skills or disable inference when the advertised list is +session-specific. + +Server capabilities stay explicit because they describe the public +application contract, not the agent's `run` method. + +The adapter re-exposes `a2a_to_run(...)` and `a2a_from_run(...)`, so a native +executor can use the same object for card setup and request conversion without +importing the standalone helpers. Adapter conversions validate against the +configured card modes by default; pass `validate_modes=False` to opt out: + +```python +run = adapter.a2a_to_run(context.message, stream=True) +parts = adapter.a2a_from_run(result) +``` + +```python +from a2a.types import AgentCapabilities, AgentInterface +from agent_framework_hosting_a2a import AgentA2AAdapter + +card = await AgentA2AAdapter( + state, + version="1.0.0", + supported_interfaces=[ + AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC") + ], + capabilities=AgentCapabilities(streaming=True), +).get_card() +``` + +## Host a workflow + +Workflow input conversion follows the single start-executor input type: + +- strings use one A2A text part; +- bytes use one raw part; +- structured and scalar JSON values use one data part. + +The output helper converts public workflow outputs to native A2A parts. +Pending human-input requests raise so the application can implement its own +continuation policy. + +```python +workflow_input = a2a_to_workflow_run(context.message, workflow) +result = await workflow.run(workflow_input, stream=False) +parts = a2a_from_workflow_run(result) +``` + +`WorkflowA2AAdapter` infers modes from the workflow's declared input and output +types. It accepts a workflow or `WorkflowState`. Supply explicit modes for an +application-specific representation: + +```python +card = await WorkflowA2AAdapter( + workflow_state, + version="1.0.0", + supported_interfaces=[ + AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC") + ], + skills=[workflow_skill], +).get_card() +``` + +It also exposes `await adapter.a2a_to_run(message)` and +`adapter.a2a_from_run(result)` for workflow conversion. These methods validate +against the effective card modes by default. When workflow output modes are +inferred, call `get_card()` before converting output so the adapter has +resolved the advertised contract. + +Streaming workflow progress, artifacts, task status, checkpoints, and +human-in-the-loop continuation remain part of the native executor and +application contract. + +The surrounding application may use Starlette, FastAPI, another ASGI +framework, or the A2A SDK's application builders. diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py index 70e446d97d..3a955d6e56 100644 --- a/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py @@ -4,7 +4,8 @@ import importlib.metadata -from ._conversion import a2a_from_run, a2a_to_run +from ._adapters import AgentA2AAdapter, WorkflowA2AAdapter +from ._conversion import a2a_from_run, a2a_from_workflow_run, a2a_to_run, a2a_to_workflow_run try: __version__ = importlib.metadata.version(__name__) @@ -12,7 +13,11 @@ __version__ = "0.0.0" __all__ = [ + "AgentA2AAdapter", + "WorkflowA2AAdapter", "__version__", "a2a_from_run", + "a2a_from_workflow_run", "a2a_to_run", + "a2a_to_workflow_run", ] diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_adapters.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_adapters.py new file mode 100644 index 0000000000..fafbaab7bc --- /dev/null +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_adapters.py @@ -0,0 +1,393 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Native A2A card adapters for Agent Framework targets.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any, Generic, TypeVar, cast + +from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill, Part +from a2a.types import Message as A2AMessage +from agent_framework import ( + AgentResponse, + AgentResponseUpdate, + Message, + Skill, + SkillsProvider, + SkillsSourceContext, + SupportsAgentRun, + Workflow, + WorkflowRunResult, +) +from agent_framework_hosting import AgentRunArgs, AgentState, WorkflowState + +from ._conversion import ( + _normalized_modes, # pyright: ignore[reportPrivateUsage] + _workflow_input_type, # pyright: ignore[reportPrivateUsage] + _workflow_type_modes, # pyright: ignore[reportPrivateUsage] +) +from ._conversion import ( + a2a_from_run as _a2a_from_run, +) +from ._conversion import ( + a2a_from_workflow_run as _a2a_from_workflow_run, +) +from ._conversion import ( + a2a_to_run as _a2a_to_run, +) +from ._conversion import ( + a2a_to_workflow_run as _a2a_to_workflow_run, +) + +AgentT = TypeVar("AgentT", bound=SupportsAgentRun) +WorkflowT = TypeVar("WorkflowT", bound=Workflow) + + +def _to_a2a_skill(skill: AgentSkill | Skill, input_modes: Sequence[str], output_modes: Sequence[str]) -> AgentSkill: + if isinstance(skill, AgentSkill): + return skill + + frontmatter = skill.frontmatter + return AgentSkill( + id=frontmatter.name, + name=frontmatter.name, + description=frontmatter.description, + tags=[frontmatter.name], + examples=[], + input_modes=input_modes, + output_modes=output_modes, + ) + + +class AgentA2AAdapter(Generic[AgentT]): + """Resolve a native A2A card from Agent Framework agent metadata.""" + + def __init__( + self, + target: AgentT | AgentState[AgentT], + *, + version: str, + supported_interfaces: Sequence[AgentInterface], + skills: Sequence[AgentSkill | Skill] = (), + infer_skills: bool = True, + capabilities: AgentCapabilities | None = None, + name: str | None = None, + description: str | None = None, + default_input_modes: Sequence[str] = ("text",), + default_output_modes: Sequence[str] = ("text",), + ) -> None: + """Create an agent-backed A2A card adapter. + + Args: + target: Agent or existing ``AgentState`` whose public metadata + should seed the card. + + Keyword Args: + version: Application-defined version advertised by the A2A server. + supported_interfaces: Public A2A endpoints exposed by the application, + with one native interface for each available protocol binding. For + example, ``from a2a.types import AgentInterface`` and then + ``[AgentInterface(url="https://example.com/a2a", + protocol_binding="JSONRPC")]``. + skills: Explicit native A2A or Agent Framework skills. + infer_skills: Whether to discover Agent Framework skills from + ``SkillsProvider`` instances on the resolved agent. Defaults to ``True``. + capabilities: Native server capabilities. Defaults to no optional capabilities. + name: Public card name override. Defaults to the resolved target name. + description: Public card description override. Defaults to the resolved target description. + default_input_modes: Input modes advertised by the card. Defaults to + ``("text",)``. Other A2A mode strings may include + ``"application/json"``, ``"application/octet-stream"``, or media + types such as ``"image/png"``; these examples are not exhaustive. + default_output_modes: Output modes advertised by the card. Defaults + to ``("text",)`` and accepts the same A2A mode strings as + ``default_input_modes``. + + Raises: + ValueError: If required application-owned card values are missing. + """ + if not version: + raise ValueError("An A2A agent card requires a version.") + if not supported_interfaces: + raise ValueError("An A2A agent card requires at least one supported interface.") + if not default_input_modes or not default_output_modes: + raise ValueError("An A2A agent card requires at least one input and output mode.") + _normalized_modes(default_input_modes) + _normalized_modes(default_output_modes) + + self.state = target if isinstance(target, AgentState) else AgentState(target) + self.version = version + self.supported_interfaces = tuple(supported_interfaces) + self.skills = tuple(skills) + self.infer_skills = infer_skills + self.capabilities = capabilities + self.name = name + self.description = description + self.default_input_modes = tuple(default_input_modes) + self.default_output_modes = tuple(default_output_modes) + + async def get_card(self) -> AgentCard: + """Return the native A2A card for the resolved agent. + + Returns: + A native A2A ``AgentCard``. + + Raises: + ValueError: If the resolved target does not provide required public metadata. + """ + target = await self.state.get_target() + card_name = self.name if self.name is not None else target.name + card_description = self.description if self.description is not None else target.description + if not card_name: + raise ValueError("An A2A agent card requires a name.") + if not card_description: + raise ValueError("An A2A agent card requires a description.") + + skills: list[AgentSkill | Skill] = list(self.skills) + if self.infer_skills: + providers = cast("Sequence[object]", getattr(target, "context_providers", ())) + source_context = SkillsSourceContext(agent=target) + for provider in providers: + if isinstance(provider, SkillsProvider): + discovered = await provider._source.get_skills( # pyright: ignore[reportPrivateUsage] + source_context + ) + skills.extend(discovered) + + card_skills: list[AgentSkill] = [] + skill_ids: set[str] = set() + for skill in skills: + a2a_skill = _to_a2a_skill(skill, self.default_input_modes, self.default_output_modes) + if a2a_skill.id not in skill_ids: + skill_ids.add(a2a_skill.id) + card_skills.append(a2a_skill) + + return AgentCard( + name=card_name, + description=card_description, + version=self.version, + default_input_modes=self.default_input_modes, + default_output_modes=self.default_output_modes, + capabilities=self.capabilities if self.capabilities is not None else AgentCapabilities(), + supported_interfaces=self.supported_interfaces, + skills=card_skills, + ) + + def a2a_to_run( + self, + message: A2AMessage, + *, + stream: bool = False, + validate_modes: bool = True, + ) -> AgentRunArgs: + """Convert a native A2A message into Agent Framework run arguments. + + Args: + message: Native A2A message to convert. + + Keyword Args: + stream: Whether the caller intends to run the agent in streaming mode. + validate_modes: Whether to validate parts against the card's + ``default_input_modes``. Defaults to ``True``. + + Returns: + Arguments corresponding to ``Agent.run(...)``. + + Raises: + ValueError: If mode validation fails. + """ + return _a2a_to_run( + message, + stream=stream, + input_modes=self.default_input_modes if validate_modes else None, + ) + + def a2a_from_run( + self, + result: AgentResponse[Any] | Message | AgentResponseUpdate, + *, + validate_modes: bool = True, + ) -> list[Part]: + """Convert Agent Framework output into native A2A parts. + + Args: + result: Completed response, response message, or streaming update. + + Keyword Args: + validate_modes: Whether to validate parts against the card's + ``default_output_modes``. Defaults to ``True``. + + Returns: + Native A2A parts ready for an A2A SDK message or artifact. + + Raises: + ValueError: If mode validation fails. + """ + return _a2a_from_run( + result, + output_modes=self.default_output_modes if validate_modes else None, + ) + + +class WorkflowA2AAdapter(Generic[WorkflowT]): + """Resolve a native A2A card from Agent Framework workflow metadata.""" + + def __init__( + self, + target: WorkflowT | WorkflowState[WorkflowT], + *, + version: str, + supported_interfaces: Sequence[AgentInterface], + skills: Sequence[AgentSkill | Skill] = (), + capabilities: AgentCapabilities | None = None, + name: str | None = None, + description: str | None = None, + default_input_modes: Sequence[str] | None = None, + default_output_modes: Sequence[str] | None = None, + ) -> None: + """Create a workflow-backed A2A card adapter. + + Args: + target: Workflow or existing ``WorkflowState`` whose metadata and + declared types should seed the card. + + Keyword Args: + version: Application-defined version advertised by the A2A server. + supported_interfaces: Public A2A endpoints exposed by the application, + with one native interface for each available protocol binding. For + example, ``from a2a.types import AgentInterface`` and then + ``[AgentInterface(url="https://example.com/a2a", + protocol_binding="JSONRPC")]``. + skills: Explicit native A2A or Agent Framework skills. + capabilities: Native server capabilities. Defaults to no optional capabilities. + name: Public card name override. Defaults to the resolved workflow name. + description: Public card description override. Defaults to the resolved workflow description. + default_input_modes: Input mode override. ``None`` or an empty sequence + infers modes from the start executor. Explicit alternatives include + ``"text"``, ``"application/json"``, ``"application/octet-stream"``, + and media types such as ``"image/png"``; A2A mode strings are + extensible, so these examples are not exhaustive. + default_output_modes: Output mode override. ``None`` or an empty sequence + infers modes from declared workflow outputs and accepts the same + explicit A2A mode strings as ``default_input_modes``. + + Raises: + ValueError: If required application-owned card values are missing. + """ + if not version: + raise ValueError("An A2A workflow card requires a version.") + if not supported_interfaces: + raise ValueError("An A2A workflow card requires at least one supported interface.") + if default_input_modes: + _normalized_modes(default_input_modes) + if default_output_modes: + _normalized_modes(default_output_modes) + self.state = target if isinstance(target, WorkflowState) else WorkflowState(target) + self.version = version + self.supported_interfaces = tuple(supported_interfaces) + self.skills = tuple(skills) + self.capabilities = capabilities + self.name = name + self.description = description + self.default_input_modes = tuple(default_input_modes) if default_input_modes else None + self.default_output_modes = tuple(default_output_modes) if default_output_modes else None + self._resolved_output_modes: tuple[str, ...] | None = self.default_output_modes + + async def get_card(self) -> AgentCard: + """Return the native A2A card for the resolved workflow. + + Returns: + A native A2A ``AgentCard``. + + Raises: + ValueError: If required metadata or modes cannot be determined. + """ + target = await self.state.get_target() + card_name = self.name if self.name is not None else target.name + card_description = self.description if self.description is not None else target.description + if not card_name: + raise ValueError("An A2A workflow card requires a name.") + if not card_description: + raise ValueError("An A2A workflow card requires a description.") + + input_modes = ( + list(self.default_input_modes) + if self.default_input_modes is not None + else _workflow_type_modes(_workflow_input_type(target)) + ) + if self.default_output_modes is not None: + output_modes = list(self.default_output_modes) + elif target.output_types: + inferred_modes = {mode for output_type in target.output_types for mode in _workflow_type_modes(output_type)} + output_modes = [ + mode for mode in ("text", "application/octet-stream", "application/json") if mode in inferred_modes + ] + else: + raise ValueError("Cannot infer A2A output modes because the workflow declares no output types.") + self._resolved_output_modes = tuple(output_modes) + + return AgentCard( + name=card_name, + description=card_description, + version=self.version, + default_input_modes=input_modes, + default_output_modes=output_modes, + capabilities=self.capabilities if self.capabilities is not None else AgentCapabilities(), + supported_interfaces=self.supported_interfaces, + skills=[_to_a2a_skill(skill, input_modes, output_modes) for skill in self.skills], + ) + + async def a2a_to_run(self, message: A2AMessage, *, validate_modes: bool = True) -> Any: + """Convert a native A2A message into validated workflow input. + + Args: + message: Native A2A message containing the workflow input. + + Keyword Args: + validate_modes: Whether to validate parts against the card's + effective input modes. Defaults to ``True``. + + Returns: + A value validated for ``Workflow.run(...)``. + + Raises: + ValueError: If workflow input or mode validation fails. + """ + target = await self.state.get_target() + input_modes = ( + list(self.default_input_modes) + if self.default_input_modes is not None + else _workflow_type_modes(_workflow_input_type(target)) + ) + return _a2a_to_workflow_run( + message, + target, + input_modes=input_modes if validate_modes else None, + ) + + def a2a_from_run(self, result: WorkflowRunResult, *, validate_modes: bool = True) -> list[Part]: + """Convert completed workflow outputs into native A2A parts. + + Args: + result: Completed non-streaming workflow result. + + Keyword Args: + validate_modes: Whether to validate parts against the card's + effective output modes. Defaults to ``True``. When output modes + are inferred, call :meth:`get_card` before enabling validation. + + Returns: + Native A2A parts for the workflow's public outputs. + + Raises: + RuntimeError: If validation requires inferred output modes and + :meth:`get_card` has not resolved them. + ValueError: If workflow output or mode validation fails. + """ + if validate_modes and self._resolved_output_modes is None: + raise RuntimeError("Call `await adapter.get_card()` before validating inferred workflow output modes.") + return _a2a_from_workflow_run( + result, + output_modes=self._resolved_output_modes if validate_modes else None, + ) 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 e9f81663bc..583a3e6c62 100644 --- a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_conversion.py @@ -7,19 +7,167 @@ import base64 import json import logging -from collections.abc import Sequence +from collections.abc import Collection, Mapping, 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 import ( + AgentResponse, + AgentResponseUpdate, + ChatOptions, + Content, + Message, + Workflow, + WorkflowRunResult, +) from agent_framework_hosting import AgentRunArgs -from google.protobuf.json_format import MessageToDict +from google.protobuf.json_format import MessageToDict, ParseDict +from google.protobuf.struct_pb2 import Value +from pydantic import TypeAdapter +from pydantic.errors import PydanticSchemaGenerationError logger = logging.getLogger("agent_framework.hosting.a2a") +_BINARY_MODE = "application/octet-stream" +_JSON_MODE = "application/json" +_TEXT_MODE = "text" +_MODE_ORDER = (_TEXT_MODE, _BINARY_MODE, _JSON_MODE) +_JSON_SCHEMA_TYPES = {"array", "boolean", "integer", "null", "number", "object"} -def a2a_to_run(message: A2AMessage, *, stream: bool = False) -> AgentRunArgs: + +def _normalized_modes(allowed_modes: Collection[str]) -> set[str]: + normalized_modes: set[str] = set() + for mode in allowed_modes: + if not isinstance(mode, str) or not mode.strip(): + raise ValueError("A2A modes must be non-empty strings.") + normalized_modes.add(mode.strip().lower()) + return normalized_modes + + +def _mode_allowed(mode: str, normalized_modes: Collection[str]) -> bool: + normalized_mode = mode.lower() + if normalized_mode in normalized_modes: + return True + return any( + allowed_mode.endswith("/*") and normalized_mode.startswith(allowed_mode[:-1]) + for allowed_mode in normalized_modes + ) + + +def _part_mode(part: Part) -> str | None: + match part.WhichOneof("content"): + case "text": + return _TEXT_MODE + case "data": + return _JSON_MODE + case "raw": + return part.media_type or _BINARY_MODE + case "url": + return part.media_type or None + case _: + return None + + +def _validate_part_modes(parts: Sequence[Part], allowed_modes: Collection[str], direction: str) -> None: + normalized_modes = _normalized_modes(allowed_modes) + for part in parts: + mode = _part_mode(part) + if mode is None: + continue + if _mode_allowed(mode, normalized_modes): + continue + raise ValueError( + f"A2A {direction} part mode '{mode}' is not included in the advertised modes: {sorted(allowed_modes)}." + ) + + +def _part_from_text( + text: str, + metadata: Mapping[str, Any], + output_modes: Collection[str] | None, +) -> Part: + if output_modes is None: + return Part(text=text, metadata=metadata) + + normalized_modes = _normalized_modes(output_modes) + if _mode_allowed(_TEXT_MODE, normalized_modes): + return Part(text=text, metadata=metadata) + if _mode_allowed(_JSON_MODE, normalized_modes): + try: + data = json.loads(text) + except json.JSONDecodeError as exc: + raise ValueError("Agent Framework text output is not valid JSON for A2A application/json mode.") from exc + value = Value() + ParseDict(data, value) + return Part(data=value, metadata=metadata) + raise ValueError( + f"Agent Framework text output cannot be converted to the advertised A2A modes: {sorted(output_modes)}." + ) + + +def _modes_for_schema(schema: Mapping[str, Any]) -> list[str]: + modes: set[str] = set() + variants = schema.get("anyOf") or schema.get("oneOf") + if isinstance(variants, list): + for variant in cast("list[object]", variants): + if isinstance(variant, dict): + modes.update(_modes_for_schema(cast("dict[str, Any]", variant))) + + schema_types = schema.get("type") + if isinstance(schema_types, str): + schema_types_list: list[object] = [schema_types] + elif isinstance(schema_types, list): + schema_types_list = cast("list[object]", schema_types) + else: + schema_types_list = [] + for schema_type in schema_types_list: + if isinstance(schema_type, str): + if schema_type == "string": + modes.add(_BINARY_MODE if schema.get("format") == "binary" else _TEXT_MODE) + elif schema_type in _JSON_SCHEMA_TYPES: + modes.add(_JSON_MODE) + + return [mode for mode in _MODE_ORDER if mode in modes] + + +def _workflow_input_type(workflow: Workflow) -> Any: + input_types = workflow.input_types + if len(input_types) != 1: + raise ValueError( + f"A2A workflow helpers require exactly one start-executor input type; found {len(input_types)}." + ) + return input_types[0] + + +def _workflow_type_adapter(value_type: Any) -> TypeAdapter[Any]: + try: + return TypeAdapter(value_type) + except PydanticSchemaGenerationError as exc: + raise ValueError(f"Cannot convert A2A values for workflow type {value_type!r}.") from exc + + +def _workflow_input_adapter(workflow: Workflow) -> TypeAdapter[Any]: + return _workflow_type_adapter(_workflow_input_type(workflow)) + + +def _workflow_type_modes(value_type: Any) -> list[str]: # pyright: ignore[reportUnusedFunction] + try: + schema = _workflow_type_adapter(value_type).json_schema() + except ValueError as exc: + raise ValueError(f"Cannot infer an A2A mode for workflow type {value_type!r}.") from exc + modes = _modes_for_schema(schema) + if not modes: + raise ValueError(f"Cannot infer an A2A mode for workflow type {value_type!r}.") + return modes + + +def a2a_to_run( + message: A2AMessage, + *, + stream: bool = False, + input_modes: Collection[str] | None = None, +) -> AgentRunArgs: """Convert an A2A message into Agent Framework run arguments. A2A text, URL, raw-byte, and structured-data parts become Agent Framework @@ -31,13 +179,19 @@ def a2a_to_run(message: A2AMessage, *, stream: bool = False) -> AgentRunArgs: Keyword Args: stream: Whether the caller intends to run the agent in streaming mode. + input_modes: Advertised A2A input modes to validate against. ``None`` + disables mode validation. Returns: Arguments corresponding to ``Agent.run(...)``. Raises: - ValueError: If the message has no supported content parts. + ValueError: If the message has no supported content parts or contains + a part outside ``input_modes``. """ + if input_modes is not None: + _validate_part_modes(message.parts, input_modes, "input") + contents: list[Content] = [] for part in message.parts: metadata = MessageToDict(part.metadata) if part.metadata else None @@ -97,7 +251,11 @@ def a2a_to_run(message: A2AMessage, *, stream: bool = False) -> AgentRunArgs: ) -def a2a_from_run(result: AgentResponse[Any] | Message | AgentResponseUpdate) -> list[Part]: +def a2a_from_run( + result: AgentResponse[Any] | Message | AgentResponseUpdate, + *, + output_modes: Collection[str] | None = None, +) -> list[Part]: """Convert Agent Framework output into native A2A parts. ``AgentResponse`` values are flattened in message order. User-role @@ -110,11 +268,16 @@ def a2a_from_run(result: AgentResponse[Any] | Message | AgentResponseUpdate) -> Args: result: A completed response, response message, or streaming update. + Keyword Args: + output_modes: Advertised A2A output modes to validate against. ``None`` + disables mode validation. + Returns: Native A2A parts ready for an A2A SDK message or artifact. Raises: - ValueError: If Agent Framework data content contains an invalid data URI. + ValueError: If Agent Framework data content contains an invalid data URI + or produces a part outside ``output_modes``. """ items: Sequence[Message | AgentResponseUpdate] = result.messages if isinstance(result, AgentResponse) else [result] @@ -126,7 +289,7 @@ def a2a_from_run(result: AgentResponse[Any] | Message | AgentResponseUpdate) -> metadata = content.additional_properties or {} match content.type: case "text" if content.text is not None: - parts.append(Part(text=content.text, metadata=metadata)) + parts.append(_part_from_text(content.text, metadata, output_modes)) case "uri" if content.uri is not None: parts.append( Part( @@ -155,4 +318,128 @@ def a2a_from_run(result: AgentResponse[Any] | Message | AgentResponseUpdate) -> "Agent Framework content type %s is not supported by A2A and was omitted.", content.type, ) + if output_modes is not None: + _validate_part_modes(parts, output_modes, "output") + return parts + + +def a2a_to_workflow_run( + message: A2AMessage, + workflow: Workflow, + *, + input_modes: Collection[str] | None = None, +) -> Any: + """Convert one native A2A message part into validated workflow input. + + The workflow must declare exactly one start-executor input type. Text, + raw-byte, and structured-data parts map to string, binary, and JSON + workflow contracts respectively. Executor, task, session, and continuation + behavior remain application-owned. + + Args: + message: Native A2A message containing the workflow input. + workflow: Workflow whose start-executor input contract should be used. + + Keyword Args: + input_modes: Advertised A2A input modes to validate against. ``None`` + disables card-mode validation; workflow type validation still applies. + + Returns: + A value validated for ``Workflow.run(...)``. + + Raises: + ValueError: If the workflow input type is unsupported, the message + does not contain exactly one compatible part, or a part is outside + ``input_modes``. + """ + if input_modes is not None: + _validate_part_modes(message.parts, input_modes, "input") + + adapter = _workflow_input_adapter(workflow) + modes = _modes_for_schema(adapter.json_schema()) + if not modes: + raise ValueError(f"Cannot convert A2A input for workflow type {workflow.input_types[0]!r}.") + + candidates: list[tuple[str, Part]] = [] + for part in message.parts: + content_type = part.WhichOneof("content") + if content_type == "text" and _TEXT_MODE in modes: + candidates.append((_TEXT_MODE, part)) + elif content_type == "raw" and _BINARY_MODE in modes: + candidates.append((_BINARY_MODE, part)) + elif content_type == "data" and _JSON_MODE in modes: + candidates.append((_JSON_MODE, part)) + + if len(candidates) != 1: + expected = ", ".join(modes) + raise ValueError( + f"A2A workflow input must contain exactly one compatible part for modes [{expected}]; " + f"found {len(candidates)}." + ) + + mode, part = candidates[0] + if mode == _TEXT_MODE: + value: Any = part.text + elif mode == _BINARY_MODE: + value = part.raw + else: + value = MessageToDict(part.data) + return adapter.validate_python(value) + + +def a2a_from_workflow_run( + result: WorkflowRunResult, + *, + output_modes: Collection[str] | None = None, +) -> list[Part]: + """Convert completed workflow outputs into native A2A parts. + + Args: + result: Completed non-streaming workflow result. + + Keyword Args: + output_modes: Advertised A2A output modes to validate against. ``None`` + disables mode validation. + + Returns: + Native A2A parts for the workflow's public outputs. + + Raises: + ValueError: If the workflow requires external input or produces a part + outside ``output_modes``. + """ + if result.get_request_info_events(): + raise ValueError( + "The workflow requires external input. A2A workflow conversion does not manage " + "human-in-the-loop continuation; handle it in the application contract." + ) + + parts: list[Part] = [] + for output in result.get_outputs(): + if isinstance(output, (AgentResponse, Message, AgentResponseUpdate)): + parts.extend( + a2a_from_run( + cast("AgentResponse[Any] | Message | AgentResponseUpdate", output), + output_modes=output_modes, + ) + ) + elif isinstance(output, str): + parts.append(_part_from_text(output, {}, output_modes)) + elif isinstance(output, bytes): + parts.append(Part(raw=output, media_type=_BINARY_MODE)) + else: + data = TypeAdapter(object).dump_python(output, mode="json", serialize_as_any=True) + normalized_modes = _normalized_modes(output_modes) if output_modes is not None else None + if ( + normalized_modes is not None + and _mode_allowed(_TEXT_MODE, normalized_modes) + and not _mode_allowed(_JSON_MODE, normalized_modes) + ): + parts.append(Part(text=json.dumps(data, separators=(",", ":"), sort_keys=True))) + else: + value = Value() + ParseDict(data, value) + parts.append(Part(data=value)) + if output_modes is not None: + _validate_part_modes(parts, output_modes, "output") return parts diff --git a/python/packages/hosting-a2a/tests/hosting_a2a/test_adapters.py b/python/packages/hosting-a2a/tests/hosting_a2a/test_adapters.py new file mode 100644 index 0000000000..c058c8da47 --- /dev/null +++ b/python/packages/hosting-a2a/tests/hosting_a2a/test_adapters.py @@ -0,0 +1,244 @@ +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace +from typing import cast + +from a2a.types import AgentCapabilities, AgentInterface, AgentSkill, Part, Role +from a2a.types import Message as A2AMessage +from agent_framework import ( + InlineSkill, + Message, + SkillFrontmatter, + SkillsProvider, + SupportsAgentRun, + WorkflowBuilder, + WorkflowContext, + WorkflowEvent, + WorkflowRunResult, + executor, +) +from agent_framework_hosting import AgentState, WorkflowState +from google.protobuf.json_format import ParseDict +from pytest import raises + +from agent_framework_hosting_a2a import AgentA2AAdapter, WorkflowA2AAdapter + + +@dataclass +class StructuredInput: + value: str + + +async def test_agent_card_infers_metadata_and_keeps_a2a_policy_explicit() -> None: + target = cast("SupportsAgentRun", SimpleNamespace(name="Travel Agent", description="Plans trips.")) + interface = AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC") + skill = AgentSkill( + id="plan_trip", + name="Plan trip", + description="Plan a trip.", + tags=["travel"], + examples=["Plan a weekend in Paris."], + ) + + adapter = AgentA2AAdapter( + AgentState(lambda: target), + version="1.0.0", + supported_interfaces=[interface], + skills=[skill], + capabilities=AgentCapabilities(streaming=True), + ) + card = await adapter.get_card() + + assert card.name == "Travel Agent" + assert card.description == "Plans trips." + assert card.default_input_modes == ["text"] + assert card.default_output_modes == ["text"] + assert card.capabilities.streaming is True + assert card.supported_interfaces == [interface] + assert card.skills == [skill] + run = adapter.a2a_to_run(A2AMessage(message_id="message-1", role=Role.ROLE_USER, parts=[Part(text="hello")])) + assert isinstance(run["messages"], list) + assert isinstance(run["messages"][0], Message) + assert run["messages"][0].text == "hello" + assert adapter.a2a_from_run(Message("assistant", ["hello"]))[0].text == "hello" + with raises(ValueError, match="audio/wav"): + adapter.a2a_to_run( + A2AMessage( + message_id="message-2", + role=Role.ROLE_USER, + parts=[Part(raw=b"audio", media_type="audio/wav")], + ) + ) + assert adapter.a2a_to_run( + A2AMessage( + message_id="message-2", + role=Role.ROLE_USER, + parts=[Part(raw=b"audio", media_type="audio/wav")], + ), + validate_modes=False, + )["messages"] + + +async def test_agent_card_requires_public_metadata_and_interface() -> None: + target = cast("SupportsAgentRun", SimpleNamespace(name=None, description=None)) + + with raises(ValueError, match="supported interface"): + AgentA2AAdapter(target, version="1.0.0", supported_interfaces=[]) + + with raises(ValueError, match="non-empty strings"): + AgentA2AAdapter( + target, + version="1.0.0", + supported_interfaces=[AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC")], + default_input_modes=[""], + ) + + with raises(ValueError, match="requires a name"): + await AgentA2AAdapter( + target, + version="1.0.0", + supported_interfaces=[AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC")], + ).get_card() + + +async def test_agent_card_infers_agent_framework_skills_and_can_disable_inference() -> None: + skill = InlineSkill( + frontmatter=SkillFrontmatter(name="plan-trip", description="Plan a trip."), + instructions="Help the user plan a trip.", + ) + target = cast( + "SupportsAgentRun", + SimpleNamespace( + name="Travel Agent", + description="Plans trips.", + context_providers=[SkillsProvider([skill])], + ), + ) + interface = AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC") + + inferred_card = await AgentA2AAdapter( + target, + version="1.0.0", + supported_interfaces=[interface], + ).get_card() + explicit_card = await AgentA2AAdapter( + target, + version="1.0.0", + supported_interfaces=[interface], + skills=[skill], + infer_skills=False, + ).get_card() + disabled_card = await AgentA2AAdapter( + target, + version="1.0.0", + supported_interfaces=[interface], + infer_skills=False, + ).get_card() + + assert inferred_card.skills[0].id == "plan-trip" + assert inferred_card.skills[0].description == "Plan a trip." + assert inferred_card.skills[0].input_modes == ["text"] + assert explicit_card.skills[0] == inferred_card.skills[0] + assert disabled_card.skills == [] + + +async def test_workflow_card_infers_json_input_and_text_output_modes() -> None: + @executor(id="structured") + async def structured(value: StructuredInput, ctx: WorkflowContext[object, str]) -> None: + await ctx.yield_output(value.value) + + workflow = WorkflowBuilder( + start_executor=structured, + name="Structured Workflow", + description="Processes structured input.", + output_from=[structured], + ).build() + + adapter = WorkflowA2AAdapter( + WorkflowState(lambda: workflow), + version="1.0.0", + supported_interfaces=[AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC")], + default_input_modes=[], + default_output_modes=[], + ) + output_result = WorkflowRunResult([WorkflowEvent("output", "hello", executor_id="structured")]) + with raises(RuntimeError, match="get_card"): + adapter.a2a_from_run(output_result) + assert adapter.a2a_from_run(output_result, validate_modes=False)[0].text == "hello" + + card = await adapter.get_card() + data_part = Part() + ParseDict({"value": "hello"}, data_part.data) + workflow_input = await adapter.a2a_to_run( + A2AMessage(message_id="message-1", role=Role.ROLE_USER, parts=[data_part]) + ) + output_parts = adapter.a2a_from_run(output_result) + + assert card.name == "Structured Workflow" + assert card.description == "Processes structured input." + assert card.default_input_modes == ["application/json"] + assert card.default_output_modes == ["text"] + assert workflow_input == StructuredInput(value="hello") + assert output_parts[0].text == "hello" + + +async def test_workflow_card_allows_explicit_modes_for_unsupported_types() -> None: + @executor(id="custom") + async def custom(value: object, ctx: WorkflowContext[object, object]) -> None: + await ctx.yield_output(value) + + workflow = WorkflowBuilder( + start_executor=custom, + name="Custom Workflow", + description="Processes a custom protocol value.", + output_from=[custom], + ).build() + interface = AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC") + + with raises(ValueError, match="Cannot infer"): + await WorkflowA2AAdapter( + workflow, + version="1.0.0", + supported_interfaces=[interface], + ).get_card() + + card = await WorkflowA2AAdapter( + workflow, + version="1.0.0", + supported_interfaces=[interface], + default_input_modes=["application/x-custom"], + default_output_modes=["application/x-custom"], + ).get_card() + + assert card.default_input_modes == ["application/x-custom"] + assert card.default_output_modes == ["application/x-custom"] + + +async def test_workflow_card_requires_one_input_type() -> None: + from agent_framework import Executor, handler + + class MultipleInputs(Executor): + @handler + async def handle_text(self, value: str, ctx: WorkflowContext[object, str]) -> None: + await ctx.yield_output(value) + + @handler + async def handle_number(self, value: int, ctx: WorkflowContext[object, str]) -> None: + await ctx.yield_output(str(value)) + + workflow = WorkflowBuilder( + start_executor=MultipleInputs(id="multiple"), + name="Multiple", + description="Multiple inputs.", + output_from="all", + ).build() + + with raises(ValueError, match="exactly one"): + await WorkflowA2AAdapter( + workflow, + version="1.0.0", + supported_interfaces=[AgentInterface(url="https://example.com/a2a", protocol_binding="JSONRPC")], + ).get_card() 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 47dd94a3a4..b8491f2ea7 100644 --- a/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py +++ b/python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py @@ -1,12 +1,48 @@ # Copyright (c) Microsoft. All rights reserved. +from dataclasses import dataclass + 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 agent_framework import ( + AgentResponse, + AgentResponseUpdate, + Content, + Message, + WorkflowBuilder, + WorkflowContext, + WorkflowEvent, + WorkflowRunResult, + executor, +) +from google.protobuf.json_format import MessageToDict, ParseDict from pytest import raises -from agent_framework_hosting_a2a import a2a_from_run, a2a_to_run +from agent_framework_hosting_a2a import ( + a2a_from_run, + a2a_from_workflow_run, + a2a_to_run, + a2a_to_workflow_run, +) + + +@dataclass +class WorkflowInput: + text: str + repeat: int + + +def create_workflow(): + @executor(id="repeat") + async def repeat_text(value: WorkflowInput, ctx: WorkflowContext[object, str]) -> None: + await ctx.yield_output(value.text * value.repeat) + + return WorkflowBuilder( + start_executor=repeat_text, + name="Repeat Workflow", + description="Repeat text a requested number of times.", + output_from=[repeat_text], + ).build() def test_a2a_to_run_converts_supported_parts() -> None: @@ -52,7 +88,8 @@ def test_a2a_to_run_omits_unsupported_parts() -> None: message_id="message-1", role=Role.ROLE_USER, parts=[Part(), Part(text="hello")], - ) + ), + input_modes=[" text "], ) messages = run["messages"] @@ -62,6 +99,19 @@ def test_a2a_to_run_omits_unsupported_parts() -> None: assert converted.text == "hello" +def test_a2a_to_run_validates_advertised_input_modes() -> None: + message = A2AMessage( + message_id="message-1", + role=Role.ROLE_USER, + parts=[Part(raw=b"audio", media_type="audio/wav")], + ) + + with raises(ValueError, match="audio/wav"): + a2a_to_run(message, input_modes=["text"]) + + assert a2a_to_run(message, input_modes=[" audio/* "])["messages"] + + def test_a2a_from_run_converts_final_response() -> None: response = AgentResponse( messages=[ @@ -128,6 +178,35 @@ def test_a2a_from_run_omits_user_messages() -> None: assert a2a_from_run(AgentResponse(messages=[Message("user", ["omit me"])])) == [] +def test_a2a_from_run_validates_advertised_output_modes() -> None: + result = Message( + "assistant", + [Content.from_uri("https://example.com/image.png", media_type="image/png")], + ) + + with raises(ValueError, match="image/png"): + a2a_from_run(result, output_modes=["text"]) + + assert a2a_from_run(result, output_modes=["image/*"])[0].url == "https://example.com/image.png" + + +def test_a2a_from_run_parses_json_text_for_json_output_mode() -> None: + parts = a2a_from_run( + Message("assistant", ['{"answer":42}']), + output_modes=["application/json"], + ) + + assert MessageToDict(parts[0].data) == {"answer": 42.0} + + with raises(ValueError, match="not valid JSON"): + a2a_from_run(Message("assistant", ["not json"]), output_modes=["application/json"]) + + +def test_a2a_from_run_rejects_text_for_incompatible_output_mode() -> None: + with raises(ValueError, match="cannot be converted"): + a2a_from_run(Message("assistant", ["hello"]), output_modes=["application/octet-stream"]) + + def test_a2a_from_run_rejects_invalid_data_uri() -> None: content = Content("data", uri="not-a-data-uri", media_type="application/octet-stream") @@ -144,3 +223,116 @@ def test_a2a_from_run_rejects_invalid_base64_data() -> None: with raises(ValueError, match="invalid base64"): a2a_from_run(Message("assistant", [content])) + + +def test_a2a_to_workflow_run_validates_structured_input() -> None: + data_part = Part() + ParseDict({"text": "go", "repeat": 2}, data_part.data) + + value = a2a_to_workflow_run( + A2AMessage(message_id="message-1", role=Role.ROLE_USER, parts=[Part(), data_part]), + create_workflow(), + input_modes=["application/json"], + ) + + assert value == WorkflowInput(text="go", repeat=2) + + +def test_a2a_to_workflow_run_rejects_invalid_structured_input() -> None: + data_part = Part() + ParseDict({"text": "go", "repeat": "not_a_number"}, data_part.data) + + with raises(ValueError, match="repeat"): + a2a_to_workflow_run( + A2AMessage(message_id="message-1", role=Role.ROLE_USER, parts=[data_part]), + create_workflow(), + ) + + +def test_a2a_to_workflow_run_supports_text_and_binary_inputs() -> None: + @executor(id="text") + async def text_input(value: str, ctx: WorkflowContext[object, str]) -> None: + await ctx.yield_output(value) + + @executor(id="binary") + async def binary_input(value: bytes, ctx: WorkflowContext[object, bytes]) -> None: + await ctx.yield_output(value) + + text_workflow = WorkflowBuilder(start_executor=text_input, output_from=[text_input]).build() + binary_workflow = WorkflowBuilder(start_executor=binary_input, output_from=[binary_input]).build() + + assert ( + a2a_to_workflow_run( + A2AMessage(message_id="text", role=Role.ROLE_USER, parts=[Part(text="hello")]), + text_workflow, + ) + == "hello" + ) + assert ( + a2a_to_workflow_run( + A2AMessage(message_id="binary", role=Role.ROLE_USER, parts=[Part(raw=b"data")]), + binary_workflow, + ) + == b"data" + ) + + +def test_a2a_to_workflow_run_requires_one_compatible_part() -> None: + @executor(id="text") + async def text_input(value: str, ctx: WorkflowContext[object, str]) -> None: + await ctx.yield_output(value) + + with raises(ValueError, match="exactly one compatible"): + a2a_to_workflow_run( + A2AMessage( + message_id="message-1", + role=Role.ROLE_USER, + parts=[Part(text="one"), Part(text="two")], + ), + WorkflowBuilder(start_executor=text_input, output_from=[text_input]).build(), + ) + + +def test_a2a_from_workflow_run_converts_public_outputs() -> None: + result = WorkflowRunResult([ + WorkflowEvent("output", "hello", executor_id="text"), + WorkflowEvent("output", b"data", executor_id="binary"), + WorkflowEvent("output", {"count": 2}, executor_id="structured"), + WorkflowEvent( + "output", + AgentResponse(messages=[Message("assistant", ["from agent"])]), + executor_id="agent", + ), + ]) + + parts = a2a_from_workflow_run(result) + + assert parts[0].text == "hello" + assert parts[1].raw == b"data" + assert parts[1].media_type == "application/octet-stream" + assert MessageToDict(parts[2].data) == {"count": 2.0} + assert parts[3].text == "from agent" + + +def test_a2a_from_workflow_run_serializes_structured_output_for_text_mode() -> None: + result = WorkflowRunResult([WorkflowEvent("output", {"count": 2}, executor_id="structured")]) + + text_parts = a2a_from_workflow_run(result, output_modes=["text"]) + json_parts = a2a_from_workflow_run(result, output_modes=["application/json"]) + + assert text_parts[0].text == '{"count":2}' + assert MessageToDict(json_parts[0].data) == {"count": 2.0} + + +def test_a2a_from_workflow_run_rejects_pending_input() -> None: + result = WorkflowRunResult([ + WorkflowEvent.request_info( + request_id="approval", + source_executor_id="review", + request_data={"question": "Approve?"}, + response_type=bool, + ) + ]) + + with raises(ValueError, match="requires external input"): + a2a_from_workflow_run(result) diff --git a/python/samples/04-hosting/a2a/README.md b/python/samples/04-hosting/a2a/README.md index 990d0ca0cf..730ca70270 100644 --- a/python/samples/04-hosting/a2a/README.md +++ b/python/samples/04-hosting/a2a/README.md @@ -14,7 +14,7 @@ in application code. The helper package does not choose a web framework. | Run this file | To... | |---------------|-------| | **[`a2a_server.py`](a2a_server.py)** | Host an Agent Framework agent as an A2A-compliant server (multi-agent). | -| **[`agent_framework_to_a2a.py`](agent_framework_to_a2a.py)** | Minimal example: expose a single agent as an A2A server. | +| **[`agent_framework_to_a2a.py`](agent_framework_to_a2a.py)** | Expose a single agent with conversion helpers and a native `AgentCard` inferred from agent and skill metadata; the A2A server remains application-owned. | ## Supporting Modules 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 b2af0f561a..6a3e3bf77e 100644 --- a/python/samples/04-hosting/a2a/agent_framework_to_a2a.py +++ b/python/samples/04-hosting/a2a/agent_framework_to_a2a.py @@ -1,9 +1,9 @@ # Copyright (c) Microsoft. All rights reserved. +import asyncio import logging import uuid -from asyncio import CancelledError -from typing import Generic, TypeVar +from typing import Any import uvicorn from a2a.helpers import new_task_from_user_message @@ -14,16 +14,14 @@ from a2a.server.tasks import InMemoryTaskStore, TaskUpdater from a2a.types import ( AgentCapabilities, - AgentCard, AgentInterface, - AgentSkill, Part, TaskState, ) -from agent_framework import Agent, SupportsAgentRun +from agent_framework import Agent, InlineSkill, SkillFrontmatter, SkillsProvider 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 agent_framework_hosting_a2a import AgentA2AAdapter from dotenv import load_dotenv from starlette.applications import Starlette @@ -31,14 +29,12 @@ logger = logging.getLogger(__name__) -AgentT = TypeVar("AgentT", bound=SupportsAgentRun) - -class AppAgentExecutor(AgentExecutor, Generic[AgentT]): +class AppAgentExecutor(AgentExecutor): """Native A2A SDK executor composed with Agent Framework conversion helpers.""" - def __init__(self, state: AgentState[AgentT]) -> None: - self.state = state + 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: @@ -59,11 +55,11 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non await updater.submit() try: await updater.start_work() - run = a2a_to_run(context.message, stream=True) - agent = await self.state.get_target() + 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.state.get_or_create_session(session_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] @@ -75,7 +71,7 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non default_artifact_id = uuid.uuid4().hex streamed_artifact_ids: set[str] = set() async for update in stream: - parts = a2a_from_run(update) + parts = self.adapter.a2a_from_run(update) if parts: artifact_id = update.message_id or default_artifact_id await updater.add_artifact( @@ -86,15 +82,15 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non streamed_artifact_ids.add(artifact_id) final_response = await stream.get_final_response() if not streamed_artifact_ids: - parts = a2a_from_run(final_response) + 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.state.set_session(session_id, session) + await self.adapter.state.set_session(session_id, session) await updater.complete() - except CancelledError: + except asyncio.CancelledError: await updater.update_status(state=TaskState.TASK_STATE_CANCELED) except Exception: logger.exception("A2A agent execution failed.") @@ -105,51 +101,38 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non if __name__ == "__main__": - # --8<-- [start:AgentSkill] - flight_skill = AgentSkill( - id="Flight_Booking", - name="Flight Booking", - description="Search and book flights across Europe.", - tags=["flights", "travel", "europe"], - examples=[], - ) - hotel_skill = AgentSkill( - id="Hotel_Booking", - name="Hotel Booking", - description="Search and book hotels across Europe.", - tags=["hotels", "travel", "accommodation"], - examples=[], + flight_skill = InlineSkill( + frontmatter=SkillFrontmatter( + name="flight-booking", + description="Search and book flights across Europe.", + ), + instructions="Help users search and book flights across Europe.", ) - # --8<-- [end:AgentSkill] - - # --8<-- [start:AgentCard] - # 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." + hotel_skill = InlineSkill( + frontmatter=SkillFrontmatter( + name="hotel-booking", + description="Search and book hotels across Europe.", ), - version="1.0.0", - default_input_modes=["text"], - default_output_modes=["text"], - capabilities=AgentCapabilities(streaming=True), - supported_interfaces=[AgentInterface(url="http://localhost:9999/", protocol_binding="JSONRPC")], - skills=[flight_skill, hotel_skill], + instructions="Help users search and book hotels across Europe.", ) - # --8<-- [end:AgentCard] - 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." - ), + 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) + 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(state), + agent_executor=AppAgentExecutor(adapter), task_store=InMemoryTaskStore(), agent_card=public_agent_card, )