From b8c390f50ddde6e61caf079d967db73d2d5596c3 Mon Sep 17 00:00:00 2001 From: Alessandro <155005371+3clyp50@users.noreply.github.com> Date: Fri, 12 Jun 2026 03:10:43 +0200 Subject: [PATCH] Enable A2A streaming capability by default Wrap FastA2A agent card generation so Agent Zero advertises streaming support on A2A endpoints by default. Add focused regression coverage for the agent-card capability rewrite and proxy wiring, and update the helper DOX profile for the new wrapper. --- helpers/fasta2a_server.py | 25 +++++++- helpers/fasta2a_server.py.dox.md | 11 +++- tests/test_fasta2a_server.py | 102 +++++++++++++++++++++++++++++++ 3 files changed, 134 insertions(+), 4 deletions(-) create mode 100644 tests/test_fasta2a_server.py diff --git a/helpers/fasta2a_server.py b/helpers/fasta2a_server.py index 89afade1c..dfac9da24 100644 --- a/helpers/fasta2a_server.py +++ b/helpers/fasta2a_server.py @@ -2,11 +2,13 @@ import asyncio import uuid import atexit +import json from typing import Any, List import contextlib import threading from helpers import settings, projects +from starlette.responses import Response as StarletteResponse from starlette.requests import Request # Local imports @@ -61,6 +63,27 @@ except ImportError: # pragma: no cover – library not installed _PRINTER = PrintStyle(italic=True, font_color="purple", padding=False) +def _enable_streaming_capability(agent_card_body: bytes) -> bytes: + """Return an agent-card JSON body with A2A streaming enabled.""" + agent_card = json.loads(agent_card_body) + capabilities = agent_card.get("capabilities") + if not isinstance(capabilities, dict): + capabilities = {} + agent_card["capabilities"] = capabilities + capabilities["streaming"] = True + return json.dumps(agent_card, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +class AgentZeroFastA2A(FastA2A): # type: ignore[misc] + """FastA2A app with Agent Zero defaults layered over library defaults.""" + + async def _agent_card_endpoint(self, request: Request) -> StarletteResponse: + response = await super()._agent_card_endpoint(request) + body = _enable_streaming_capability(response.body) + self._agent_card_json_schema = body + return StarletteResponse(content=body, media_type="application/json") + + class AgentZeroWorker(Worker): # type: ignore[misc] """Agent Zero implementation of FastA2A Worker.""" @@ -242,7 +265,7 @@ class DynamicA2AProxy: } # Create new FastA2A app with proper thread safety - new_app = FastA2A( # type: ignore + new_app = AgentZeroFastA2A( # type: ignore storage=storage, broker=broker, name="Agent Zero", diff --git a/helpers/fasta2a_server.py.dox.md b/helpers/fasta2a_server.py.dox.md index cd8e00755..1f2dcba78 100644 --- a/helpers/fasta2a_server.py.dox.md +++ b/helpers/fasta2a_server.py.dox.md @@ -11,6 +11,8 @@ - `fasta2a_server.py` owns the runtime implementation. - `fasta2a_server.py.dox.md` owns durable notes about responsibilities, contracts, side effects, and verification for that implementation. - Classes: +- `AgentZeroFastA2A` (`FastA2A`) + - `async _agent_card_endpoint(self, request: Request) -> StarletteResponse` - `AgentZeroWorker` (`Worker`) - `async run_task(self, params: Any) -> None` - `async cancel_task(self, params: Any) -> None` @@ -20,6 +22,7 @@ - `get_instance()` - `reconfigure(self, token: str)` - Top-level functions: +- `_enable_streaming_capability(agent_card_body: bytes) -> bytes`: Ensure the generated A2A agent card advertises streaming support. - `is_available()`: Check if FastA2A is available and properly configured. - `get_proxy()`: Get the FastA2A proxy instance. - Notable constants/configuration names: `_PRINTER`. @@ -28,12 +31,13 @@ - Helper modules own reusable framework APIs and must preserve public callers unless all callers, tests, and docs are updated together. - Update this file whenever public functions, classes, persistence behavior, path/security assumptions, side effects, or cross-module contracts change. +- Agent Zero wraps `FastA2A` with `AgentZeroFastA2A` so the generated agent card sets `capabilities.streaming` to `true` by default. - Observed side-effect areas: filesystem writes, filesystem deletion, settings/state persistence, secret handling, scheduler state. -- Imported dependency areas include: `agent`, `asyncio`, `atexit`, `contextlib`, `helpers`, `helpers.persist_chat`, `helpers.print_style`, `initialize`, `starlette.requests`, `threading`, `typing`, `uuid`. +- Imported dependency areas include: `agent`, `asyncio`, `atexit`, `contextlib`, `helpers`, `helpers.persist_chat`, `helpers.print_style`, `initialize`, `json`, `starlette.requests`, `starlette.responses`, `threading`, `typing`, `uuid`. ## Key Concepts -- Important called helpers/classes observed in the source: `PrintStyle`, `DynamicA2AProxy.get_instance`, `super.__init__`, `join`, `UserMessage`, `threading.Lock`, `atexit.register`, `self._configure`, `settings.get_settings`, `path.startswith`, `self._convert_message`, `initialize_agent`, `AgentContext`, `context.log.log`, `context.communicate`, `context.reset`, `AgentContext.remove`, `remove_chat`, `self.storage.update_task`, `self._register_shutdown`. +- Important called helpers/classes observed in the source: `PrintStyle`, `DynamicA2AProxy.get_instance`, `json.loads`, `json.dumps`, `super.__init__`, `super._agent_card_endpoint`, `join`, `UserMessage`, `threading.Lock`, `atexit.register`, `self._configure`, `settings.get_settings`, `path.startswith`, `self._convert_message`, `initialize_agent`, `AgentContext`, `context.log.log`, `context.communicate`, `context.reset`, `AgentContext.remove`, `remove_chat`, `self.storage.update_task`, `self._register_shutdown`. - Keep request/response, tool, or helper semantics documented here at the same time as source changes. ## Work Guidance @@ -45,7 +49,8 @@ ## Verification - Run targeted tests for changed helper behavior; run security regressions for auth, filesystem, WebSocket, tunnel, upload, or secret-handling helpers. -- No direct test reference was found by name search; choose the nearest behavioral test or perform a focused smoke check. +- Related tests: + - `tests/test_fasta2a_server.py` ## Child DOX Index diff --git a/tests/test_fasta2a_server.py b/tests/test_fasta2a_server.py new file mode 100644 index 000000000..a8c65442e --- /dev/null +++ b/tests/test_fasta2a_server.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import importlib +import json +import sys +import types +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +def _load_fasta2a_server(monkeypatch): + settings_stub = types.ModuleType("helpers.settings") + settings_stub.get_settings = lambda: { + "a2a_server_enabled": True, + "mcp_server_token": "test-token", + } + monkeypatch.setitem(sys.modules, "helpers.settings", settings_stub) + + projects_stub = types.ModuleType("helpers.projects") + projects_stub.activate_project = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "helpers.projects", projects_stub) + + print_style_stub = types.ModuleType("helpers.print_style") + + class _PrintStyle: + def __init__(self, *args, **kwargs): + pass + + def print(self, *args, **kwargs): + pass + + print_style_stub.PrintStyle = _PrintStyle + monkeypatch.setitem(sys.modules, "helpers.print_style", print_style_stub) + + starlette_stub = types.ModuleType("starlette") + starlette_responses_stub = types.ModuleType("starlette.responses") + + class _Response: + def __init__(self, content=b"", media_type=None, *args, **kwargs): + self.body = content if isinstance(content, bytes) else str(content).encode() + self.media_type = media_type + + starlette_responses_stub.Response = _Response + starlette_requests_stub = types.ModuleType("starlette.requests") + starlette_requests_stub.Request = object + monkeypatch.setitem(sys.modules, "starlette", starlette_stub) + monkeypatch.setitem(sys.modules, "starlette.responses", starlette_responses_stub) + monkeypatch.setitem(sys.modules, "starlette.requests", starlette_requests_stub) + + agent_stub = types.ModuleType("agent") + agent_stub.AgentContext = type( + "AgentContext", + (), + {"remove": staticmethod(lambda *args, **kwargs: None)}, + ) + agent_stub.UserMessage = lambda **kwargs: types.SimpleNamespace(**kwargs) + agent_stub.AgentContextType = types.SimpleNamespace(BACKGROUND="background") + monkeypatch.setitem(sys.modules, "agent", agent_stub) + + initialize_stub = types.ModuleType("initialize") + initialize_stub.initialize_agent = lambda: {} + monkeypatch.setitem(sys.modules, "initialize", initialize_stub) + + persist_chat_stub = types.ModuleType("helpers.persist_chat") + persist_chat_stub.remove_chat = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "helpers.persist_chat", persist_chat_stub) + + sys.modules.pop("helpers.fasta2a_server", None) + return importlib.import_module("helpers.fasta2a_server") + + +def test_a2a_agent_card_streaming_capability_is_enabled_by_default(monkeypatch): + module = _load_fasta2a_server(monkeypatch) + + updated = module._enable_streaming_capability( + b'{"name":"Agent Zero","capabilities":{"streaming":false,"pushNotifications":false}}' + ) + + agent_card = json.loads(updated) + assert agent_card["capabilities"]["streaming"] is True + assert agent_card["capabilities"]["pushNotifications"] is False + + +def test_a2a_agent_card_streaming_capability_creates_missing_block(monkeypatch): + module = _load_fasta2a_server(monkeypatch) + + updated = module._enable_streaming_capability(b'{"name":"Agent Zero"}') + + assert json.loads(updated)["capabilities"] == {"streaming": True} + + +def test_a2a_proxy_uses_streaming_enabled_fast_a2a_wrapper(monkeypatch): + module = _load_fasta2a_server(monkeypatch) + proxy = object.__new__(module.DynamicA2AProxy) + + proxy._configure() + + assert isinstance(proxy.app, module.AgentZeroFastA2A)