diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md index 0cda8f4a1d8..3da74eb7b6e 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-hosting-telegram` | `python/packages/hosting-telegram` | `alpha` | | `agent-framework-hyperlight` | `python/packages/hyperlight` | `beta` | 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..4beab43952e --- /dev/null +++ b/python/packages/hosting-a2a/README.md @@ -0,0 +1,36 @@ +# agent-framework-hosting-a2a + +Agent-to-Agent (A2A) protocol channel for `agent-framework-hosting`. + +Exposes the hosted target (an `Agent` or a `Workflow`) as an A2A peer agent: it +publishes an agent card and JSON-RPC routes and drives every request through the +host pipeline, so host sessions, request metadata, and run/response hooks all +apply. + +```python +from agent_framework.openai import OpenAIChatClient +from agent_framework_hosting import AgentFrameworkHost +from agent_framework_hosting_a2a import A2AChannel + +agent = OpenAIChatClient().as_agent(name="Assistant") + +host = AgentFrameworkHost( + target=agent, + channels=[A2AChannel(url="https://my-host.example.com/")], +) +host.serve(port=8000) +``` + +By default the channel mounts at the app root so the well-known agent card is +reachable at `/.well-known/agent-card.json`, with the JSON-RPC endpoint at `/`. +The A2A `context_id` maps onto the host session (caller-supplied session family). +A default agent card is derived from the target's name and description; pass a +fully-specified `agent_card` to override it. To advertise additional protocol +bindings in the generated card, pass `supported_interfaces`. + +> **Note:** Task state is held in an in-memory A2A task store for this version; it +> is independent of the host's session storage and is not persisted across +> restarts. + +The base host plumbing lives in +[`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/). 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..c2cfab8cad5 --- /dev/null +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/__init__.py @@ -0,0 +1,24 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""A2A (Agent-to-Agent) channel for :mod:`agent_framework_hosting`. + +Exposes the hosted target (an ``Agent`` or a ``Workflow``) as an A2A peer agent +— publishing an agent card and JSON-RPC routes — while routing every request +through the host pipeline so sessions, request metadata, and hooks apply. +""" + +import importlib.metadata + +from ._channel import A2AChannel +from ._executor import HostAgentExecutor + +try: + __version__ = importlib.metadata.version(__name__) +except importlib.metadata.PackageNotFoundError: + __version__ = "0.0.0" + +__all__ = [ + "A2AChannel", + "HostAgentExecutor", + "__version__", +] diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_channel.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_channel.py new file mode 100644 index 00000000000..585725ac636 --- /dev/null +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_channel.py @@ -0,0 +1,141 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""A2A (Agent-to-Agent) channel for :mod:`agent_framework_hosting`. + +Exposes the hosted target as an A2A peer agent: it publishes an agent card and +JSON-RPC routes, and drives every request through the host pipeline via +:class:`HostAgentExecutor`. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +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.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill +from agent_framework_hosting import ( + ChannelContext, + ChannelContribution, + ChannelResponseHook, + ChannelRunHook, +) + +from ._executor import HostAgentExecutor + + +class A2AChannel: + """Channel that exposes the hosted target over the A2A protocol. + + The A2A ``context_id`` maps onto the host session (caller-supplied session + family) and each request is routed through :class:`ChannelContext`, so host + session resolution and hooks apply. + + Note: + Task state is held in an in-memory A2A task store for this version; it + is independent of the host's session storage and is not persisted. + """ + + name: str = "a2a" + + def __init__( + self, + *, + name: str | None = None, + path: str = "", + url: str = "/", + agent_name: str | None = None, + agent_description: str | None = None, + agent_version: str = "1.0.0", + agent_card: AgentCard | None = None, + skills: Sequence[AgentSkill] | None = None, + supported_interfaces: Sequence[AgentInterface] | None = None, + streaming: bool = True, + rpc_url: str = "/", + card_url: str = "/.well-known/agent-card.json", + run_hook: ChannelRunHook | None = None, + response_hook: ChannelResponseHook | None = None, + ) -> None: + """Configure the A2A channel. + + Keyword Args: + name: Override the channel name (defaults to ``"a2a"``). + path: Sub-path to mount the channel under; empty string (default) + mounts the agent-card and JSON-RPC routes at the app root so + the well-known card path is reachable. + url: Public URL advertised in the agent card's interface (the base + URL clients use to reach the JSON-RPC endpoint). + agent_name: Name advertised in the default agent card. Defaults to + the hosted target's name. + agent_description: Description advertised in the default agent card. + Defaults to the hosted target's description. + agent_version: Version advertised in the default agent card. + agent_card: A fully-specified agent card; when provided it takes + precedence over the ``agent_*``/``url``/``skills`` fields. + skills: Skills advertised in the default agent card. + supported_interfaces: Interfaces advertised in the default agent card. + Defaults to one JSON-RPC interface using ``url``. + streaming: Consume the target via streaming and publish incremental + A2A task artifacts (default ``True``). + rpc_url: Path for the JSON-RPC endpoint (relative to ``path``). + card_url: Path for the agent-card endpoint (relative to ``path``). + run_hook: Optional run hook applied to each request. + response_hook: Optional response hook applied to originating replies. + """ + if name is not None: + self.name = name + self.path = path + self._url = url + self._agent_name = agent_name + self._agent_description = agent_description + self._agent_version = agent_version + self._agent_card = agent_card + self._skills = list(skills) if skills is not None else [] + self._supported_interfaces = list(supported_interfaces) if supported_interfaces is not None else None + self._streaming = streaming + self._rpc_url = rpc_url + self._card_url = card_url + self._run_hook = run_hook + self._response_hook = response_hook + + def _build_agent_card(self, context: ChannelContext) -> AgentCard: + """Derive a default agent card from the hosted target, if not supplied.""" + if self._agent_card is not None: + return self._agent_card + target: Any = context.target + name = self._agent_name or getattr(target, "name", None) or self.name + description = self._agent_description or getattr(target, "description", None) or f"{name} (A2A)" + return AgentCard( + name=name, + description=description, + version=self._agent_version, + default_input_modes=["text"], + default_output_modes=["text"], + capabilities=AgentCapabilities(streaming=self._streaming), + supported_interfaces=self._supported_interfaces + or [AgentInterface(url=self._url, protocol_binding="JSONRPC")], + skills=self._skills, + ) + + def contribute(self, context: ChannelContext) -> ChannelContribution: + """Build the A2A request handler and contribute its routes.""" + agent_card = self._build_agent_card(context) + executor = HostAgentExecutor( + context, + channel_name=self.name, + streaming=self._streaming, + run_hook=self._run_hook, + response_hook=self._response_hook, + ) + handler = DefaultRequestHandler( + agent_executor=executor, + task_store=InMemoryTaskStore(), + agent_card=agent_card, + ) + routes = [ + *create_agent_card_routes(agent_card, card_url=self._card_url), + *create_jsonrpc_routes(handler, self._rpc_url), + ] + return ChannelContribution(routes=routes) diff --git a/python/packages/hosting-a2a/agent_framework_hosting_a2a/_executor.py b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_executor.py new file mode 100644 index 00000000000..6deab3e68c1 --- /dev/null +++ b/python/packages/hosting-a2a/agent_framework_hosting_a2a/_executor.py @@ -0,0 +1,243 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Host-routed A2A :class:`AgentExecutor`. + +Unlike ``agent_framework_a2a.A2AExecutor`` (which calls ``agent.run`` directly +and manages its own session), :class:`HostAgentExecutor` routes every incoming +A2A request through the host pipeline via :class:`ChannelContext` — so host +session resolution, request metadata, and run/response hooks all apply. The A2A +``context_id`` maps onto :class:`ChannelSession` (caller-supplied session +family). +""" + +from __future__ import annotations + +import base64 +import re +from asyncio import CancelledError +from dataclasses import replace +from typing import Any, cast + +from a2a.server.agent_execution import AgentExecutor, RequestContext +from a2a.server.events import EventQueue +from a2a.server.tasks import TaskUpdater +from a2a.types import Part, Task, TaskState +from agent_framework import Content +from agent_framework import Message as AFMessage +from agent_framework_hosting import ( + ChannelContext, + ChannelIdentity, + ChannelRequest, + ChannelResponseHook, + ChannelRunHook, + ChannelSession, + logger, +) + +try: + from a2a.helpers import new_task_from_user_message +except ImportError: # pragma: no cover - older a2a-sdk layout + from a2a.utils import new_task_from_user_message # type: ignore[no-redef, attr-defined, import-not-found] + +_DATA_URI_PATTERN = re.compile(r"^data:(?P[^;]+);base64,(?P[A-Za-z0-9+/=]+)$") + + +def _contents_to_parts(contents: list[Content]) -> list[Part]: + """Convert Agent Framework contents into A2A parts (text, uri, inline data).""" + parts: list[Part] = [] + for content in contents: + if content.type == "text": + # Empty text is not "unsupported" — just nothing to emit. + if content.text: + parts.append(Part(text=content.text)) + elif content.type == "uri" and content.uri: + parts.append(Part(url=content.uri, media_type=content.media_type or "")) + elif content.type == "data" and content.uri: + match = _DATA_URI_PATTERN.match(content.uri) + if match is None: + logger.warning("A2AChannel could not parse data URI; omitted.") + continue + parts.append(Part(raw=base64.b64decode(match.group("data")), media_type=content.media_type or "")) + else: + # function_call/function_result/usage etc. are routine intermediate + # content during a turn — debug, not a warning per chunk. + logger.debug("A2AChannel does not support content type: %s. Omitted.", content.type) + return parts + + +def _value_to_parts(value: Any) -> list[Part]: + """Convert workflow outputs and fallback values into A2A parts.""" + if isinstance(value, Content): + return _contents_to_parts([value]) + if isinstance(value, AFMessage): + return _contents_to_parts(list(value.contents)) + if isinstance(value, str): + return [Part(text=value)] + return [Part(text=str(value))] + + +def _strip_options_hook(request: ChannelRequest, **_: Any) -> ChannelRequest: + """Default run hook: remove all parsed options before reaching the agent. + + When no custom ``run_hook`` is configured this prevents untrusted A2A + callers from injecting generation parameters. Supply a custom hook to + forward or transform specific options. + """ + return replace(request, options=None) + + +class HostAgentExecutor(AgentExecutor): + """A2A executor that drives the hosted target through :class:`ChannelContext`.""" + + def __init__( + self, + context: ChannelContext, + *, + channel_name: str, + streaming: bool = True, + run_hook: ChannelRunHook | None = None, + response_hook: ChannelResponseHook | None = None, + ) -> None: + """Bind the executor to the host context. + + Args: + context: The host-supplied :class:`ChannelContext`. + + Keyword Args: + channel_name: The owning channel's name (stamped on requests). + streaming: When ``True`` (default) the target is consumed via + :meth:`ChannelContext.run_stream` and incremental updates are + published as A2A task artifacts; otherwise the full reply is + published as a single working-state message. + run_hook: Optional :data:`ChannelRunHook` applied to the request. + When omitted, a default hook that strips all caller-supplied + options is applied so untrusted A2A callers cannot inject + generation parameters. + response_hook: Optional :data:`ChannelResponseHook` applied to the + originating final response. + """ + super().__init__() + self._ctx = context + self._channel_name = channel_name + self._streaming = streaming + self._run_hook: ChannelRunHook = run_hook if run_hook is not None else _strip_options_hook + self._response_hook = response_hook + + async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: + """Publish a cancellation event for the in-flight task.""" + if context.context_id is None: + raise ValueError("Context ID must be provided in the RequestContext") + updater = TaskUpdater(event_queue, context.task_id or "", context.context_id) + await updater.cancel() + + async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: + """Route an A2A request through the host and publish task events.""" + if context.context_id is None: + raise ValueError("Context ID must be provided in the RequestContext") + if context.message is None: + raise ValueError("Message must be provided in the RequestContext") + + query = context.get_user_input() + task: Task | None = context.current_task + if not task: + task = cast(Task, new_task_from_user_message(context.message)) + await event_queue.enqueue_event(task) + + task_id: str = task.id + updater = TaskUpdater(event_queue, task_id, context.context_id) + await updater.submit() + + try: + await updater.start_work() + request = self._build_request(query, context, task_id) + if request.stream: + await self._run_stream(request, updater, protocol_request=context.message) + else: + await self._run(request, updater, protocol_request=context.message) + await updater.complete() + except CancelledError: + await updater.update_status(state=TaskState.TASK_STATE_CANCELED) + except Exception as exc: + logger.exception("A2AChannel encountered an error during execution.") + await updater.update_status( + state=TaskState.TASK_STATE_FAILED, + message=updater.new_agent_message([Part(text=str(exc))]), + ) + + def _build_request(self, query: Any, context: RequestContext, task_id: str) -> ChannelRequest: + """Build the channel-neutral request from the A2A request context.""" + context_id = cast(str, context.context_id) + return ChannelRequest( + channel=self._channel_name, + operation="message.create", + input=query if isinstance(query, str) else str(query), + session=ChannelSession(isolation_key=context_id), + stream=self._streaming, + identity=ChannelIdentity(channel=self._channel_name, native_id=context_id), + attributes={"task_id": task_id}, + ) + + async def _run(self, request: ChannelRequest, updater: TaskUpdater, *, protocol_request: Any) -> None: + """Non-streaming: run the target and publish the reply as task messages.""" + result = await self._ctx.run( + request, + run_hook=self._run_hook, + protocol_request=protocol_request, + response_hook=self._response_hook, + channel_name=self._channel_name, + ) + response: Any = result.result + messages: list[Any] = list(getattr(response, "messages", None) or []) + get_outputs = cast("Any", getattr(response, "get_outputs", None)) + if callable(get_outputs): + messages.extend(cast("list[Any]", get_outputs())) + for message in messages: + if getattr(message, "role", None) == "user": + continue + parts = _value_to_parts(message) + if parts: + await updater.update_status( + state=TaskState.TASK_STATE_WORKING, + message=updater.new_agent_message(parts=parts), + ) + + async def _run_stream(self, request: ChannelRequest, updater: TaskUpdater, *, protocol_request: Any) -> None: + """Streaming: publish incremental updates as task artifacts.""" + stream_artifact_id = f"{request.attributes.get('task_id', 'stream')}:stream" + appended = False + stream = await self._ctx.run_stream( + request, + run_hook=self._run_hook, + protocol_request=protocol_request, + response_hook=self._response_hook, + channel_name=self._channel_name, + ) + async for update in stream: + parts = _contents_to_parts(update.contents) + if not parts: + continue + await updater.add_artifact( + parts=parts, + artifact_id=stream_artifact_id, + append=True if appended else None, + ) + appended = True + final = await stream.get_final_response() + # A configured response_hook can rewrite/add the final assistant reply; + # only get_final_response() yields that shaped result. Project it so the + # hook's output reaches A2A clients (incremental deltas already cover the + # unhooked case, so skip this when no hook is configured to avoid dupes). + if self._response_hook is not None: + messages: list[Any] = list(getattr(final, "messages", None) or []) + get_outputs = cast("Any", getattr(final, "get_outputs", None)) + if callable(get_outputs): + messages.extend(cast("list[Any]", get_outputs())) + for message in messages: + if getattr(message, "role", None) == "user": + continue + parts = _value_to_parts(message) + if parts: + await updater.add_artifact( + parts=parts, artifact_id=stream_artifact_id, append=True if appended else None + ) + appended = True diff --git a/python/packages/hosting-a2a/pyproject.toml b/python/packages/hosting-a2a/pyproject.toml new file mode 100644 index 00000000000..075567854ab --- /dev/null +++ b/python/packages/hosting-a2a/pyproject.toml @@ -0,0 +1,86 @@ +[project] +name = "agent-framework-hosting-a2a" +description = "Agent-to-Agent (A2A) protocol channel for agent-framework-hosting." +authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}] +readme = "README.md" +requires-python = ">=3.10" +version = "1.0.0a260625" +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 = [ + "agent-framework-core>=1.2.0,<2", + "agent-framework-hosting>=1.0.0a260424,<2", + "a2a-sdk>=1.0.0,<2", + "starlette>=0.37", +] + +[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" + +[dependency-groups] +dev = [ + "uvicorn[standard]>=0.34.0", +] diff --git a/python/packages/hosting-a2a/tests/hosting_a2a/test_channel.py b/python/packages/hosting-a2a/tests/hosting_a2a/test_channel.py new file mode 100644 index 00000000000..62ea2438240 --- /dev/null +++ b/python/packages/hosting-a2a/tests/hosting_a2a/test_channel.py @@ -0,0 +1,394 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for :class:`A2AChannel` and :class:`HostAgentExecutor`.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Awaitable +from contextlib import asynccontextmanager +from dataclasses import dataclass, field +from typing import Any, cast + +import pytest +import uvicorn +from a2a.server.events import EventQueue +from a2a.types import AgentCard, AgentInterface, Message, Part, Role, Task, TaskState +from agent_framework import AgentResponse, Content +from agent_framework import Message as AFMessage +from agent_framework_a2a import A2AAgent +from agent_framework_hosting import AgentFrameworkHost, ChannelContribution, ChannelRequest, HostedRunResult +from starlette.types import ASGIApp + +from agent_framework_hosting_a2a import A2AChannel, HostAgentExecutor + +# --------------------------------------------------------------------------- # +# Fakes # +# --------------------------------------------------------------------------- # + + +@dataclass +class _FakeResp: + text: str + messages: list[Message] = field(default_factory=list) + + +@dataclass +class _FakeUpdate: + text: str + contents: list[Content] = field(default_factory=list) + message_id: str | None = None + + +class _FakeStream: + def __init__(self, chunks: list[str]) -> None: + self._chunks = chunks + self._final = _FakeResp( + text="".join(chunks), + messages=[Message(role=Role.ROLE_AGENT, parts=[Part(text="".join(chunks))])], + ) + + def __aiter__(self) -> AsyncIterator[_FakeUpdate]: + async def _gen() -> AsyncIterator[_FakeUpdate]: + for i, c in enumerate(self._chunks): + yield _FakeUpdate(text=c, contents=[Content.from_text(text=c)], message_id=f"m{i}") + + return _gen() + + async def get_final_response(self) -> _FakeResp: + return self._final + + +@dataclass +class _FakeTarget: + name: str = "Assistant" + description: str = "A helpful assistant." + + +class _FakeContext: + def __init__( + self, + *, + reply: str = "hello", + chunks: list[str] | None = None, + ) -> None: + self.target = _FakeTarget() + self._reply = reply + self._chunks = chunks or [reply] + self.requests: list[ChannelRequest] = [] + + async def run( + self, + request: ChannelRequest, + *, + run_hook: Any | None = None, + protocol_request: Any | None = None, + response_hook: Any | None = None, + channel_name: str | None = None, + ) -> HostedRunResult[Any]: + if run_hook is not None: + maybe_request = run_hook(request, target=self.target, protocol_request=protocol_request) + if isinstance(maybe_request, Awaitable): + request = await maybe_request + else: + request = maybe_request + self.requests.append(request) + msg = Message(role=Role.ROLE_AGENT, parts=[Part(text=self._reply)]) + result = HostedRunResult(_FakeResp(text=self._reply, messages=[msg])) + if response_hook is not None: + maybe_result = response_hook(result, request=request, channel_name=channel_name or request.channel) + if isinstance(maybe_result, Awaitable): + return await maybe_result + return maybe_result + return result + + async def run_stream( + self, + request: ChannelRequest, + *, + run_hook: Any | None = None, + protocol_request: Any | None = None, + stream_update_hook: Any | None = None, + response_hook: Any | None = None, + channel_name: str | None = None, + ) -> _FakeStream: + if run_hook is not None: + maybe_request = run_hook(request, target=self.target, protocol_request=protocol_request) + if isinstance(maybe_request, Awaitable): + request = await maybe_request + else: + request = maybe_request + self.requests.append(request) + return _FakeStream(self._chunks) + + +class _RecordingEventQueue(EventQueue): + def __init__(self) -> None: + super().__init__() + self.events: list[Any] = [] + + async def enqueue_event(self, event: Any) -> None: + self.events.append(event) + + +class _FakeRequestContext: + def __init__(self, *, context_id: str, text: str, current_task: Task | None = None) -> None: + self.context_id = context_id + self.task_id: str | None = None + self.message = Message( + message_id="msg-1", + context_id=context_id, + role=Role.ROLE_USER, + parts=[Part(text=text)], + ) + self.current_task = current_task + self._text = text + + def get_user_input(self) -> str: + return self._text + + +class _HostedAgent: + id = "hosted-agent" + name: str | None = "HostedAssistant" + description: str | None = "A hosted test assistant." + + async def run(self, messages: Any = None, *, stream: bool = False, **_kwargs: Any) -> AgentResponse[Any]: + text = messages.text if isinstance(messages, AFMessage) else str(messages) + return AgentResponse(messages=[AFMessage(role="assistant", contents=[Content.from_text(text=f"host: {text}")])]) + + def create_session(self, *, session_id: str | None = None) -> Any: + return {"session_id": session_id} + + def get_session(self, service_session_id: str, *, session_id: str | None = None) -> Any: + return {"service_session_id": service_session_id, "session_id": session_id} + + +@asynccontextmanager +async def _serve_app(app: ASGIApp, *, port: int) -> AsyncIterator[str]: + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning", lifespan="on") + server = uvicorn.Server(config) + task = asyncio.create_task(server.serve()) + try: + for _ in range(100): + if server.started: + break + await asyncio.sleep(0.01) + else: + raise RuntimeError("Test A2A server did not start") + yield f"http://127.0.0.1:{port}" + finally: + server.should_exit = True + await task + + +def _status_states(events: list[Any]) -> list[int]: + states: list[int] = [] + for event in events: + status = getattr(event, "status", None) + if status is not None and getattr(status, "state", None): + states.append(status.state) + return states + + +def _status_texts(events: list[Any]) -> list[str]: + texts: list[str] = [] + for event in events: + status = getattr(event, "status", None) + message = getattr(status, "message", None) + for part in cast("list[Any]", getattr(message, "parts", None) or []): + text = getattr(part, "text", None) + if isinstance(text, str): + texts.append(text) + return texts + + +# --------------------------------------------------------------------------- # +# A2AChannel tests # +# --------------------------------------------------------------------------- # + + +def test_default_name_and_root_path() -> None: + channel = A2AChannel() + assert channel.name == "a2a" + assert channel.path == "" + + +def test_build_agent_card_defaults_from_target() -> None: + channel = A2AChannel(url="https://example.com/") + card = channel._build_agent_card(cast(Any, _FakeContext())) + assert card.name == "Assistant" + assert card.description == "A helpful assistant." + assert card.capabilities.streaming is True + assert card.supported_interfaces[0].url == "https://example.com/" + + +def test_build_agent_card_accepts_supported_interfaces() -> None: + interfaces = [ + AgentInterface(url="https://example.com/jsonrpc", protocol_binding="JSONRPC"), + AgentInterface(url="https://example.com/grpc", protocol_binding="GRPC"), + ] + channel = A2AChannel(supported_interfaces=interfaces) + card = channel._build_agent_card(cast(Any, _FakeContext())) + assert card.supported_interfaces == interfaces + + +def test_build_agent_card_override_wins() -> None: + custom = AgentCard(name="Custom", description="custom card", version="9.9.9") + channel = A2AChannel(agent_card=custom) + card = channel._build_agent_card(cast(Any, _FakeContext())) + assert card.name == "Custom" + assert card.version == "9.9.9" + + +def test_contribute_returns_card_and_jsonrpc_routes() -> None: + channel = A2AChannel(url="https://example.com/") + contribution = channel.contribute(cast(Any, _FakeContext())) + assert isinstance(contribution, ChannelContribution) + paths = {getattr(r, "path", None) for r in contribution.routes} + assert "/.well-known/agent-card.json" in paths + assert any(p == "/" for p in paths) + + +# --------------------------------------------------------------------------- # +# HostAgentExecutor tests # +# --------------------------------------------------------------------------- # + + +async def test_execute_routes_through_host_and_completes() -> None: + ctx = _FakeContext(reply="hi back") + executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=False) + queue = _RecordingEventQueue() + request_context = _FakeRequestContext(context_id="conv-1", text="hello") + + await executor.execute(cast(Any, request_context), queue) + + # Routed through the host with the context id mapped onto the session. + assert len(ctx.requests) == 1 + request = ctx.requests[0] + assert request.channel == "a2a" + assert request.input == "hello" + assert request.session is not None + assert request.session.isolation_key == "conv-1" + assert request.identity is not None + assert request.identity.native_id == "conv-1" + # Task progressed to a completed state. + assert TaskState.TASK_STATE_COMPLETED in _status_states(queue.events) + + +async def test_execute_streaming_emits_artifacts() -> None: + ctx = _FakeContext(chunks=["foo", "bar"]) + executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=True) + queue = _RecordingEventQueue() + request_context = _FakeRequestContext(context_id="conv-2", text="hello") + + await executor.execute(cast(Any, request_context), queue) + + artifact_events = [e for e in queue.events if getattr(e, "artifact", None)] + assert artifact_events, "expected at least one artifact update event" + artifact_ids = {getattr(getattr(e, "artifact", None), "artifact_id", None) for e in artifact_events} + assert len(artifact_ids) == 1 + assert None not in artifact_ids + assert ctx.requests[0].stream is True + assert TaskState.TASK_STATE_COMPLETED in _status_states(queue.events) + + +async def test_execute_streaming_with_response_hook_emits_final_reply() -> None: + ctx = _FakeContext(chunks=["foo", "bar"]) + + def _hook(result: Any, **_: Any) -> Any: + return result + + executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=True, response_hook=_hook) + queue = _RecordingEventQueue() + request_context = _FakeRequestContext(context_id="conv-hook", text="hello") + + await executor.execute(cast(Any, request_context), queue) + + artifact_events = [e for e in queue.events if getattr(e, "artifact", None)] + # delta chunks + the projected final reply all land on a single stream artifact + assert len(artifact_events) >= 3 + assert TaskState.TASK_STATE_COMPLETED in _status_states(queue.events) + + +async def test_execute_requires_context_id() -> None: + ctx = _FakeContext() + executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a") + queue = _RecordingEventQueue() + request_context = _FakeRequestContext(context_id="x", text="hello") + cast(Any, request_context).context_id = None + + with pytest.raises(ValueError, match="Context ID"): + await executor.execute(cast(Any, request_context), queue) + + +async def test_a2a_agent_can_call_hosted_channel(unused_tcp_port: int) -> None: + host = AgentFrameworkHost(target=cast(Any, _HostedAgent()), channels=[A2AChannel(streaming=False)]) + + async with ( + _serve_app(host.app, port=unused_tcp_port) as base_url, + A2AAgent( + url=base_url, + timeout=5.0, + ) as agent, + ): + response = await agent.run("hello") + + assert response.messages[0].text == "host: hello" + + +async def test_execute_projects_workflow_outputs() -> None: + class _WorkflowResult: + value = None + + def get_outputs(self) -> list[AFMessage]: + return [AFMessage(role="assistant", contents=[Content.from_text("workflow output")])] + + class _WorkflowContext(_FakeContext): + async def run( + self, + request: ChannelRequest, + *, + run_hook: Any | None = None, + protocol_request: Any | None = None, + response_hook: Any | None = None, + channel_name: str | None = None, + ) -> HostedRunResult[Any]: + self.requests.append(request) + return HostedRunResult(_WorkflowResult()) + + ctx = _WorkflowContext() + executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", streaming=False) + queue = _RecordingEventQueue() + request_context = _FakeRequestContext(context_id="conv-workflow", text="hello") + + await executor.execute(cast(Any, request_context), queue) + + assert "workflow output" in _status_texts(queue.events) + + +def test_contents_to_parts_conversion() -> None: + from agent_framework_hosting_a2a._executor import _contents_to_parts + + contents = [ + Content.from_text(text="hello"), + Content.from_uri(uri="https://x/y.png", media_type="image/png"), + Content.from_data(data=b"AAAA", media_type="image/png"), + ] + parts = _contents_to_parts(contents) + assert parts[0].text == "hello" + assert parts[1].url == "https://x/y.png" + assert parts[2].raw == b"AAAA" + + +async def test_default_hook_strips_options_when_no_run_hook_supplied() -> None: + """When no run_hook is provided, the default hook strips all options.""" + ctx = _FakeContext(reply="ok") + executor = HostAgentExecutor(cast(Any, ctx), channel_name="a2a", run_hook=None) + queue = _RecordingEventQueue() + request_context = _FakeRequestContext(context_id="ctx-default-hook", text="hello") + + await executor.execute(cast(Any, request_context), queue) + + assert len(ctx.requests) == 1 + assert ctx.requests[0].options is None diff --git a/python/pyproject.toml b/python/pyproject.toml index aa204d33d21..4516aff257f 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-hosting-telegram = { workspace = true } agent-framework-hyperlight = { workspace = true } diff --git a/python/samples/04-hosting/af-hosting/README.md b/python/samples/04-hosting/af-hosting/README.md index c21ebbecd5e..9a524b2bf55 100644 --- a/python/samples/04-hosting/af-hosting/README.md +++ b/python/samples/04-hosting/af-hosting/README.md @@ -14,6 +14,7 @@ its own package. This first sample set includes | [`local_responses_workflow/`](./local_responses_workflow) | A 4-step `Workflow` (typed `SloganBrief` intake → writer → legal → formatter) hosted behind the Responses channel via a `run_hook` that parses inbound text/JSON into the workflow's typed input. The host writes per-conversation checkpoints via `checkpoint_location=…`. Demonstrates workflow targets + structured input adaptation + resume-across-turns. Includes a `call_server.rest` file with REST examples. | **Local only.** | | [`local_telegram/`](./local_telegram) | Telegram bot with `@tool`, `FileHistoryProvider`, `run_hook`, and slash commands (`/new`, `/whoami`, `/weather`). Pure Telegram — no HTTP endpoint. | **Local only.** Start here to learn the Telegram channel. | | [`local_multi_channel/`](./local_multi_channel) | Same agent behind two channels at once: `ResponsesChannel` + `TelegramChannel`. Shared `FileHistoryProvider` enables cross-channel session resumption (resume a Telegram chat from the Responses endpoint by passing the Telegram isolation key as `previous_response_id`). | **Local only.** | +| [`local_a2a/`](./local_a2a) | A `WeatherAgent` served over the [A2A protocol](https://a2a-protocol.org/latest/) with `A2AChannel`. Includes a `call_client.py` that uses `A2AAgent` to discover the card and call the hosted agent (both non-streaming and streaming). | **Local only.** | Each sample is fully self-contained — its own `pyproject.toml`, `uv.lock`, server `app.py`, calling script(s), and `storage/` directory. Every diff --git a/python/samples/04-hosting/af-hosting/local_a2a/README.md b/python/samples/04-hosting/af-hosting/local_a2a/README.md new file mode 100644 index 00000000000..3b234a46f49 --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_a2a/README.md @@ -0,0 +1,49 @@ +# local_a2a — WeatherAgent over A2A + +This sample hosts a `WeatherAgent` using `AgentFrameworkHost` with `A2AChannel` and then +calls it from another process using `A2AAgent`. + +## What this demonstrates + +- Hosting an agent over the [A2A protocol](https://a2a-protocol.org/latest/) with + `agent-framework-hosting-a2a`. +- A `run_hook` that strips caller-supplied generation options so the host + controls model selection. +- `FileHistoryProvider` for cross-restart session continuity. +- Both non-streaming and streaming calls from a client using `A2AAgent`. + +## Prerequisites + +- Azure AI Foundry project endpoint and model name. +- `az login` (uses `DefaultAzureCredential`). + +## Running + +### 1. Start the server + +```bash +uv sync +az login +export FOUNDRY_PROJECT_ENDPOINT=https://.services.ai.azure.com +export FOUNDRY_MODEL=gpt-4o +uv run python app.py +``` + +The agent is now reachable at `http://localhost:8000/a2a`. + +### 2. Run the client + +In a second terminal: + +```bash +uv run python call_client.py +``` + +The client resolves the hosted agent's A2A card, sends a weather question with +non-streaming, then sends another with streaming. + +### Production-style multi-worker start + +```bash +uv run hypercorn app:app --bind 0.0.0.0:8000 +``` diff --git a/python/samples/04-hosting/af-hosting/local_a2a/app.py b/python/samples/04-hosting/af-hosting/local_a2a/app.py new file mode 100644 index 00000000000..9507927042a --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_a2a/app.py @@ -0,0 +1,110 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""A2A channel hosting sample. + +Exposes a single ``WeatherAgent`` over the Agent-to-Agent (A2A) protocol. +Any A2A-compatible client — another agent or an A2A SDK consumer — can call +the hosted agent over JSONRPC at ``/a2a``. + +What this sample shows: + +- ``A2AChannel`` serving a ``WeatherAgent`` at ``/a2a``. +- A ``run_hook`` that strips all caller-supplied options so the host owns model + selection (the same security seam as ``ResponsesChannel``). +- ``FileHistoryProvider`` for per-session history persisted across restarts. + +Required env: ``FOUNDRY_PROJECT_ENDPOINT``, ``FOUNDRY_MODEL``. +Auth uses ``DefaultAzureCredential``. + +Run +--- +``app`` is a module-level Starlette ASGI app:: + + uv sync + az login + export FOUNDRY_PROJECT_ENDPOINT=https://.services.ai.azure.com + export FOUNDRY_MODEL=gpt-4o + uv run python app.py + +Multi-process via Hypercorn:: + + uv run hypercorn app:app --bind 0.0.0.0:8000 +""" + +from __future__ import annotations + +import os +from dataclasses import replace +from pathlib import Path +from typing import Annotated + +from agent_framework import Agent, FileHistoryProvider, tool +from agent_framework_foundry import FoundryChatClient +from agent_framework_hosting import AgentFrameworkHost, ChannelRequest +from agent_framework_hosting_a2a import A2AChannel +from azure.identity.aio import DefaultAzureCredential + +# import logging +# logging.basicConfig(level=logging.DEBUG) + +SESSIONS_DIR = Path(__file__).resolve().parent / "storage" / "sessions" +SESSIONS_DIR.mkdir(parents=True, exist_ok=True) + + +# --------------------------------------------------------------------------- # +# Tool +# --------------------------------------------------------------------------- # + + +@tool(approval_mode="never_require") +def lookup_weather( + location: Annotated[str, "The city to look up weather for."], +) -> str: + """Return a deterministic weather report for a city.""" + high_temp = 5 + (sum(location.encode("utf-8")) % 21) + reports = { + "Seattle": f"Seattle is rainy with a high of {high_temp}°C.", + "Amsterdam": f"Amsterdam is cloudy with a high of {high_temp}°C.", + "Tokyo": f"Tokyo is clear with a high of {high_temp}°C.", + } + return reports.get(location, f"{location} is sunny with a high of {high_temp}°C.") + + +# --------------------------------------------------------------------------- # +# Run hook +# --------------------------------------------------------------------------- # + + +def run_hook(request: ChannelRequest, **_: object) -> ChannelRequest: + """Strip all caller-supplied options; the host owns model selection.""" + return replace(request, options=None) + + +# --------------------------------------------------------------------------- # +# Agent +# --------------------------------------------------------------------------- # + + +def _build_app() -> AgentFrameworkHost: + credential = DefaultAzureCredential() + agent = Agent( + client=FoundryChatClient(credential=credential), + name="WeatherAgent", + instructions=( + "You are a friendly weather assistant. Use the lookup_weather tool " + "for any weather question and answer in one short sentence." + ), + tools=[lookup_weather], + context_providers=[FileHistoryProvider(SESSIONS_DIR)], + ) + return AgentFrameworkHost( + target=agent, + channels=[A2AChannel(run_hook=run_hook, streaming=True)], + ) + + +app = _build_app().app + + +if __name__ == "__main__": + _build_app().serve(host="0.0.0.0", port=int(os.environ.get("PORT", "8000"))) diff --git a/python/samples/04-hosting/af-hosting/local_a2a/call_client.py b/python/samples/04-hosting/af-hosting/local_a2a/call_client.py new file mode 100644 index 00000000000..4cf789818c9 --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_a2a/call_client.py @@ -0,0 +1,49 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""A2A client script for the local_a2a sample. + +Connects to a running ``local_a2a`` server using the A2A protocol, discovers the +hosted agent's capabilities, and sends a weather question. + +Usage:: + + # 1. Start the server in another terminal: + uv run python app.py + + # 2. Run this client: + uv run python call_client.py +""" + +from __future__ import annotations + +import asyncio + +from agent_framework_a2a import A2AAgent + + +async def main() -> None: + """Discover and call the hosted A2A agent.""" + base_url = "http://127.0.0.1:8000" + + async with A2AAgent(url=base_url) as agent: + print(f"Connected to A2A agent at {base_url}") + + # Non-streaming request + print("\n--- Non-streaming ---") + response = await agent.run("What is the weather in Seattle?") + print(response.text) + + # Streaming request — print chunks as they arrive + print("\n--- Streaming ---") + stream = agent.run("What is the weather in Tokyo?", stream=True) + async for update in stream: + for content in update.contents: + if content.text: + print(content.text, end="", flush=True) + print() + final = await stream.get_final_response() + print(f"\nFinal response: {final.text}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/samples/04-hosting/af-hosting/local_a2a/pyproject.toml b/python/samples/04-hosting/af-hosting/local_a2a/pyproject.toml new file mode 100644 index 00000000000..86da4460e06 --- /dev/null +++ b/python/samples/04-hosting/af-hosting/local_a2a/pyproject.toml @@ -0,0 +1,22 @@ +[project] +name = "agent-framework-hosting-sample-local-a2a" +version = "0.0.1" +description = "A2A channel hosting sample: WeatherAgent served over A2A, consumed by a client using A2AAgent." +requires-python = ">=3.10" +dependencies = [ + "agent-framework-foundry", + "agent-framework-hosting", + "agent-framework-hosting-a2a", + "agent-framework-a2a", + "azure-identity", + "hypercorn>=0.17", + "uvicorn[standard]>=0.34", +] + +[tool.uv] +package = false + +[tool.uv.sources] +agent-framework-hosting = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting" } +agent-framework-hosting-a2a = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/hosting-a2a" } +agent-framework-a2a = { git = "https://github.com/microsoft/agent-framework.git", branch = "main", subdirectory = "python/packages/a2a" } diff --git a/python/uv.lock b/python/uv.lock index 9e16f664933..d140c2b8be9 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -48,6 +48,7 @@ members = [ "agent-framework-gemini", "agent-framework-github-copilot", "agent-framework-hosting", + "agent-framework-hosting-a2a", "agent-framework-hosting-responses", "agent-framework-hosting-telegram", "agent-framework-hyperlight", @@ -665,6 +666,33 @@ provides-extras = ["serve", "disk"] [package.metadata.requires-dev] dev = [{ name = "httpx", specifier = ">=0.28.1" }] +[[package]] +name = "agent-framework-hosting-a2a" +version = "1.0.0a260625" +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'" }, + { name = "starlette", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" }, +] + +[package.dev-dependencies] +dev = [ + { name = "uvicorn", extra = ["standard"], 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" }, + { name = "starlette", specifier = ">=0.37" }, +] + +[package.metadata.requires-dev] +dev = [{ name = "uvicorn", extras = ["standard"], specifier = ">=0.34.0" }] + [[package]] name = "agent-framework-hosting-responses" version = "1.0.0a260625"