diff --git a/.plans/a2a-protocol.md b/.plans/a2a-protocol.md new file mode 100644 index 000000000000..d759659b2502 --- /dev/null +++ b/.plans/a2a-protocol.md @@ -0,0 +1,162 @@ +# A2A (Agent2Agent) Protocol Server for Hermes Agent + +## Motivation + +[A2A](https://a2a-protocol.org) is the open standard for **agent-to-agent** +interoperability: it lets one agent discover another, delegate a task, and +stream back status/artifacts over plain HTTP — regardless of framework, vendor, +or language. Where MCP connects an agent to _tools_, A2A connects an agent to +_other agents_. Exposing Hermes over A2A makes it a first-class participant in +multi-agent systems: any A2A-speaking orchestrator (LangGraph, CrewAI, Google +ADK, custom routers, the `a2a-inspector`) can call Hermes as a remote worker. + +Hermes already ships sibling protocol adapters — `acp_adapter/` (editor +integration over stdio) and `mcp_serve.py` (tools over MCP). A2A is the missing +third edge, implemented through the bundled platform plugin surface: +**Hermes as a callable agent for other agents.** + +## What It Enables + +``` +┌────────────────────┐ ┌──────────────────────┐ +│ A2A client / peer │ GET /.well-known/agent-card.json │ hermes-a2a │ +│ • LangGraph │ ────────────────────────────────► │ (A2A plugin) │ +│ • CrewAI │ │ │ +│ • Google ADK │ POST / message/send │ ┌────────────────┐ │ +│ • a2a-inspector │ ────────────────────────────────► │ │ HermesAgent │ │ +│ • another Hermes │ │ │ Executor │ │ +│ │ POST / message/stream (SSE) │ └───────┬────────┘ │ +│ │ ◄──────────────────────────────── │ │ │ +│ │ TaskStatusUpdate / Artifact │ run_conversation() │ +└────────────────────┘ │ AIAgent │ + └──────────────────────┘ +``` + +A user would: + +1. `pip install hermes-agent[a2a]` +2. `hermes-a2a --host 0.0.0.0 --port 9100` (or `python -m plugins.platforms.a2a`) +3. Point any A2A client at `http://localhost:9100` — it fetches the Agent Card, + then sends messages and receives streamed task updates. + +## Scope (this cut: working vertical slice) + +**In:** + +- Agent Card served at `/.well-known/agent-card.json` (A2A v0.3, JSON-RPC transport). +- `message/send` — synchronous request/response (returns a completed `Task`). +- `message/stream` — SSE streaming of `TaskStatusUpdateEvent` + `TaskArtifactUpdateEvent`. +- `tasks/get` / `tasks/cancel` — provided by the SDK's `DefaultRequestHandler` + + a bounded in-memory store; `cancel` wired into Hermes interruption with + monotonic terminal-state persistence. +- Conversation continuity: A2A `contextId` ↔ a persistent Hermes session + (one `AIAgent` + history per context). +- Live agent progress: Hermes tool-calls, reasoning, and streamed text mapped + to A2A working-status updates; the final answer delivered as an artifact. + +**Out (deferred, not designed away):** + +- Push-notification webhooks (`tasks/pushNotificationConfig/*`). +- Persistent (DB-backed) task store and `tasks/resubscribe`. +- Auth schemes on the card (served unauthenticated; document `0.0.0.0` risk). +- gRPC / HTTP+JSON transports (JSON-RPC only for the slice). +- Multimodal input parts (text-only in; the seam accepts more later). + +## Protocol ↔ Hermes mapping + +| A2A concept | Hermes equivalent | +| --------------------- | ---------------------------------------------------------------------------------------------- | +| Agent Card | Built from `hermes_cli.__version__` + a curated skill list (mirrors `acp_registry/agent.json`) | +| `contextId` | A Hermes session: one `AIAgent` instance + `conversation_history` | +| `taskId` | One `run_conversation()` turn within a context | +| `message/send` (text) | `agent.run_conversation(user_message=..., conversation_history=..., task_id=...)` | +| streamed text delta | `agent.stream_delta_callback` → `TaskUpdater.update_status(working, msg)` | +| tool start | `agent.tool_progress_callback` (`tool.started`) → working status + tool metadata | +| tool result / step | `agent.step_callback` → working status with result metadata | +| model reasoning | `agent.reasoning_callback` → working status (metadata `kind=reasoning`) | +| final response | `result["final_response"]` → `TaskUpdater.add_artifact(...)` + `complete()` | +| `tasks/cancel` | `session.cancel_event.set()` + `agent.interrupt()` | + +This reuses the **exact** callback seam that `acp_adapter/events.py` uses; the +only difference is the translation target (A2A `TaskUpdater` events instead of +ACP `session_update`s). + +## Architecture + +`AIAgent.run_conversation()` is **synchronous and blocking**, while the a2a-sdk +`AgentExecutor.execute()` is **async** and owns the request's event loop. So, +mirroring the ACP adapter, the agent turn runs in a dedicated bounded worker +pool and its callbacks marshal A2A events back onto the loop +with `asyncio.run_coroutine_threadsafe`. The SDK's `EventQueue` + +`DefaultRequestHandler` turn those events into the JSON-RPC response (or SSE +stream); `A2AStarletteApplication` serves the card and RPC endpoint over uvicorn. + +### Module layout (`plugins/platforms/a2a/`) + +| File | Responsibility | +| ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `adapter.py` | `BasePlatformAdapter` lifecycle + `ctx.register_platform()` integration | +| `card.py` | `build_agent_card(url)` → `AgentCard` (version, skills, capabilities) | +| `sessions.py` | `ContextSessionStore`: `contextId → HermesSession(agent, history, cancel_event)`; `agent_factory` injection for tests; real `AIAgent` build mirrors `acp_adapter.session._make_agent` | +| `events.py` | Callback factories: AIAgent callbacks → `TaskUpdater` events via a thread-safe scheduler (no heavy Hermes imports, so the adapter is unit-testable standalone) | +| `executor.py` | `HermesAgentExecutor(AgentExecutor)`: `execute()` (resolve session → new task → wire callbacks → run turn in thread → stream → artifact + complete) and `cancel()` | +| `entry.py` | CLI: load `~/.hermes/.env`, logging, args (`--host/--port/--check/--version`), build card+handler+app, `uvicorn.run` | +| `__main__.py` | `python -m plugins.platforms.a2a` | + +### Data flow (`message/stream`) + +``` +client ──POST message/stream──► DefaultRequestHandler ──► HermesAgentExecutor.execute() + │ │ new_task() → enqueue Task + │ │ TaskUpdater.start_work() + │ │ to_thread(agent.run_conversation) + │ │ ├─ stream_delta_cb → update_status(working, text) + │ ◄────────── SSE: TaskStatusUpdateEvent (working) ────────┤ ├─ tool_progress_cb → update_status(working, tool meta) + │ │ └─ step_cb → update_status(working, result meta) + │ │ add_artifact(final_response) + │ ◄────────── SSE: TaskArtifactUpdateEvent ────────────────┤ complete() + │ ◄────────── SSE: TaskStatusUpdateEvent (completed) ──────┘ +``` + +## Packaging + +Mirrors the ACP adapter exactly: + +- `[project.optional-dependencies]`: `a2a = ["a2a-sdk[http-server]==0.3.26"]` + (pydantic-based 0.3.x line — matches the broad A2A client ecosystem and + Hermes' pydantic idioms; pinned exact per repo policy; published 2026-04-09, + clears the 7-day cooldown). +- `[project.scripts]`: `hermes-a2a = "plugins.platforms.a2a.entry:main"`. +- Bundled discovery: `plugins/platforms/a2a/plugin.yaml` + + `ctx.register_platform(name="a2a", ...)`. +- `[all]`: add `hermes-agent[a2a]` (parity with `acp`; not lazy-installable). + +## Testing + +Unit/integration tests under `tests/a2a/`, runnable without model credentials by +injecting a `FakeAgent` (same pattern as `tests/acp_adapter`): + +- `test_card.py` — card has required fields and is served at the well-known URL + (Starlette `TestClient`). +- `test_executor.py` — a fake agent drives `execute()`; assert the emitted event + sequence is `Task → working → artifact(final_response) → completed`. +- `test_sessions.py` — same `contextId` reuses one agent/history; cancel sets the + event and calls `interrupt()`. +- `test_end_to_end_echo.py` — build the real Starlette app around an echo + executor and drive it in-process via `httpx.ASGITransport` with the A2A client, + proving the full JSON-RPC + SSE path with no network/LLM. + +## Why a2a-sdk 0.3.26 (not 1.1.0) + +The 1.x line is protobuf-first (`AgentCard`/`Message`/`Part` are proto messages, +verbose to construct, and the ASGI app builder moved). 0.3.26 is the pydantic +line the entire A2A tutorial/client/inspector ecosystem targets today, it reads +naturally alongside Hermes' pydantic code, and it still exposes +`A2AStarletteApplication` + helper builders. For a clean, interoperable slice +it's the better engineering choice; revisit 1.x when the ecosystem's clients move. + +## Non-goals / known gaps + +- Served unauthenticated by default — bind to `127.0.0.1` unless fronted by a + reverse proxy / auth layer. Documented in `entry.py --help` and the card. +- Bounded in-memory task store: tasks are lost on restart (acceptable for the slice). diff --git a/AGENTS.md b/AGENTS.md index 63247b2bf474..a4f2dc067473 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -263,6 +263,7 @@ hermes-agent/ │ └── src/ # entry.tsx, app.tsx, gatewayClient.ts + app/components/hooks/lib ├── tui_gateway/ # Python JSON-RPC backend for the TUI ├── acp_adapter/ # ACP server (VS Code / Zed / JetBrains integration) +├── plugins/platforms/a2a/ # A2A server plugin (agent-to-agent task delegation) ├── cron/ # Scheduler — jobs.py, scheduler.py ├── scripts/ # run_tests.sh, release.py, auxiliary scripts ├── website/ # Docusaurus docs site diff --git a/hermes_bootstrap.py b/hermes_bootstrap.py index ae23cc976296..bc9e8fbd06ca 100644 --- a/hermes_bootstrap.py +++ b/hermes_bootstrap.py @@ -14,8 +14,9 @@ This module fixes both on Windows *only* — POSIX is untouched. It should be imported at the very top of every Hermes entry point -(``hermes``, ``hermes-agent``, ``hermes-acp``, ``python -m gateway.run``, -``batch_runner.py``, ``cron/scheduler.py``) before any other imports +(``hermes``, ``hermes-agent``, ``hermes-acp``, ``hermes-a2a``, +``python -m gateway.run``, ``batch_runner.py``, ``cron/scheduler.py``) +before any other imports that might do file I/O or print to stdout. What this module does on Windows: diff --git a/hermes_cli/config.py b/hermes_cli/config.py index ac3b0aaf82fc..850219685483 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2918,6 +2918,44 @@ def _ensure_hermes_home_managed(home: Path): "force_ipv4": False, }, + # Agent2Agent protocol server. This is a bundled platform plugin, but its + # standalone console entry point reads the same section so gateway-managed + # and standalone launches share one behavioral configuration surface. + "a2a": { + "enabled": False, + "host": "127.0.0.1", + "port": 9100, + "public_url": None, + # Maximum number of blocking AIAgent turns serviced concurrently. + "max_concurrency": 16, + # In-memory context/session LRU cap. + "max_sessions": 512, + # Retained protocol tasks and per-task status-history bounds. + "max_tasks": 2048, + "max_task_history": 100, + # Peer-visible tool metadata: preview (bounded), none, or full. + "tool_io": "preview", + }, + + # Default tools for the bundled A2A platform plugin. Keeping this in the + # generic platform tool configuration makes `hermes tools` selections and + # agent.disabled_toolsets authoritative for remotely callable sessions. + "platform_toolsets": { + "a2a": [ + "web", + "terminal", + "file", + "vision", + "skills", + "browser", + "todo", + "memory", + "session_search", + "code_execution", + "delegation", + ], + }, + # Gateway settings — control how messaging platforms (Telegram, Discord, # Slack, etc.) deliver agent-produced files as native attachments. "gateway": { diff --git a/nix/checks.nix b/nix/checks.nix index 7f625ca4fe4b..abe170278d15 100644 --- a/nix/checks.nix +++ b/nix/checks.nix @@ -96,7 +96,7 @@ json.dump(sorted(leaf_paths(DEFAULT_CONFIG)), sys.stdout, indent=2) entry-points-sync = pkgs.runCommand "hermes-entry-points-sync" { } '' set -e echo "=== Checking entry points match pyproject.toml [project.scripts] ===" - for bin in hermes hermes-agent hermes-acp; do + for bin in hermes hermes-agent hermes-acp hermes-a2a; do test -x ${hermes-agent}/bin/$bin || (echo "FAIL: $bin binary missing from Nix package"; exit 1) echo "PASS: $bin present" done diff --git a/nix/hermes-agent.nix b/nix/hermes-agent.nix index 043d1e942021..121ca56f444c 100644 --- a/nix/hermes-agent.nix +++ b/nix/hermes-agent.nix @@ -211,6 +211,7 @@ stdenv.mkDerivation (finalAttrs: { "hermes" "hermes-agent" "hermes-acp" + "hermes-a2a" ] } diff --git a/packaging/homebrew/hermes-agent.rb b/packaging/homebrew/hermes-agent.rb index 7c00fc6acf8f..9473dde8fccf 100644 --- a/packaging/homebrew/hermes-agent.rb +++ b/packaging/homebrew/hermes-agent.rb @@ -26,7 +26,7 @@ def install pkgshare.install "skills", "optional-skills" - %w[hermes hermes-agent hermes-acp].each do |exe| + %w[hermes hermes-agent hermes-acp hermes-a2a].each do |exe| next unless (libexec/"bin"/exe).exist? (bin/exe).write_env_script( diff --git a/plugins/platforms/a2a/__init__.py b/plugins/platforms/a2a/__init__.py new file mode 100644 index 000000000000..a89e30adb036 --- /dev/null +++ b/plugins/platforms/a2a/__init__.py @@ -0,0 +1,26 @@ +"""A2A (Agent2Agent) protocol server for Hermes Agent. + +Exposes the Hermes ``AIAgent`` as an A2A-compliant remote agent so any +A2A-speaking client or peer agent can discover it (via the Agent Card) and +delegate tasks over JSON-RPC + SSE. Sibling to ``acp_adapter`` (editor +integration over stdio) and ``mcp_serve`` (tools over MCP). + +Run it with ``hermes-a2a`` or ``python -m plugins.platforms.a2a``. See +``.plans/a2a-protocol.md`` for the design. +""" + +from typing import Any + + +def register(ctx: Any) -> None: + """Load the platform adapter only when plugin registration runs. + + Keeping this import lazy lets the standalone entry point execute its + bootstrap and import-path hardening before gateway modules are imported. + """ + from .adapter import register as register_adapter + + register_adapter(ctx) + + +__all__ = ["register"] diff --git a/plugins/platforms/a2a/__main__.py b/plugins/platforms/a2a/__main__.py new file mode 100644 index 000000000000..f15e3f4fae65 --- /dev/null +++ b/plugins/platforms/a2a/__main__.py @@ -0,0 +1,5 @@ +"""Allow running the server as ``python -m plugins.platforms.a2a``.""" + +from .entry import main + +main() diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py new file mode 100644 index 000000000000..269df4a9843e --- /dev/null +++ b/plugins/platforms/a2a/adapter.py @@ -0,0 +1,212 @@ +"""Gateway lifecycle adapter for the bundled A2A protocol server.""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +from typing import Any + +from gateway.config import Platform +from gateway.platforms.base import ( + BasePlatformAdapter, + SendResult, + is_network_accessible, +) + +from .config import apply_yaml_config, settings_from_platform_config +from .entry import build_app + +logger = logging.getLogger(__name__) + + +class A2AAdapter(BasePlatformAdapter): + """Run the A2A ASGI server as a gateway-managed platform plugin.""" + + def __init__(self, config: Any): + super().__init__(config, Platform("a2a")) + self._server: Any = None + self._serve_task: asyncio.Task[None] | None = None + self._serve_error: str | None = None + self._stopping = False + self._fatal_notify_task: asyncio.Task[None] | None = None + + async def _notify_fatal_error_safely(self) -> None: + try: + await self._notify_fatal_error() + except Exception: + logger.exception("A2A fatal-error notification failed") + + def _consume_fatal_notification(self, task: asyncio.Task[None]) -> None: + if self._fatal_notify_task is task: + self._fatal_notify_task = None + try: + task.result() + except asyncio.CancelledError: + pass + except Exception: + logger.exception("A2A fatal-error notification task failed") + + async def _serve_embedded(self) -> None: + """Run Uvicorn without letting process-level exits escape the task.""" + try: + await self._server.serve() + except SystemExit as exc: + self._serve_error = f"Uvicorn exited during startup ({exc.code})" + except Exception as exc: # noqa: BLE001 - isolate server task failures + self._serve_error = f"{type(exc).__name__}: {exc}" + finally: + if self._running and not self._stopping: + message = self._serve_error or "A2A server stopped unexpectedly" + logger.error("%s", message) + self._set_fatal_error("a2a_server_stopped", message, retryable=True) + notify_task = asyncio.create_task( + self._notify_fatal_error_safely(), + name="hermes-a2a-fatal-notify", + ) + self._fatal_notify_task = notify_task + notify_task.add_done_callback(self._consume_fatal_notification) + + async def connect(self, *, is_reconnect: bool = False) -> bool: + del is_reconnect + if self._serve_task is not None and not self._serve_task.done(): + return True + if self._serve_task is not None: + await self.disconnect() + + import uvicorn + + settings = settings_from_platform_config(self.config) + if is_network_accessible(settings.host): + logger.warning( + "A2A is binding to a network-accessible host (%s) without " + "built-in authentication. Put it behind a trusted reverse " + "proxy or authentication layer.", + settings.host, + ) + app = build_app( + settings.host, + settings.port, + settings.public_url, + settings=settings, + ) + + class _EmbeddedServer(uvicorn.Server): + def capture_signals(self): + return contextlib.nullcontext() + + self._serve_error = None + self._stopping = False + self._server = _EmbeddedServer( + uvicorn.Config( + app, + host=settings.host, + port=settings.port, + log_level="info", + timeout_graceful_shutdown=10, + ) + ) + self._serve_task = asyncio.create_task( + self._serve_embedded(), name="hermes-a2a-server" + ) + + for _ in range(100): + if self._server.started: + self._running = True + logger.info( + "A2A server listening on http://%s:%d", + settings.host, + settings.port, + ) + return True + if self._serve_task.done(): + await self._serve_task + message = self._serve_error or "A2A server stopped during startup" + logger.error("%s", message) + self._set_fatal_error("a2a_startup_failed", message, retryable=True) + return False + await asyncio.sleep(0.05) + + logger.error("Timed out waiting for the A2A server to start") + self._set_fatal_error( + "a2a_startup_timeout", + "Timed out waiting for the A2A server to start", + retryable=True, + ) + await self.disconnect() + return False + + async def disconnect(self) -> None: + self._stopping = True + self._running = False + if self._server is not None: + self._server.should_exit = True + task = self._serve_task + if task is not None: + try: + await asyncio.wait_for(asyncio.shield(task), timeout=10) + except asyncio.TimeoutError: + if self._server is not None: + self._server.force_exit = True + task.cancel() + await asyncio.gather(task, return_exceptions=True) + self._serve_task = None + self._server = None + self._stopping = False + + async def send( + self, + chat_id: str, + content: str, + reply_to: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> SendResult: + del chat_id, content, reply_to, metadata + return SendResult( + success=False, + error="A2A responses are delivered through the protocol task lifecycle.", + ) + + async def get_chat_info(self, chat_id: str) -> dict[str, Any]: + """Represent an A2A context as a direct agent conversation.""" + return {"name": f"A2A context {chat_id}", "type": "dm", "chat_id": chat_id} + + +def check_requirements() -> bool: + """Return whether the optional A2A server dependencies are installed.""" + try: + import uvicorn # noqa: F401 + + import a2a # noqa: F401 + except ImportError: + return False + return True + + +def validate_config(config: Any) -> bool: + settings = settings_from_platform_config(config) + return bool(settings.host and 1 <= settings.port <= 65535) + + +def is_connected(config: Any) -> bool: + return bool(getattr(config, "enabled", False)) and validate_config(config) + + +def register(ctx: Any) -> None: + """Register A2A through the generic gateway platform plugin surface.""" + ctx.register_platform( + name="a2a", + label="A2A (Agent2Agent)", + adapter_factory=lambda cfg: A2AAdapter(cfg), + check_fn=check_requirements, + validate_config=validate_config, + is_connected=is_connected, + install_hint="Install the optional dependencies with: pip install -e '.[a2a]'", + apply_yaml_config_fn=apply_yaml_config, + emoji="🤖", + allow_update_command=False, + platform_hint=( + "You are serving a remote agent over the A2A protocol. Return a " + "self-contained task result and do not ask interactive questions." + ), + ) diff --git a/plugins/platforms/a2a/card.py b/plugins/platforms/a2a/card.py new file mode 100644 index 000000000000..be1556d319b4 --- /dev/null +++ b/plugins/platforms/a2a/card.py @@ -0,0 +1,105 @@ +"""Build the Hermes A2A Agent Card. + +The Agent Card is the discovery document an A2A client fetches from +``/.well-known/agent-card.json`` before sending any message. It mirrors the +intent of ``acp_registry/agent.json`` but follows the A2A schema. +""" + +from __future__ import annotations + +from a2a.types import ( + AgentCapabilities, + AgentCard, + AgentProvider, + AgentSkill, + TransportProtocol, +) + +_DESCRIPTION = ( + "Self-improving open-source AI agent by Nous Research, with persistent " + "memory, skills, and rich tool support (shell, filesystem, web search, " + "code editing). Exposed over the Agent2Agent protocol so other agents can " + "delegate coding and reasoning tasks to it." +) + + +def _hermes_version() -> str: + """Best-effort Hermes version; falls back to a sentinel when the package + metadata isn't importable (e.g. running the adapter in isolation).""" + try: + from hermes_cli import __version__ + + return str(__version__) + except Exception: + return "0.0.0+local" + + +def build_skills() -> list[AgentSkill]: + """The skills advertised on the card. Coarse-grained on purpose — Hermes is + a general agent, not a fixed-function service.""" + return [ + AgentSkill( + id="general-agent", + name="General coding & reasoning agent", + description=( + "Plans and executes multi-step tasks: writes and edits code, " + "runs shell commands, inspects and modifies files, and reasons " + "over the results to reach a goal." + ), + tags=["coding", "shell", "filesystem", "reasoning", "autonomous"], + examples=[ + "Refactor the auth module to use async and add tests.", + "Find why the build is failing and fix it.", + "Summarize what this repository does and how it's structured.", + ], + ), + AgentSkill( + id="research", + name="Web research & synthesis", + description=( + "Searches the web, reads sources, and synthesizes a concise, " + "cited answer to a question." + ), + tags=["research", "web-search", "summarization"], + examples=[ + "What changed in the latest release of ?", + "Compare these three approaches and recommend one.", + ], + ), + ] + + +def build_agent_card( + url: str, + *, + version: str | None = None, + streaming: bool = True, +) -> AgentCard: + """Construct the Hermes Agent Card. + + Args: + url: The externally reachable base URL of this A2A service (the + JSON-RPC endpoint is served at the root of it). + version: Override the advertised version; defaults to the Hermes version. + streaming: Whether to advertise SSE streaming support. + """ + return AgentCard( + name="Hermes Agent", + description=_DESCRIPTION, + url=url, + version=version or _hermes_version(), + protocol_version="0.3.0", + preferred_transport=TransportProtocol.jsonrpc, + provider=AgentProvider( + organization="Nous Research", + url="https://github.com/NousResearch/hermes-agent", + ), + capabilities=AgentCapabilities( + streaming=streaming, + push_notifications=False, + state_transition_history=False, + ), + default_input_modes=["text/plain"], + default_output_modes=["text/plain"], + skills=build_skills(), + ) diff --git a/plugins/platforms/a2a/config.py b/plugins/platforms/a2a/config.py new file mode 100644 index 000000000000..44de84e818da --- /dev/null +++ b/plugins/platforms/a2a/config.py @@ -0,0 +1,114 @@ +"""Configuration helpers for the bundled A2A platform plugin.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 9100 +DEFAULT_MAX_CONCURRENCY = 16 +DEFAULT_MAX_SESSIONS = 512 +DEFAULT_MAX_TASKS = 2048 +DEFAULT_MAX_TASK_HISTORY = 100 +DEFAULT_TOOL_IO = "preview" +TOOL_IO_MODES = frozenset({"preview", "none", "full"}) + + +def _positive_int(value: Any, default: int, *, maximum: int | None = None) -> int: + if isinstance(value, bool): + return default + try: + parsed = int(value) + except (TypeError, ValueError): + return default + if parsed <= 0 or (maximum is not None and parsed > maximum): + return default + return parsed + + +def _tool_io(value: Any) -> str: + mode = str(value or "").strip().lower() + return mode if mode in TOOL_IO_MODES else DEFAULT_TOOL_IO + + +@dataclass(frozen=True) +class A2ASettings: + """Validated behavioral settings from the ``a2a`` config.yaml section.""" + + enabled: bool = False + host: str = DEFAULT_HOST + port: int = DEFAULT_PORT + public_url: str | None = None + max_concurrency: int = DEFAULT_MAX_CONCURRENCY + max_sessions: int = DEFAULT_MAX_SESSIONS + max_tasks: int = DEFAULT_MAX_TASKS + max_task_history: int = DEFAULT_MAX_TASK_HISTORY + tool_io: str = DEFAULT_TOOL_IO + + @classmethod + def from_mapping(cls, values: Mapping[str, Any] | None) -> "A2ASettings": + data = values if isinstance(values, Mapping) else {} + host = str(data.get("host") or DEFAULT_HOST).strip() or DEFAULT_HOST + public_url = str(data.get("public_url") or "").strip() or None + max_concurrency = _positive_int( + data.get("max_concurrency"), DEFAULT_MAX_CONCURRENCY + ) + max_tasks = max( + max_concurrency, + _positive_int(data.get("max_tasks"), DEFAULT_MAX_TASKS), + ) + return cls( + enabled=bool(data.get("enabled", False)), + host=host, + port=_positive_int(data.get("port"), DEFAULT_PORT, maximum=65535), + public_url=public_url, + max_concurrency=max_concurrency, + max_sessions=_positive_int(data.get("max_sessions"), DEFAULT_MAX_SESSIONS), + max_tasks=max_tasks, + max_task_history=_positive_int( + data.get("max_task_history"), DEFAULT_MAX_TASK_HISTORY + ), + tool_io=_tool_io(data.get("tool_io")), + ) + + +def load_a2a_settings() -> A2ASettings: + """Load the merged ``a2a`` section from the active Hermes profile.""" + from hermes_cli.config import load_config + + config = load_config() + section = config.get("a2a") if isinstance(config, dict) else None + return A2ASettings.from_mapping(section) + + +def settings_from_platform_config(platform_config: Any) -> A2ASettings: + """Build settings from a gateway ``PlatformConfig`` instance.""" + extra = getattr(platform_config, "extra", None) + values = dict(extra) if isinstance(extra, Mapping) else {} + values["enabled"] = bool(getattr(platform_config, "enabled", False)) + return A2ASettings.from_mapping(values) + + +def apply_yaml_config( + _yaml_config: dict[str, Any], platform_config: dict[str, Any] +) -> dict[str, Any]: + """Seed A2A-specific ``PlatformConfig.extra`` fields from config.yaml.""" + if not isinstance(platform_config, dict): + return {} + extra = platform_config.get("extra") + seeded = dict(extra) if isinstance(extra, dict) else {} + for key in ( + "host", + "port", + "public_url", + "max_concurrency", + "max_sessions", + "max_tasks", + "max_task_history", + "tool_io", + ): + if key in platform_config: + seeded[key] = platform_config[key] + return seeded diff --git a/plugins/platforms/a2a/entry.py b/plugins/platforms/a2a/entry.py new file mode 100644 index 000000000000..fd9371d036b0 --- /dev/null +++ b/plugins/platforms/a2a/entry.py @@ -0,0 +1,227 @@ +"""CLI entry point for the hermes-agent A2A server. + +Usage:: + + hermes-a2a # serve on 127.0.0.1:9100 + hermes-a2a --host 0.0.0.0 --port 9100 + python -m plugins.platforms.a2a --check + +The Agent Card is served at ``/.well-known/agent-card.json`` and the JSON-RPC +endpoint at ``/``. +""" + +# IMPORTANT: hermes_bootstrap must be the very first import — it configures +# UTF-8 stdio on Windows. No-op on POSIX. See hermes_bootstrap.py for the +# full rationale (mirrors acp_adapter/entry.py). +try: + import hermes_bootstrap # noqa: F401 +except ModuleNotFoundError: + # Graceful fallback when hermes_bootstrap isn't registered in the venv + # yet (e.g. a half-finished ``hermes update``). UTF-8 stdio setup is then + # skipped on Windows; POSIX is unaffected. + pass +else: + hermes_bootstrap.harden_import_path() + +import argparse +import logging +import sys +from contextlib import asynccontextmanager +from dataclasses import replace +from pathlib import Path + +from gateway.platforms.base import is_network_accessible + +from .config import ( + DEFAULT_HOST, + DEFAULT_PORT, + A2ASettings, + load_a2a_settings, +) + + +def _setup_logging() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + logging.getLogger("httpx").setLevel(logging.WARNING) + logging.getLogger("httpcore").setLevel(logging.WARNING) + + +def _load_env() -> None: + """Load ``~/.hermes/.env`` so the agent picks up provider credentials.""" + try: + from hermes_cli.env_loader import load_hermes_dotenv + from hermes_constants import get_hermes_home + + load_hermes_dotenv(hermes_home=get_hermes_home()) + except Exception: + logging.getLogger(__name__).debug( + "Could not load ~/.hermes/.env; using system env", exc_info=True + ) + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="hermes-a2a", + description="Run Hermes Agent as an A2A (Agent2Agent) server.", + ) + parser.add_argument( + "--host", + default=None, + help=( + f"Bind host (default {DEFAULT_HOST}). Use 0.0.0.0 to expose on the " + "network — the endpoint is UNAUTHENTICATED, so put it behind a " + "reverse proxy or auth layer." + ), + ) + parser.add_argument( + "--port", + type=int, + default=None, + help=f"Bind port (default {DEFAULT_PORT}).", + ) + parser.add_argument( + "--public-url", + default=None, + help="Base URL advertised in the Agent Card (default http://:/).", + ) + parser.add_argument( + "--check", action="store_true", help="Verify A2A deps + card build, then exit." + ) + parser.add_argument( + "--version", action="store_true", help="Print Hermes version and exit." + ) + return parser.parse_args(argv) + + +def _default_service_url(host: str, port: int) -> str: + advertised_host = f"[{host}]" if ":" in host and not host.startswith("[") else host + return f"http://{advertised_host}:{port}/" + + +def build_app( + host: str, + port: int, + public_url: str | None = None, + *, + settings: A2ASettings | None = None, +): + """Build the A2A Starlette ASGI app backed by the real Hermes agent.""" + from a2a.server.apps import A2AStarletteApplication + from a2a.server.request_handlers import DefaultRequestHandler + + from .card import build_agent_card + from .executor import HermesAgentExecutor + from .sessions import ContextSessionStore + from .task_store import BoundedTaskStore + + resolved = settings or load_a2a_settings() + url = public_url or _default_service_url(host, port) + executor = HermesAgentExecutor( + ContextSessionStore(max_sessions=resolved.max_sessions), + max_concurrency=resolved.max_concurrency, + tool_io_mode=resolved.tool_io, + ) + handler = DefaultRequestHandler( + agent_executor=executor, + task_store=BoundedTaskStore( + max_tasks=resolved.max_tasks, + max_history_messages=resolved.max_task_history, + ), + ) + + @asynccontextmanager + async def lifespan(_app): + try: + yield + finally: + await executor.aclose() + + return A2AStarletteApplication( + agent_card=build_agent_card(url), + http_handler=handler, + ).build(lifespan=lifespan) + + +def _run_check() -> None: + import a2a # noqa: F401 + + from .card import build_agent_card + from .executor import HermesAgentExecutor # noqa: F401 + + card = build_agent_card("http://127.0.0.1:9100/") + assert card.name and card.skills, "Agent card is missing name/skills" + print("Hermes A2A check OK") + + +def main(argv: list[str] | None = None) -> None: + args = _parse_args(argv) + + if args.version: + from .card import _hermes_version + + print(_hermes_version()) + return + if args.check: + _run_check() + return + + _setup_logging() + _load_env() + logger = logging.getLogger(__name__) + + settings = load_a2a_settings() + if args.host is not None: + settings = replace(settings, host=args.host) + if args.port is not None: + if not 1 <= args.port <= 65535: + raise SystemExit("--port must be between 1 and 65535") + settings = replace(settings, port=args.port) + if args.public_url is not None: + settings = replace(settings, public_url=args.public_url) + + # Ensure the project root is importable so ``from run_agent import AIAgent`` + # works when launched as a console script. + project_root = str(Path(__file__).resolve().parents[3]) + if project_root not in sys.path: + sys.path.insert(0, project_root) + + # MCP tool discovery from config.yaml — run before serving so the agents + # spawned per A2A context expose the user's configured MCP tools (mirrors + # acp_adapter/entry.py). + try: + from tools.mcp_tool import discover_mcp_tools + + discover_mcp_tools() + except Exception: + logger.debug("MCP tool discovery failed at A2A startup", exc_info=True) + + import uvicorn + + app = build_app( + settings.host, + settings.port, + settings.public_url, + settings=settings, + ) + logger.info( + "Starting hermes-agent A2A server on http://%s:%d " + "(card: /.well-known/agent-card.json)", + settings.host, + settings.port, + ) + if is_network_accessible(settings.host): + logger.warning( + "Binding %s — the A2A endpoint is UNAUTHENTICATED. " + "Put it behind a reverse proxy or auth layer.", + settings.host, + ) + + uvicorn.run(app, host=settings.host, port=settings.port, log_level="info") + + +if __name__ == "__main__": + main() diff --git a/plugins/platforms/a2a/events.py b/plugins/platforms/a2a/events.py new file mode 100644 index 000000000000..37128db9733b --- /dev/null +++ b/plugins/platforms/a2a/events.py @@ -0,0 +1,212 @@ +"""Translate AIAgent callbacks into A2A ``TaskUpdater`` events. + +``AIAgent.run_conversation`` runs synchronously in a worker thread, but the A2A +event queue lives on the server's asyncio loop. These callback factories marshal +each agent event back onto that loop with ``run_coroutine_threadsafe`` and block +briefly on it — so updates preserve order relative to the agent's own progress, +and all working-status updates land before the final artifact. + +Best-effort: a failed status update is logged and swallowed, never aborting the +turn. This module intentionally imports nothing from Hermes, so the adapter is +unit-testable on its own. +""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +import json +import logging +from typing import Any, Callable + +from a2a.server.tasks import TaskUpdater +from a2a.types import Part, TaskState, TextPart + +logger = logging.getLogger(__name__) + +# How long a worker-thread callback waits for the loop to enqueue its event. +# Bounded (and modest) so a slow or disconnected SSE consumer can't throttle the +# agent loop to a crawl — each progress update is best-effort. Matches the ACP +# adapter's 5s ceiling. +_SCHEDULE_TIMEOUT = 5.0 +_RESULT_PREVIEW_LIMIT = 2000 + + +def _tool_call_metadata( + name: str | None, args: Any, tool_io_mode: str = "preview" +) -> dict[str, Any]: + """Metadata for a ``tool.started`` status update, honoring the I/O mode.""" + metadata: dict[str, Any] = {"hermes/kind": "tool-call", "hermes/tool": name} + if tool_io_mode == "none": + return metadata + metadata["hermes/args"] = ( + _json_safe(args) if tool_io_mode == "full" else _bounded(args) + ) + return metadata + + +def _tool_result_metadata( + name: str | None, result: Any, tool_io_mode: str = "preview" +) -> dict[str, Any]: + """Metadata for a completed-tool status update, honoring the I/O mode.""" + metadata: dict[str, Any] = {"hermes/kind": "tool-result", "hermes/tool": name} + if tool_io_mode == "none": + return metadata + metadata["hermes/result"] = ( + _json_safe(result) if tool_io_mode == "full" else _bounded(result) + ) + return metadata + + +def _schedule(loop: asyncio.AbstractEventLoop, coro: Any) -> None: + """Run *coro* on *loop* from a worker thread and wait for it (bounded).""" + try: + future = asyncio.run_coroutine_threadsafe(coro, loop) + except RuntimeError: + # Loop already closed — drop the update. + return + try: + future.result(timeout=_SCHEDULE_TIMEOUT) + except concurrent.futures.TimeoutError: + # Cancel so the orphaned coroutine can't run later and emit a status + # update after the task has already reached a terminal state. + future.cancel() + logger.debug("A2A status update timed out; cancelled") + except Exception: + logger.debug("A2A status update failed", exc_info=True) + + +def _emit_working( + updater: TaskUpdater, + loop: asyncio.AbstractEventLoop, + text: str, + metadata: dict[str, Any] | None = None, +) -> None: + message = ( + updater.new_agent_message([Part(root=TextPart(text=text))]) if text else None + ) + _schedule( + loop, + updater.update_status(TaskState.working, message=message, metadata=metadata), + ) + + +def make_stream_delta_cb( + updater: TaskUpdater, loop: asyncio.AbstractEventLoop +) -> Callable[[str], None]: + """Stream incremental agent text as working-status message chunks.""" + + def _cb(text: str) -> None: + if text: + _emit_working(updater, loop, text) + + return _cb + + +def make_reasoning_cb( + updater: TaskUpdater, loop: asyncio.AbstractEventLoop +) -> Callable[[str], None]: + """Surface provider/model reasoning, tagged so clients can render it apart.""" + + def _cb(text: str) -> None: + if text: + _emit_working(updater, loop, text, metadata={"hermes/kind": "reasoning"}) + + return _cb + + +def make_tool_progress_cb( + updater: TaskUpdater, + loop: asyncio.AbstractEventLoop, + *, + tool_io_mode: str = "preview", +) -> Callable[..., None]: + """Report tool-call starts as working-status updates with tool metadata. + + Matches AIAgent's signature: + ``tool_progress_callback(event_type, name, preview, args, **kwargs)``. + """ + + def _cb( + event_type: str | None = None, + name: str | None = None, + preview: str | None = None, + args: Any = None, + **_kwargs: Any, + ) -> None: + if event_type != "tool.started": + return + _emit_working( + updater, + loop, + f"⚙ {name}", + metadata=_tool_call_metadata(name, args, tool_io_mode), + ) + + return _cb + + +def make_step_cb( + updater: TaskUpdater, + loop: asyncio.AbstractEventLoop, + *, + tool_io_mode: str = "preview", +) -> Callable[..., None]: + """Report completed tool calls from AIAgent's ``step_callback`` payload. + + Signature: ``step_callback(api_call_count, prev_tools)`` where ``prev_tools`` + is a list of dicts describing the tools that ran in the previous step. + """ + + def _cb(api_call_count: int | None = None, prev_tools: Any = None) -> None: + if not isinstance(prev_tools, list): + return + for tool in prev_tools: + if not isinstance(tool, dict): + continue + name = tool.get("name") or tool.get("function_name") + if not name: + continue + result = tool.get("result") or tool.get("output") + _emit_working( + updater, + loop, + f"✓ {name}", + metadata=_tool_result_metadata(name, result, tool_io_mode), + ) + + return _cb + + +def _json_safe(value: Any) -> Any: + """Coerce *value* to something JSON-serializable for event metadata.""" + if value is None or isinstance(value, (str, int, float, bool)): + return value + try: + json.dumps(value) + return value + except (TypeError, ValueError): + return str(value) + + +def _truncate(value: Any, limit: int = _RESULT_PREVIEW_LIMIT) -> Any: + if not isinstance(value, str): + return value + return value if len(value) <= limit else value[:limit] + "…" + + +def _bounded(value: Any, limit: int = _RESULT_PREVIEW_LIMIT) -> Any: + """Bound a value for peer-facing metadata. + + Small structured values pass through unchanged; large strings or large + structures are rendered to a truncated string preview so a single tool call + can't blast megabytes (or a wall of secrets) at the peer. + """ + safe = _json_safe(value) + if isinstance(safe, str): + return _truncate(safe, limit) + try: + serialized = json.dumps(safe) + except (TypeError, ValueError): + return _truncate(str(safe), limit) + return safe if len(serialized) <= limit else _truncate(serialized, limit) diff --git a/plugins/platforms/a2a/executor.py b/plugins/platforms/a2a/executor.py new file mode 100644 index 000000000000..5439c5611bf4 --- /dev/null +++ b/plugins/platforms/a2a/executor.py @@ -0,0 +1,309 @@ +"""Hermes implementation of the A2A ``AgentExecutor``. + +This is the bridge between the a2a-sdk request lifecycle and Hermes' agent loop: +resolve the conversation context, run one ``AIAgent`` turn in a worker thread, +stream its progress as task-status updates, and deliver the final answer as an +artifact. +""" + +from __future__ import annotations + +import asyncio +import logging +import threading +from concurrent.futures import ThreadPoolExecutor +from typing import Any + +from a2a.server.agent_execution import AgentExecutor, RequestContext +from a2a.server.events import EventQueue +from a2a.server.tasks import TaskUpdater +from a2a.types import InternalError, InvalidParamsError, Part, TaskState, TextPart +from a2a.utils import new_task +from a2a.utils.errors import ServerError + +from .config import DEFAULT_TOOL_IO, load_a2a_settings +from .events import ( + make_reasoning_cb, + make_step_cb, + make_stream_delta_cb, + make_tool_progress_cb, +) +from .sessions import ContextSessionStore + +logger = logging.getLogger(__name__) + + +class HermesAgentExecutor(AgentExecutor): + """Runs Hermes turns in response to A2A ``message/send`` and ``message/stream``.""" + + def __init__( + self, + store: ContextSessionStore | None = None, + *, + max_concurrency: int | None = None, + tool_io_mode: str | None = None, + ): + settings = load_a2a_settings() + self._store = store or ContextSessionStore() + self._max_concurrency = max_concurrency or settings.max_concurrency + self._tool_io_mode = tool_io_mode or settings.tool_io or DEFAULT_TOOL_IO + # Dedicated, bounded pool so A2A turns neither saturate nor are starved + # by asyncio's shared default executor. Created lazily on first turn. + self._turn_pool: ThreadPoolExecutor | None = None + self._admission_lock = threading.Lock() + self._active_turns = 0 + self._closed = False + + def _pool(self) -> ThreadPoolExecutor: + if self._turn_pool is None: + self._turn_pool = ThreadPoolExecutor( + max_workers=self._max_concurrency, + thread_name_prefix="hermes-a2a-turn", + ) + return self._turn_pool + + def _submit_admitted_turn( + self, + context_id: str, + user_text: str, + task_id: str, + callbacks: dict[str, Any], + ): + """Lease a session and submit its turn atomically against shutdown. + + An admitted coroutine can await protocol event delivery before it is + ready to start a worker. Holding the admission lock across the final + closed check, session acquisition, and pool submission ensures + ``aclose()`` either sees and drains that worker or closes first and + prevents any post-close session/pool recreation. + """ + with self._admission_lock: + if self._closed: + raise ServerError( + error=InternalError(message="A2A server is stopping.") + ) + session = self._store.acquire(context_id) + try: + worker_future = self._pool().submit( + session.run_turn, + user_text, + task_id, + callbacks=callbacks, + ) + except Exception: + self._store.release(session) + raise + return session, worker_future + + async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: + # Reject empty/blank input before constructing a task. ``new_task`` + # itself raises on an empty TextPart, so guarding here avoids an + # unhandled error and returns a proper JSON-RPC error to the client. + user_text = context.get_user_input() + if not user_text.strip(): + raise ServerError( + error=InvalidParamsError(message="Message contains no text to act on.") + ) + + with self._admission_lock: + if self._closed: + raise ServerError( + error=InternalError(message="A2A server is stopping.") + ) + if self._active_turns >= self._max_concurrency: + raise ServerError( + error=InternalError( + message="A2A server is at turn capacity; retry later." + ) + ) + self._active_turns += 1 + + reservation = {"transferred": False} + try: + await self._execute_admitted( + context, + event_queue, + user_text, + reservation=reservation, + ) + finally: + if not reservation["transferred"]: + self._release_admission() + + async def _execute_admitted( + self, + context: RequestContext, + event_queue: EventQueue, + user_text: str, + *, + reservation: dict[str, bool], + ) -> None: + task = context.current_task + if task is None: + message = context.message + if message is None: + raise ServerError( + error=InvalidParamsError( + message="A2A request has no message to start a task from." + ) + ) + task = new_task(message) + await event_queue.enqueue_event(task) + + updater = TaskUpdater(event_queue, task.id, task.context_id) + await updater.start_work() + + loop = asyncio.get_running_loop() + + # AIAgent's callbacks are bound onto the (shared, per-context) agent + # *inside* run_turn under the session lock — never here on the shared + # instance — so two concurrent turns on the same context can't overwrite + # each other's TaskUpdater. ``thinking_callback`` is silenced (local + # "kawaii" status spam, not for A2A). + callbacks = { + "stream_delta_callback": make_stream_delta_cb(updater, loop), + "reasoning_callback": make_reasoning_cb(updater, loop), + "tool_progress_callback": make_tool_progress_cb( + updater, loop, tool_io_mode=self._tool_io_mode + ), + "step_callback": make_step_cb( + updater, loop, tool_io_mode=self._tool_io_mode + ), + "thinking_callback": None, + } + + try: + session, worker_future = self._submit_admitted_turn( + task.context_id, + user_text, + task.id, + callbacks, + ) + reservation["transferred"] = True + + def release_worker_resources(_future): + self._store.release(session) + self._release_admission() + + worker_future.add_done_callback(release_worker_resources) + result = await asyncio.wrap_future(worker_future) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 — surface any agent error as task failure + logger.exception("Hermes turn failed for task %s", task.id) + await self._safe_terminal( + updater, + updater.failed( + updater.new_agent_message([ + Part(root=TextPart(text=self._peer_error_message(exc))) + ]) + ), + ) + return + + # Map AIAgent's turn outcome onto the A2A task state. ``run_conversation`` + # catches its own failures and *returns* a dict (failed/interrupted/error) + # rather than raising, so we must inspect it — otherwise a failed or + # truncated turn would be reported to the peer agent as a successful, + # empty completion. + if session.cancel_event.is_set() or result.get("interrupted"): + await self._safe_terminal( + updater, updater.update_status(TaskState.canceled, final=True) + ) + return + + final_text = str(result.get("final_response") or "").strip() + err = str(result.get("error") or "").strip() + + # A turn is unsuccessful when it set an explicit ``failed`` flag, OR + # carries an ``error`` string, OR produced no usable text. Several + # degraded/partial early-returns in run_conversation (thinking-budget + # exhausted, response truncation) set ``error`` + a human-readable + # ``final_response`` but never reach ``finalize_turn``, so the dict has + # no ``failed`` key — without the ``err`` check those reach the peer as a + # "completed" task whose artifact is really an error notice. + if result.get("failed") or err or not final_text: + if final_text and err: + text = f"{final_text}\n\n(error: {err})" + elif final_text: + text = final_text + elif err: + text = f"Agent turn failed: {err}" + else: + text = "Agent produced no response." + await self._safe_terminal( + updater, + updater.failed( + updater.new_agent_message([Part(root=TextPart(text=text))]) + ), + ) + return + + await updater.add_artifact( + [Part(root=TextPart(text=final_text))], name="response", last_chunk=True + ) + await self._safe_terminal(updater, updater.complete()) + + def _release_admission(self) -> None: + with self._admission_lock: + self._active_turns = max(0, self._active_turns - 1) + + async def aclose(self) -> None: + """Stop accepting work, interrupt turns, and release agent resources.""" + with self._admission_lock: + if self._closed: + return + self._closed = True + self._store.cancel_all() + pool = self._turn_pool + if pool is not None: + await asyncio.to_thread(pool.shutdown, wait=True, cancel_futures=True) + self._turn_pool = None + self._store.close() + + @staticmethod + def _peer_error_message(exc: Exception) -> str: + """A peer-safe failure message. + + The full traceback is logged server-side; the remote peer gets only the + exception *type*, never the raw ``str(exc)`` — which can carry file + paths, prompts, or other internal detail to an untrusted caller. + """ + return ( + f"The agent encountered an internal error ({type(exc).__name__}) " + "while processing the task." + ) + + @staticmethod + async def _safe_terminal(updater: TaskUpdater, coro) -> None: + """Await a terminal TaskUpdater call, tolerating an already-terminal task. + + ``cancel()`` may drive the same task to a terminal state from a separate + TaskUpdater; swallow the resulting ``RuntimeError`` instead of crashing + the turn. + """ + try: + await coro + except RuntimeError: + logger.debug("Task %s already terminal; skipping update", updater.task_id) + + async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: + task = context.current_task + # Prefer the task's context_id (the key the store is indexed by in + # execute); fall back to the request context_id. + context_id = ( + task.context_id if task is not None else None + ) or context.context_id + if context_id: + session = self._store.get(context_id) + if session is not None: + # Scope the cancel to this specific task so it can't interrupt a + # different concurrent turn on the same context. + session.cancel(task.id if task is not None else None) + + if task is not None: + updater = TaskUpdater(event_queue, task.id, task.context_id) + try: + await updater.update_status(TaskState.canceled, final=True) + except Exception: + logger.debug("cancel status update failed", exc_info=True) diff --git a/plugins/platforms/a2a/plugin.yaml b/plugins/platforms/a2a/plugin.yaml new file mode 100644 index 000000000000..77dd3d908a06 --- /dev/null +++ b/plugins/platforms/a2a/plugin.yaml @@ -0,0 +1,8 @@ +name: a2a-platform +label: A2A (Agent2Agent) +kind: platform +version: 1.0.0 +description: > + A2A protocol server for Hermes Agent. Exposes an Agent Card and accepts + agent tasks over JSON-RPC and server-sent events. +author: Yuri Gui (@yugui923) and Hermes Agent diff --git a/plugins/platforms/a2a/sessions.py b/plugins/platforms/a2a/sessions.py new file mode 100644 index 000000000000..ab0819780109 --- /dev/null +++ b/plugins/platforms/a2a/sessions.py @@ -0,0 +1,374 @@ +"""Map an A2A ``contextId`` to a persistent Hermes ``AIAgent`` session. + +An A2A *context* is a conversation thread; a *task* is one turn within it. We +hold one ``AIAgent`` (plus its rolling history) per context so follow-up +messages on the same ``contextId`` continue the same conversation — the same +relationship ``acp_adapter.session.SessionManager`` maintains for ACP sessions. + +Sessions are in-memory only for this cut (no DB persistence; see +``.plans/a2a-protocol.md``). The real ``AIAgent`` build mirrors +``acp_adapter.session._make_agent``; tests inject a fake via ``agent_factory``. +""" + +from __future__ import annotations + +import logging +import threading +from collections import OrderedDict +from dataclasses import dataclass, field +from typing import Any, Callable + +logger = logging.getLogger(__name__) + +# Cap on concurrently-retained contexts so a long-lived server doesn't grow +# unbounded (one AIAgent + history per context). Least-recently-used contexts +# are evicted past this; a follow-up message on an evicted context simply +# starts a fresh session. +DEFAULT_MAX_SESSIONS = 512 + + +# Upper bound on remembered "cancel arrived before the turn started" task ids, +# so a peer spamming cancels for never-seen tasks can't grow the set unbounded. +_MAX_PENDING_CANCELS = 1024 + + +@dataclass +class HermesSession: + """One A2A context: a Hermes agent, its history, and a cancel signal.""" + + context_id: str + agent: Any # AIAgent instance (or a test fake) + history: list[dict[str, Any]] = field(default_factory=list) + cancel_event: threading.Event = field(default_factory=threading.Event) + lock: threading.Lock = field(default_factory=threading.Lock) + # Guards _active_task_id / _cancelled_task_ids. Distinct from ``lock`` (which + # serializes whole turns) so cancel() can run without waiting for the turn. + _state_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + _active_task_id: str | None = field(default=None, repr=False) + _cancelled_task_ids: set[str] = field(default_factory=set, repr=False) + _leases: int = field(default=0, repr=False) + _closing: bool = field(default=False, repr=False) + + def run_turn( + self, + user_text: str, + task_id: str, + *, + callbacks: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Run one blocking agent turn and fold the result into history. + + Called from a worker thread (the agent loop is synchronous). Serialized + per session by ``lock`` so two tasks on the same context can't race the + same history. + + ``callbacks`` (name -> callable|None) are bound onto the shared agent + *inside* the lock and cleared afterwards, so two concurrent turns on the + same context can never cross-wire their streams onto the other's task. + """ + with self.lock: + with self._state_lock: + if self._closing: + return {"final_response": None, "interrupted": True} + # A cancel that raced ahead of this turn starting: skip it. + if task_id in self._cancelled_task_ids: + self._cancelled_task_ids.discard(task_id) + return {"final_response": None, "interrupted": True} + # Clean slate: drop any stale cancel/interrupt left by a prior + # task before we mark this one active, so a cancel targeting + # *this* task (arriving after this point) is not erased. + self.cancel_event.clear() + self._clear_agent_interrupt() + self._active_task_id = task_id + + agent = self.agent + terminal_tool = None + previous_approval_cb = None + approval_guard_error: Exception | None = None + + def _deny_remote_approval(command: str, description: str, **_kwargs: Any): + logger.warning( + "A2A auto-denied dangerous command requiring approval: %s (%s)", + command, + description, + ) + return "deny" + + try: + from tools import terminal_tool + + previous_approval_cb = terminal_tool._get_approval_callback() + terminal_tool.set_approval_callback(_deny_remote_approval) + except Exception: + logger.exception("Could not install A2A approval guard") + approval_guard_error = RuntimeError( + "A2A approval guard could not be installed" + ) + if callbacks: + for name, cb in callbacks.items(): + setattr(agent, name, cb) + try: + if approval_guard_error is not None: + raise approval_guard_error + result = agent.run_conversation( + user_message=user_text, + conversation_history=self.history, + # Hermes resources are scoped to the stable context/session, + # while task_id remains A2A protocol bookkeeping above. + task_id=self.context_id, + persist_user_message=user_text, + ) + if isinstance(result, dict): + messages = result.get("messages") + if isinstance(messages, list): + self.history = messages + return result + return {"final_response": str(result)} + finally: + if terminal_tool is not None: + try: + terminal_tool.set_approval_callback(previous_approval_cb) + except Exception: + logger.exception("Could not restore approval callback") + if callbacks: + for name in callbacks: + setattr(agent, name, None) + with self._state_lock: + self._active_task_id = None + + def is_busy(self) -> bool: + """True while a turn is executing for this context.""" + with self._state_lock: + return self._active_task_id is not None + + def acquire_lease(self) -> None: + with self._state_lock: + self._leases += 1 + + def release_lease(self) -> None: + with self._state_lock: + self._leases = max(0, self._leases - 1) + + def is_in_use(self) -> bool: + with self._state_lock: + return self._active_task_id is not None or self._leases > 0 + + def begin_close(self) -> None: + """Prevent queued turns from starting and interrupt the active turn.""" + with self._state_lock: + self._closing = True + self.cancel_event.set() + self._interrupt_agent() + + def cancel(self, task_id: str | None = None) -> None: + """Cancel a turn on this context. + + ``task_id=None`` cancels whatever is currently running. A specific + ``task_id`` only interrupts the agent when it is the running turn — + otherwise it is recorded so the turn is skipped if it starts later + (covers a cancel that races ahead of ``run_turn``). This prevents one + task's cancel from killing a different concurrent turn on the same + context. + """ + with self._state_lock: + active = self._active_task_id + if task_id is not None and task_id != active: + if len(self._cancelled_task_ids) < _MAX_PENDING_CANCELS: + self._cancelled_task_ids.add(task_id) + return + self.cancel_event.set() + self._interrupt_agent() + + def _interrupt_agent(self) -> None: + interrupt = getattr(self.agent, "interrupt", None) + if callable(interrupt): + try: + interrupt() + except Exception: + pass + + def _clear_agent_interrupt(self) -> None: + clear = getattr(self.agent, "clear_interrupt", None) + if callable(clear): + try: + clear() + except Exception: + pass + + def close(self) -> None: + """Release resources owned by an idle session agent.""" + self.begin_close() + close = getattr(self.agent, "close", None) + if callable(close): + try: + close() + except Exception: + logger.exception("Failed to close A2A session %s", self.context_id) + + +class ContextSessionStore: + """Thread-safe ``contextId -> HermesSession`` store with lazy agent creation.""" + + def __init__( + self, + agent_factory: Callable[[], Any] | None = None, + *, + cwd: str = ".", + max_sessions: int = DEFAULT_MAX_SESSIONS, + ): + self._agent_factory = agent_factory + self._cwd = cwd + self._max_sessions = max_sessions + # OrderedDict as an LRU: most-recently-used at the end. + self._sessions: OrderedDict[str, HermesSession] = OrderedDict() + self._lock = threading.Lock() + + def get(self, context_id: str) -> HermesSession | None: + with self._lock: + session = self._sessions.get(context_id) + if session is not None: + self._sessions.move_to_end(context_id) + return session + + def get_or_create(self, context_id: str) -> HermesSession: + evicted: list[HermesSession] = [] + with self._lock: + session = self._sessions.get(context_id) + if session is not None: + self._sessions.move_to_end(context_id) + return session + session = HermesSession( + context_id=context_id, + agent=self._make_agent(context_id), + ) + self._sessions[context_id] = session + evicted = self._evict_lru() + self._close_sessions(evicted) + return session + + def acquire(self, context_id: str) -> HermesSession: + """Atomically look up/create and lease a session against eviction.""" + evicted: list[HermesSession] + with self._lock: + session = self._sessions.get(context_id) + if session is None: + session = HermesSession( + context_id=context_id, + agent=self._make_agent(context_id), + ) + self._sessions[context_id] = session + else: + self._sessions.move_to_end(context_id) + session.acquire_lease() + evicted = self._evict_lru() + self._close_sessions(evicted) + return session + + def release(self, session: HermesSession) -> None: + with self._lock: + session.release_lease() + evicted = self._evict_lru() + self._close_sessions(evicted) + + def remove(self, context_id: str) -> bool: + with self._lock: + session = self._sessions.get(context_id) + if session is None or session.is_in_use(): + return False + del self._sessions[context_id] + session.close() + return True + + def _evict_lru(self) -> list[HermesSession]: + """Drop least-recently-used *idle* sessions past the configured cap. + + Sessions with a turn in flight are skipped — evicting one would orphan + the running worker thread and silently fork its history into a fresh, + empty agent on the next message. If every session over the cap is busy, + the store temporarily overshoots rather than corrupting a live turn. + + Caller must hold ``self._lock``. + """ + evicted: list[HermesSession] = [] + if not self._max_sessions: + return evicted + while len(self._sessions) > self._max_sessions: + victim = next( + (cid for cid, sess in self._sessions.items() if not sess.is_in_use()), + None, + ) + if victim is None: + break # all over-cap sessions are busy; allow temporary overshoot + evicted.append(self._sessions.pop(victim)) + return evicted + + @staticmethod + def _close_sessions(sessions: list[HermesSession]) -> None: + for session in sessions: + session.close() + + def cancel_all(self) -> None: + with self._lock: + sessions = list(self._sessions.values()) + for session in sessions: + session.begin_close() + + def close(self) -> None: + with self._lock: + sessions = list(self._sessions.values()) + self._sessions.clear() + self._close_sessions(sessions) + + def size(self) -> int: + with self._lock: + return len(self._sessions) + + def _make_agent(self, context_id: str) -> Any: + if self._agent_factory is not None: + return self._agent_factory() + + # Real runtime build — mirrors acp_adapter.session._make_agent so the + # A2A agent picks up the user's configured provider/model and toolsets. + from hermes_cli.config import load_config + from hermes_cli.runtime_provider import resolve_runtime_provider + from run_agent import AIAgent + + config = load_config() + model_cfg = config.get("model") + default_model = "" + config_provider = None + if isinstance(model_cfg, dict): + default_model = str(model_cfg.get("default") or "") + config_provider = model_cfg.get("provider") + elif isinstance(model_cfg, str) and model_cfg.strip(): + default_model = model_cfg.strip() + + from hermes_cli.tools_config import _get_platform_tools + + enabled_toolsets = sorted(_get_platform_tools(config, "a2a")) + agent_cfg = config.get("agent") or {} + disabled_toolsets = agent_cfg.get("disabled_toolsets") or None + + kwargs: dict[str, Any] = { + "platform": "a2a", + "enabled_toolsets": enabled_toolsets, + "disabled_toolsets": disabled_toolsets, + "quiet_mode": True, + "session_id": context_id, + "model": default_model, + } + try: + runtime = resolve_runtime_provider(requested=config_provider) + kwargs.update({ + "provider": runtime.get("provider"), + "api_mode": runtime.get("api_mode"), + "base_url": runtime.get("base_url"), + "api_key": runtime.get("api_key"), + "command": runtime.get("command"), + "args": list(runtime.get("args") or []), + }) + except Exception: + # Fall back to AIAgent's own default provider resolution. + pass + + return AIAgent(**kwargs) diff --git a/plugins/platforms/a2a/task_store.py b/plugins/platforms/a2a/task_store.py new file mode 100644 index 000000000000..1cd6abcc4906 --- /dev/null +++ b/plugins/platforms/a2a/task_store.py @@ -0,0 +1,100 @@ +"""Bounded A2A task persistence with monotonic terminal states.""" + +from __future__ import annotations + +import asyncio +from collections import OrderedDict + +from a2a.server.context import ServerCallContext +from a2a.server.tasks import TaskStore +from a2a.types import Task, TaskState + +_TERMINAL_STATES = frozenset({ + TaskState.completed, + TaskState.canceled, + TaskState.failed, + TaskState.rejected, +}) + + +def _clone(task: Task) -> Task: + return task.model_copy(deep=True) + + +class BoundedTaskStore(TaskStore): + """Keep task state bounded and prevent terminal-state resurrection. + + The SDK's task managers mutate retrieved task objects before saving them. + Returning deep copies prevents concurrent consumers (notably cancel and the + original non-blocking send consumer) from racing through shared references. + Once a terminal state is persisted it is authoritative and cannot be + overwritten by an older queued ``working`` event. + """ + + def __init__(self, *, max_tasks: int, max_history_messages: int): + if max_tasks <= 0 or max_history_messages <= 0: + raise ValueError("A2A task-store bounds must be positive") + self._max_tasks = max_tasks + self._max_history_messages = max_history_messages + self._tasks: OrderedDict[str, Task] = OrderedDict() + self._lock = asyncio.Lock() + + async def save(self, task: Task, context: ServerCallContext | None = None) -> None: + del context + incoming = _clone(task) + if incoming.history and len(incoming.history) > self._max_history_messages: + incoming.history = incoming.history[-self._max_history_messages :] + + async with self._lock: + current = self._tasks.get(incoming.id) + if ( + current is not None + and current.status.state in _TERMINAL_STATES + and incoming.status.state not in _TERMINAL_STATES + ): + return + if ( + current is not None + and current.status.state in _TERMINAL_STATES + and incoming.status.state in _TERMINAL_STATES + ): + return + self._tasks[incoming.id] = incoming + self._tasks.move_to_end(incoming.id) + while len(self._tasks) > self._max_tasks: + victim = next( + ( + task_id + for task_id, retained in self._tasks.items() + if retained.status.state in _TERMINAL_STATES + ), + None, + ) + if victim is None: + # Never make a running task disappear from tasks/get or + # tasks/cancel. Admission bounds active tasks; repair the + # temporary overshoot as soon as one becomes terminal. + break + del self._tasks[victim] + + async def get( + self, task_id: str, context: ServerCallContext | None = None + ) -> Task | None: + del context + async with self._lock: + task = self._tasks.get(task_id) + if task is None: + return None + self._tasks.move_to_end(task_id) + return _clone(task) + + async def delete( + self, task_id: str, context: ServerCallContext | None = None + ) -> None: + del context + async with self._lock: + self._tasks.pop(task_id, None) + + async def size(self) -> int: + async with self._lock: + return len(self._tasks) diff --git a/pyproject.toml b/pyproject.toml index 2b9242118e87..cbde3ee6af84 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -215,6 +215,16 @@ teams = ["microsoft-teams-apps==2.0.13.4", "aiohttp==3.14.1"] # aiohttp 3.14.1: # to it, which is already provided by the `mcp` extra. computer-use = ["mcp==1.26.0", "starlette==1.0.1"] # starlette: CVE-2026-48710 acp = ["agent-client-protocol==0.9.0"] +# A2A (Agent2Agent) server — exposes Hermes as a remote agent that other agents +# can call over JSON-RPC + SSE. See plugins/platforms/a2a/ and +# .plans/a2a-protocol.md. +# Pinned to the pydantic-based 0.3.x line (the one the broad A2A client +# ecosystem targets) rather than the proto-first 1.x. The http-server extra +# pulls the Starlette/uvicorn/sse stack the adapter serves on. starlette==1.0.1 +# is pinned directly for CVE-2026-48710 (BadHost) — a2a-sdk[http-server] pulls +# Starlette transitively with no upper bound, so pin the patched floor here the +# same way the web/mcp/computer-use extras do. +a2a = ["a2a-sdk[http-server]==0.3.26", "starlette==1.0.1"] # starlette: CVE-2026-48710 # mistral: Voxtral STT + TTS. Pinned to an exact verified-clean version. # The `mistralai` PyPI project was quarantined 2026-05-12 after the malicious # 2.4.6 release (Mini Shai-Hulud worm); 2.4.6 was removed from PyPI and the @@ -299,6 +309,7 @@ all = [ "hermes-agent[homeassistant]", "hermes-agent[sms]", "hermes-agent[acp]", + "hermes-agent[a2a]", "hermes-agent[google]", "hermes-agent[web]", "hermes-agent[youtube]", @@ -308,6 +319,7 @@ all = [ hermes = "hermes_cli.main:main" hermes-agent = "run_agent:main" hermes-acp = "acp_adapter.entry:main" +hermes-a2a = "plugins.platforms.a2a.entry:main" [tool.setuptools] py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajectory_compressor", "toolset_distributions", "cli", "hermes_bootstrap", "hermes_constants", "hermes_state", "hermes_time", "hermes_logging", "utils", "mcp_serve"] diff --git a/scripts/release.py b/scripts/release.py index dc4187115a18..795fc15a9261 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -72,6 +72,7 @@ "75556242+webtecnica@users.noreply.github.com": "webtecnica", # PR #63360 salvage (nous: restore inference-api base_url) "skosarevivan@yandex.ru": "Epoxidex", # PR #29820 salvage (ollama: top-level reasoning_effort=none; #25758) "jdjiayou@163.com": "JiaDe-Wu", # PR #34742 salvage (bedrock: bearer routing + streaming fallback + image decode; #28156) + "yugui923@users.noreply.github.com": "yugui923", "changhyun.min@gmail.com": "minchang", # PR #42231 salvage (providers: add Upstage Solar) "neo@neodeMac-mini.local": "neo-claw-bot", # PR #58465 salvage (moa: drop empty user turns from advisory view) "2024104039@mails.szu.edu.cn": "pixel4039", # PR #64420 salvage (streaming: retry zero-chunk streams) diff --git a/setup-hermes.sh b/setup-hermes.sh index 42cf2b759a5d..f2832698a0b7 100755 --- a/setup-hermes.sh +++ b/setup-hermes.sh @@ -215,7 +215,7 @@ else _BROKEN_EXTRAS=() # populate when an extra becomes unresolvable _ALL_EXTRAS=( modal daytona messaging matrix cron cli dev tts-premium slack - pty honcho mcp homeassistant sms acp voice dingtalk feishu google + pty honcho mcp homeassistant sms acp a2a voice dingtalk feishu google bedrock web youtube ) _SAFE_EXTRAS=() diff --git a/tests/a2a/__init__.py b/tests/a2a/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/a2a/conftest.py b/tests/a2a/conftest.py new file mode 100644 index 000000000000..8ccbacc9e09d --- /dev/null +++ b/tests/a2a/conftest.py @@ -0,0 +1,108 @@ +"""Shared fakes for the A2A adapter tests. + +All of these let the adapter run without model credentials: ``FakeAgent`` +stands in for ``AIAgent`` (injected via ``ContextSessionStore(agent_factory=...)``), +``FakeContext`` / ``RecordingQueue`` let us drive ``HermesAgentExecutor.execute`` +directly, and ``make_user_message`` builds a valid A2A request message. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest +from a2a.types import Message, Part, Role, TextPart + + +class FakeAgent: + """Minimal stand-in for ``AIAgent``. + + Exercises the callback bridge (one streamed delta, one tool start, one tool + result) and returns a deterministic echo response plus updated history. + """ + + def __init__(self) -> None: + self.stream_delta_callback = None + self.reasoning_callback = None + self.tool_progress_callback = None + self.step_callback = None + self.thinking_callback = None + self.interrupted = False + self.runs: list[str] = [] + + def run_conversation( + self, + *, + user_message: str, + conversation_history: list[dict[str, Any]] | None = None, + task_id: str | None = None, + **_kwargs: Any, + ) -> dict[str, Any]: + self.runs.append(user_message) + if self.stream_delta_callback: + self.stream_delta_callback("thinking... ") + if self.tool_progress_callback: + self.tool_progress_callback( + "tool.started", name="read_file", args={"path": "x.py"} + ) + if self.step_callback: + self.step_callback(1, [{"name": "read_file", "result": "file contents"}]) + final = f"echo: {user_message}" + messages = list(conversation_history or []) + messages.append({"role": "user", "content": user_message}) + messages.append({"role": "assistant", "content": final}) + return {"final_response": final, "messages": messages} + + def interrupt(self, message: str | None = None) -> None: + self.interrupted = True + + +class RecordingQueue: + """Captures events enqueued by the executor / TaskUpdater.""" + + def __init__(self) -> None: + self.events: list[Any] = [] + + async def enqueue_event(self, event: Any) -> None: + self.events.append(event) + + +class FakeContext: + """Stands in for a2a-sdk's ``RequestContext`` for direct executor tests.""" + + def __init__( + self, + user_text: str, + message: Message, + *, + current_task: Any = None, + context_id: str | None = None, + ) -> None: + self._user_text = user_text + self.message = message + self.current_task = current_task + self.context_id = context_id + + def get_user_input(self, delimiter: str = "\n") -> str: + return self._user_text + + +def make_user_message(text: str, context_id: str | None = None) -> Message: + return Message( + role=Role.user, + kind="message", + message_id="msg-test", + parts=[Part(root=TextPart(text=text))], + context_id=context_id, + ) + + +@pytest.fixture +def fakes() -> SimpleNamespace: + return SimpleNamespace( + FakeAgent=FakeAgent, + RecordingQueue=RecordingQueue, + FakeContext=FakeContext, + make_user_message=make_user_message, + ) diff --git a/tests/a2a/test_bootstrap.py b/tests/a2a/test_bootstrap.py new file mode 100644 index 000000000000..fdac9e118a00 --- /dev/null +++ b/tests/a2a/test_bootstrap.py @@ -0,0 +1,36 @@ +"""Behavioral bootstrap coverage for the standalone A2A entry point.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + + +def test_standalone_entry_hardens_import_path_before_gateway_imports(tmp_path): + shadow = tmp_path / "utils" + shadow.mkdir() + (shadow / "__init__.py").write_text( + 'raise RuntimeError("project-local utils was imported")\n', + encoding="utf-8", + ) + repo_root = Path(__file__).resolve().parents[2] + env = os.environ.copy() + env["PYTHONPATH"] = os.pathsep.join([ + str(repo_root), + env.get("PYTHONPATH", ""), + ]).rstrip(os.pathsep) + + result = subprocess.run( + [sys.executable, "-m", "plugins.platforms.a2a", "--version"], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + timeout=15, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert result.stdout.strip() diff --git a/tests/a2a/test_card.py b/tests/a2a/test_card.py new file mode 100644 index 000000000000..b89ece68bf73 --- /dev/null +++ b/tests/a2a/test_card.py @@ -0,0 +1,42 @@ +"""The Agent Card is correct and served at the well-known URL.""" + +from __future__ import annotations + +from starlette.testclient import TestClient + +from plugins.platforms.a2a.card import build_agent_card +from plugins.platforms.a2a.entry import _default_service_url, build_app + + +def test_card_has_required_fields(): + card = build_agent_card("http://localhost:9100/") + assert card.name == "Hermes Agent" + assert card.url == "http://localhost:9100/" + assert card.version + assert card.capabilities.streaming is True + assert card.skills and card.skills[0].id == "general-agent" + + +def test_card_serializes_with_camelcase_aliases(): + card = build_agent_card("http://localhost:9100/") + dumped = card.model_dump(by_alias=True, exclude_none=True) + # A2A wire format is camelCase. + assert "defaultInputModes" in dumped + assert "protocolVersion" in dumped + assert dumped["preferredTransport"] == "JSONRPC" + + +def test_card_served_at_well_known_url(): + app = build_app("127.0.0.1", 9100) + with TestClient(app) as client: + resp = client.get("/.well-known/agent-card.json") + assert resp.status_code == 200 + body = resp.json() + assert body["name"] == "Hermes Agent" + assert body["preferredTransport"] == "JSONRPC" + assert any(skill["id"] == "general-agent" for skill in body["skills"]) + + +def test_default_service_url_brackets_ipv6_hosts(): + assert _default_service_url("::1", 9100) == "http://[::1]:9100/" + assert _default_service_url("127.0.0.1", 9100) == "http://127.0.0.1:9100/" diff --git a/tests/a2a/test_config.py b/tests/a2a/test_config.py new file mode 100644 index 000000000000..70c116905959 --- /dev/null +++ b/tests/a2a/test_config.py @@ -0,0 +1,105 @@ +"""A2A behavioral settings are loaded from config.yaml-shaped data.""" + +from copy import deepcopy + +from hermes_cli.config import DEFAULT_CONFIG +from hermes_cli.tools_config import _get_platform_tools +from plugins.platforms.a2a.config import ( + DEFAULT_MAX_CONCURRENCY, + DEFAULT_TOOL_IO, + A2ASettings, + apply_yaml_config, +) + + +def test_behavioral_settings_accept_config_values(): + settings = A2ASettings.from_mapping({ + "max_concurrency": 7, + "max_sessions": 99, + "tool_io": "none", + "host": "0.0.0.0", + "port": 9200, + }) + + assert settings.max_concurrency == 7 + assert settings.max_sessions == 99 + assert settings.tool_io == "none" + assert settings.host == "0.0.0.0" + assert settings.port == 9200 + + +def test_invalid_behavioral_settings_fall_back_safely(): + settings = A2ASettings.from_mapping({ + "max_concurrency": 0, + "max_tasks": 1, + "tool_io": "secrets", + "port": 70000, + }) + + assert settings.max_concurrency == DEFAULT_MAX_CONCURRENCY + assert settings.max_tasks >= settings.max_concurrency + assert settings.tool_io == DEFAULT_TOOL_IO + assert settings.port == 9100 + + +def test_platform_yaml_bridge_seeds_a2a_extras(): + seeded = apply_yaml_config( + {}, + { + "enabled": True, + "host": "127.0.0.1", + "max_concurrency": 4, + "tool_io": "full", + }, + ) + + assert seeded == { + "host": "127.0.0.1", + "max_concurrency": 4, + "tool_io": "full", + } + + +def test_gateway_loader_applies_a2a_config_through_plugin_registry( + tmp_path, monkeypatch +): + """Exercise the real YAML -> plugin hook -> PlatformConfig chain.""" + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + "a2a:\n" + " enabled: true\n" + " host: 0.0.0.0\n" + " port: 9200\n" + " max_concurrency: 7\n" + " max_sessions: 99\n" + " tool_io: none\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + from gateway.config import Platform, load_gateway_config + + platform_config = load_gateway_config().platforms[Platform("a2a")] + + assert platform_config.enabled is True + assert platform_config.extra == { + "host": "0.0.0.0", + "port": 9200, + "max_concurrency": 7, + "max_sessions": 99, + "tool_io": "none", + } + + +def test_default_and_disabled_tools_use_generic_platform_configuration(): + defaults = _get_platform_tools(DEFAULT_CONFIG, "a2a") + assert {"terminal", "file", "web"}.issubset(defaults) + + configured = deepcopy(DEFAULT_CONFIG) + configured["platform_toolsets"]["a2a"] = ["terminal", "file", "no_mcp"] + configured.setdefault("agent", {})["disabled_toolsets"] = ["terminal"] + + resolved = _get_platform_tools(configured, "a2a") + assert "file" in resolved + assert "terminal" not in resolved diff --git a/tests/a2a/test_end_to_end_echo.py b/tests/a2a/test_end_to_end_echo.py new file mode 100644 index 000000000000..843d8fb555d3 --- /dev/null +++ b/tests/a2a/test_end_to_end_echo.py @@ -0,0 +1,175 @@ +"""End-to-end JSON-RPC over the real Starlette app, in-process, no LLM. + +Builds the actual A2A app (card + DefaultRequestHandler + BoundedTaskStore) +around an echo agent and drives a ``message/send`` request through it via the +Starlette test client — exercising the full transport path (routing, JSON-RPC +decode, executor, event consumption, task assembly). +""" + +from __future__ import annotations + +import json +import threading +from contextlib import asynccontextmanager + +from a2a.server.apps import A2AStarletteApplication +from a2a.server.request_handlers import DefaultRequestHandler +from starlette.testclient import TestClient + +from plugins.platforms.a2a.card import build_agent_card +from plugins.platforms.a2a.executor import HermesAgentExecutor +from plugins.platforms.a2a.sessions import ContextSessionStore +from plugins.platforms.a2a.task_store import BoundedTaskStore + + +def _build_echo_client(fakes, agent_factory=None) -> TestClient: + store = ContextSessionStore(agent_factory=agent_factory or fakes.FakeAgent) + executor = HermesAgentExecutor(store) + handler = DefaultRequestHandler( + agent_executor=executor, + task_store=BoundedTaskStore(max_tasks=32, max_history_messages=16), + ) + + @asynccontextmanager + async def lifespan(_app): + try: + yield + finally: + await executor.aclose() + + app = A2AStarletteApplication( + agent_card=build_agent_card("http://test/"), + http_handler=handler, + ).build(lifespan=lifespan) + return TestClient(app) + + +def test_message_send_returns_completed_task_with_echo(fakes): + request = { + "jsonrpc": "2.0", + "id": "1", + "method": "message/send", + "params": { + "message": { + "role": "user", + "kind": "message", + "messageId": "m1", + "parts": [{"kind": "text", "text": "ping"}], + } + }, + } + with _build_echo_client(fakes) as client: + resp = client.post("/", json=request) + assert resp.status_code == 200 + body = resp.json() + assert "error" not in body, body + result = body["result"] + + # message/send returns the final Task once it reaches a terminal state. + assert result["kind"] == "task" + assert result["status"]["state"] == "completed" + + # The echo response is delivered as an artifact. + text = result["artifacts"][0]["parts"][0]["text"] + assert text == "echo: ping" + # Sanity: the whole payload mentions the echo. + assert "echo: ping" in json.dumps(body) + + +def test_nonblocking_cancel_remains_canceled_in_task_store(fakes): + class BlockingAgent: + def __init__(self): + self.stream_delta_callback = None + self.reasoning_callback = None + self.tool_progress_callback = None + self.step_callback = None + self.thinking_callback = None + self._interrupted = threading.Event() + + def run_conversation(self, **_kwargs): + self._interrupted.wait(5) + return {"final_response": None, "interrupted": True, "messages": []} + + def interrupt(self, _message=None): + self._interrupted.set() + + def clear_interrupt(self): + self._interrupted.clear() + + send_request = { + "jsonrpc": "2.0", + "id": "send", + "method": "message/send", + "params": { + "message": { + "role": "user", + "kind": "message", + "messageId": "m-cancel", + "parts": [{"kind": "text", "text": "wait"}], + }, + "configuration": {"blocking": False}, + }, + } + + with _build_echo_client(fakes, BlockingAgent) as client: + send = client.post("/", json=send_request).json() + assert "error" not in send, send + task_id = send["result"]["id"] + + canceled = client.post( + "/", + json={ + "jsonrpc": "2.0", + "id": "cancel", + "method": "tasks/cancel", + "params": {"id": task_id}, + }, + ).json() + assert "error" not in canceled, canceled + assert canceled["result"]["status"]["state"] == "canceled" + + fetched = client.post( + "/", + json={ + "jsonrpc": "2.0", + "id": "get", + "method": "tasks/get", + "params": {"id": task_id}, + }, + ).json() + assert "error" not in fetched, fetched + assert fetched["result"]["status"]["state"] == "canceled" + + +def test_message_stream_emits_sse_artifact_and_terminal_status(fakes): + request = { + "jsonrpc": "2.0", + "id": "stream", + "method": "message/stream", + "params": { + "message": { + "role": "user", + "kind": "message", + "messageId": "m-stream", + "parts": [{"kind": "text", "text": "stream ping"}], + } + }, + } + + with _build_echo_client(fakes) as client: + with client.stream("POST", "/", json=request) as response: + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + payloads = [ + json.loads(line.removeprefix("data: ")) + for line in response.iter_lines() + if line.startswith("data: ") + ] + + results = [payload["result"] for payload in payloads if "result" in payload] + assert any(result.get("kind") == "artifact-update" for result in results) + assert any( + result.get("kind") == "status-update" + and result["status"]["state"] == "completed" + for result in results + ) diff --git a/tests/a2a/test_events.py b/tests/a2a/test_events.py new file mode 100644 index 000000000000..bce23f30a5ad --- /dev/null +++ b/tests/a2a/test_events.py @@ -0,0 +1,47 @@ +"""Tool-progress metadata sent to the peer is bounded and operator-controllable. + +An A2A server delegates work for a remote peer; echoing unbounded tool arguments +and results (file contents, shell output, secrets) into status metadata is both a +size and a disclosure hazard. ``a2a.tool_io`` controls the exposure: +``preview`` (default, bounded), ``none`` (names only), ``full`` (unbounded). +""" + +from __future__ import annotations + +from plugins.platforms.a2a.events import ( + _RESULT_PREVIEW_LIMIT, + _tool_call_metadata, + _tool_result_metadata, +) + + +def test_preview_mode_bounds_large_args(): + md = _tool_call_metadata("read_file", {"blob": "x" * 5000}) + assert md["hermes/kind"] == "tool-call" + assert md["hermes/tool"] == "read_file" + assert isinstance(md["hermes/args"], str) + assert len(md["hermes/args"]) <= _RESULT_PREVIEW_LIMIT + 1 # + ellipsis + + +def test_preview_mode_keeps_small_structured_args(): + md = _tool_call_metadata("read_file", {"path": "a.py"}) + assert md["hermes/args"] == {"path": "a.py"} + + +def test_preview_mode_bounds_large_results(): + md = _tool_result_metadata("terminal", "y" * 5000) + assert md["hermes/kind"] == "tool-result" + assert isinstance(md["hermes/result"], str) + assert len(md["hermes/result"]) <= _RESULT_PREVIEW_LIMIT + 1 + + +def test_none_mode_omits_args_and_results(): + call = _tool_call_metadata("terminal", {"cmd": "cat secrets.env"}, "none") + result = _tool_result_metadata("terminal", "API_KEY=sk-very-secret", "none") + assert call["hermes/tool"] == "terminal" and "hermes/args" not in call + assert result["hermes/tool"] == "terminal" and "hermes/result" not in result + + +def test_full_mode_preserves_unbounded_structure(): + md = _tool_call_metadata("read_file", {"path": "a.py", "blob": "x" * 5000}, "full") + assert md["hermes/args"] == {"path": "a.py", "blob": "x" * 5000} diff --git a/tests/a2a/test_executor.py b/tests/a2a/test_executor.py new file mode 100644 index 000000000000..d70c0d497731 --- /dev/null +++ b/tests/a2a/test_executor.py @@ -0,0 +1,340 @@ +"""HermesAgentExecutor maps AIAgent turn outcomes onto A2A task states.""" + +from __future__ import annotations + +import asyncio +import contextlib +import threading +import time + +import pytest +from a2a.types import ( + Task, + TaskArtifactUpdateEvent, + TaskState, + TaskStatusUpdateEvent, +) +from a2a.utils.errors import ServerError + +from plugins.platforms.a2a.executor import HermesAgentExecutor +from plugins.platforms.a2a.sessions import ContextSessionStore + + +class _OutcomeAgent: + """Agent stub that returns a caller-specified ``run_conversation`` result.""" + + def __init__(self, result: dict): + self._result = result + self.stream_delta_callback = None + self.reasoning_callback = None + self.tool_progress_callback = None + self.step_callback = None + self.thinking_callback = None + self.runs: list[str] = [] + + def run_conversation( + self, *, user_message, conversation_history=None, task_id=None, **kw + ): + self.runs.append(user_message) + return self._result + + def interrupt(self, message=None): + pass + + +def _run(fakes, agent, user_text="do the thing", context_id="ctx-out"): + store = ContextSessionStore(agent_factory=lambda: agent) + executor = HermesAgentExecutor(store) + message = fakes.make_user_message(user_text, context_id=context_id) + context = fakes.FakeContext( + user_text, message, current_task=None, context_id=context_id + ) + queue = fakes.RecordingQueue() + asyncio.run(executor.execute(context, queue)) + return queue.events + + +def _terminal_state(events): + statuses = [e for e in events if isinstance(e, TaskStatusUpdateEvent)] + return statuses[-1].status.state if statuses else None + + +# --- happy path ------------------------------------------------------------ + + +def test_execute_event_sequence(fakes): + agent = fakes.FakeAgent() + events = _run(fakes, agent, user_text="hello") + + assert isinstance(events[0], Task) # Task enqueued first + statuses = [e for e in events if isinstance(e, TaskStatusUpdateEvent)] + artifacts = [e for e in events if isinstance(e, TaskArtifactUpdateEvent)] + + assert any(s.status.state == TaskState.working for s in statuses) + assert statuses[-1].status.state == TaskState.completed + assert len(artifacts) == 1 + assert artifacts[0].artifact.parts[0].root.text == "echo: hello" + assert agent.runs == ["hello"] + + +def test_callback_bridge_surfaces_tool_activity(fakes): + events = _run(fakes, fakes.FakeAgent(), user_text="hello") + statuses = [e for e in events if isinstance(e, TaskStatusUpdateEvent)] + kinds = {s.metadata.get("hermes/kind") for s in statuses if s.metadata} + assert "tool-call" in kinds + assert "tool-result" in kinds + + +# --- outcome mapping (regression for the "failures look like success" bug) -- + + +def test_failed_result_marks_task_failed(fakes): + events = _run( + fakes, + _OutcomeAgent({ + "final_response": None, + "failed": True, + "error": "provider 500", + }), + ) + assert _terminal_state(events) == TaskState.failed + assert not any(isinstance(e, TaskArtifactUpdateEvent) for e in events) + + +def test_failed_result_carries_explanatory_text(fakes): + events = _run( + fakes, + _OutcomeAgent({"final_response": "blocked by content policy", "failed": True}), + ) + assert _terminal_state(events) == TaskState.failed + statuses = [e for e in events if isinstance(e, TaskStatusUpdateEvent)] + msg = statuses[-1].status.message + assert msg is not None and msg.parts[0].root.text == "blocked by content policy" + assert not any(isinstance(e, TaskArtifactUpdateEvent) for e in events) + + +def test_interrupted_result_marks_canceled(fakes): + events = _run( + fakes, _OutcomeAgent({"final_response": "partial", "interrupted": True}) + ) + assert _terminal_state(events) == TaskState.canceled + assert not any(isinstance(e, TaskArtifactUpdateEvent) for e in events) + + +def test_empty_final_response_marks_failed(fakes): + events = _run(fakes, _OutcomeAgent({"final_response": "", "messages": []})) + assert _terminal_state(events) == TaskState.failed + assert not any(isinstance(e, TaskArtifactUpdateEvent) for e in events) + + +def test_error_result_without_failed_flag_marks_failed(fakes): + """Degraded turns carry ``error`` but no ``failed``/``interrupted`` flag. + + ``run_conversation`` has early-return paths (e.g. thinking-budget exhausted, + truncation) that set ``error`` + ``partial`` + a human-readable + ``final_response`` but never reach ``finalize_turn``, so the dict has no + ``failed`` key. These must be reported to the peer as a failure, not as a + successful completion whose artifact is actually an error notice. + """ + events = _run( + fakes, + _OutcomeAgent({ + "final_response": "⚠️ Thinking Budget Exhausted", + "error": "thinking budget exhausted before any response", + "partial": True, + "completed": False, + }), + ) + assert _terminal_state(events) == TaskState.failed + assert not any(isinstance(e, TaskArtifactUpdateEvent) for e in events) + + +def test_failed_no_response_preserves_error_detail(fakes): + """When there is no usable text, the specific error must survive. + + Previously an empty ``final_response`` collapsed every distinct failure into + a generic "Agent produced no response." — discarding the diagnostic the peer + needs. + """ + events = _run( + fakes, + _OutcomeAgent({ + "final_response": None, + "error": "Response truncated due to output length limit", + }), + ) + assert _terminal_state(events) == TaskState.failed + statuses = [e for e in events if isinstance(e, TaskStatusUpdateEvent)] + text = statuses[-1].status.message.parts[0].root.text + assert "Response truncated due to output length limit" in text + + +# --- concurrency bound ----------------------------------------------------- + + +class _ConcurrencyProbe: + """Records the peak number of turns running its loop simultaneously.""" + + _state_lock = threading.Lock() + live = 0 + peak = 0 + + def __init__(self) -> None: + self.stream_delta_callback = None + self.reasoning_callback = None + self.tool_progress_callback = None + self.step_callback = None + self.thinking_callback = None + + def run_conversation( + self, *, user_message, conversation_history=None, task_id=None, **kw + ): + with _ConcurrencyProbe._state_lock: + _ConcurrencyProbe.live += 1 + _ConcurrencyProbe.peak = max(_ConcurrencyProbe.peak, _ConcurrencyProbe.live) + try: + time.sleep(0.05) + finally: + with _ConcurrencyProbe._state_lock: + _ConcurrencyProbe.live -= 1 + return {"final_response": "ok", "messages": []} + + def interrupt(self, message=None): + pass + + def clear_interrupt(self): + pass + + +def test_concurrent_turns_are_bounded_and_overload_is_rejected(fakes): + """The limit bounds admitted work rather than merely worker threads.""" + _ConcurrencyProbe.live = 0 + _ConcurrencyProbe.peak = 0 + store = ContextSessionStore(agent_factory=_ConcurrencyProbe) + executor = HermesAgentExecutor(store, max_concurrency=1) + + async def drive(): + async def one(ctx): + message = fakes.make_user_message("go", context_id=ctx) + context = fakes.FakeContext( + "go", message, current_task=None, context_id=ctx + ) + await executor.execute(context, fakes.RecordingQueue()) + + first = asyncio.create_task(one("ctx-a")) + for _ in range(100): + if _ConcurrencyProbe.live: + break + await asyncio.sleep(0.001) + assert _ConcurrencyProbe.live == 1 + + with pytest.raises(ServerError, match="at turn capacity"): + await one("ctx-b") + + await first + await executor.aclose() + + asyncio.run(drive()) + assert _ConcurrencyProbe.peak == 1 + + +def test_canceled_request_keeps_capacity_reserved_until_worker_exits(fakes): + started = threading.Event() + release = threading.Event() + + class StubbornAgent(_OutcomeAgent): + def __init__(self): + super().__init__({"final_response": "done", "messages": []}) + + def run_conversation(self, **_kwargs): + started.set() + release.wait(5) + return self._result + + executor = HermesAgentExecutor( + ContextSessionStore(agent_factory=StubbornAgent), max_concurrency=1 + ) + + async def one(context_id): + message = fakes.make_user_message("go", context_id=context_id) + context = fakes.FakeContext( + "go", message, current_task=None, context_id=context_id + ) + await executor.execute(context, fakes.RecordingQueue()) + + async def drive(): + first = asyncio.create_task(one("ctx-a")) + assert await asyncio.to_thread(started.wait, 5) + first.cancel() + with contextlib.suppress(asyncio.CancelledError): + await first + + with pytest.raises(ServerError, match="at turn capacity"): + await one("ctx-b") + + release.set() + await executor.aclose() + + asyncio.run(drive()) + + +def test_close_prevents_admitted_request_from_starting_after_await(fakes): + """Shutdown cannot be bypassed by a pre-worker protocol await. + + A request reserves capacity before publishing its initial task. If that + publish blocks while ``aclose()`` runs, resuming it must not recreate the + closed session store or lazy worker pool. + """ + enqueue_started = asyncio.Event() + release_enqueue = asyncio.Event() + agent = _OutcomeAgent({"final_response": "late", "messages": []}) + store = ContextSessionStore(agent_factory=lambda: agent) + executor = HermesAgentExecutor(store) + + class BlockingQueue(fakes.RecordingQueue): + async def enqueue_event(self, event): + enqueue_started.set() + await release_enqueue.wait() + await super().enqueue_event(event) + + async def drive(): + message = fakes.make_user_message("late", context_id="ctx-late") + context = fakes.FakeContext( + "late", message, current_task=None, context_id="ctx-late" + ) + queue = BlockingQueue() + request = asyncio.create_task(executor.execute(context, queue)) + await enqueue_started.wait() + + await executor.aclose() + release_enqueue.set() + + await request + assert _terminal_state(queue.events) == TaskState.failed + + asyncio.run(drive()) + assert executor._turn_pool is None + assert not store._sessions + assert agent.runs == [] + + +# --- invalid input --------------------------------------------------------- + + +def test_blank_or_empty_message_rejected(fakes): + """Empty/whitespace input is a JSON-RPC error, not a 'completed' task. + + Both whitespace (" ") and a truly-empty TextPart ("") must be rejected + before ``new_task`` (which itself raises on an empty TextPart). + """ + store = ContextSessionStore(agent_factory=fakes.FakeAgent) + executor = HermesAgentExecutor(store) + for text in (" ", ""): + message = fakes.make_user_message(text, context_id="ctx-empty") + context = fakes.FakeContext( + text, message, current_task=None, context_id="ctx-empty" + ) + queue = fakes.RecordingQueue() + with pytest.raises(ServerError): + asyncio.run(executor.execute(context, queue)) + assert queue.events == [] # nothing enqueued for invalid input diff --git a/tests/a2a/test_plugin.py b/tests/a2a/test_plugin.py new file mode 100644 index 000000000000..421afcb5a930 --- /dev/null +++ b/tests/a2a/test_plugin.py @@ -0,0 +1,78 @@ +"""A2A registers through the generic bundled platform plugin interface.""" + +import asyncio +import socket +from unittest.mock import MagicMock + +from gateway.config import PlatformConfig +from gateway.platforms.base import BasePlatformAdapter +from plugins.platforms.a2a.adapter import A2AAdapter, register + + +def test_register_uses_platform_plugin_surface(): + ctx = MagicMock() + + register(ctx) + + ctx.register_platform.assert_called_once() + kwargs = ctx.register_platform.call_args.kwargs + assert kwargs["name"] == "a2a" + assert kwargs["adapter_factory"](PlatformConfig()) + assert callable(kwargs["apply_yaml_config_fn"]) + + +def test_adapter_implements_base_platform_contract(): + adapter = A2AAdapter(PlatformConfig(enabled=True)) + + assert isinstance(adapter, BasePlatformAdapter) + + +def test_adapter_returns_false_when_port_is_already_bound(): + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen() + port = listener.getsockname()[1] + adapter = A2AAdapter( + PlatformConfig(enabled=True, extra={"host": "127.0.0.1", "port": port}) + ) + + async def exercise(): + assert await adapter.connect() is False + assert adapter.has_fatal_error + await adapter.disconnect() + + try: + asyncio.run(exercise()) + finally: + listener.close() + + +def test_unexpected_server_exit_notifies_without_self_await(): + adapter = A2AAdapter(PlatformConfig(enabled=True)) + + class StoppedServer: + should_exit = False + + async def serve(self): + return None + + async def exercise(): + notified = asyncio.Event() + + async def fatal_handler(failed_adapter): + await failed_adapter.disconnect() + notified.set() + + adapter._server = StoppedServer() + adapter._running = True + adapter.set_fatal_error_handler(fatal_handler) + serve_task = asyncio.create_task(adapter._serve_embedded()) + adapter._serve_task = serve_task + + await serve_task + await asyncio.wait_for(notified.wait(), timeout=2) + + assert adapter.has_fatal_error + assert adapter._serve_task is None + + asyncio.run(exercise()) diff --git a/tests/a2a/test_sessions.py b/tests/a2a/test_sessions.py new file mode 100644 index 000000000000..39ac5721b9c7 --- /dev/null +++ b/tests/a2a/test_sessions.py @@ -0,0 +1,360 @@ +"""ContextSessionStore: one agent per context, history continuity, cancellation.""" + +from __future__ import annotations + +import threading + +from plugins.platforms.a2a.sessions import ContextSessionStore + + +class _InterruptibleAgent: + """Fake that mimics AIAgent's interrupt semantics. + + ``interrupt()`` sets a sticky ``_interrupt_requested`` flag; a turn that + sees the flag set at its start returns interrupted (as AIAgent's loop does); + ``clear_interrupt()`` resets it. + """ + + def __init__(self) -> None: + self._interrupt_requested = False + self.stream_delta_callback = None + self.reasoning_callback = None + self.tool_progress_callback = None + self.step_callback = None + self.thinking_callback = None + self.runs: list[str] = [] + + def run_conversation( + self, *, user_message, conversation_history=None, task_id=None, **kw + ): + if self._interrupt_requested: + return {"final_response": None, "interrupted": True} + self.runs.append(user_message) + msgs = list(conversation_history or []) + msgs += [ + {"role": "user", "content": user_message}, + {"role": "assistant", "content": f"echo: {user_message}"}, + ] + return {"final_response": f"echo: {user_message}", "messages": msgs} + + def interrupt(self, message=None): + self._interrupt_requested = True + + def clear_interrupt(self): + self._interrupt_requested = False + + +def test_same_context_reuses_one_agent(fakes): + created = [] + + def factory(): + agent = fakes.FakeAgent() + created.append(agent) + return agent + + store = ContextSessionStore(agent_factory=factory) + first = store.get_or_create("ctx-1") + again = store.get_or_create("ctx-1") + other = store.get_or_create("ctx-2") + + assert first is again + assert other is not first + assert len(created) == 2 + + +def test_run_turn_appends_to_history(fakes): + store = ContextSessionStore(agent_factory=fakes.FakeAgent) + session = store.get_or_create("ctx-1") + + first = session.run_turn("hello", task_id="t1") + assert first["final_response"] == "echo: hello" + assert session.history[-1] == {"role": "assistant", "content": "echo: hello"} + + # Second turn sees the prior history. + session.run_turn("again", task_id="t2") + assert session.agent.runs == ["hello", "again"] + assert len(session.history) == 4 + + +def test_cancel_sets_event_and_interrupts(fakes): + agent = fakes.FakeAgent() + store = ContextSessionStore(agent_factory=lambda: agent) + session = store.get_or_create("ctx-1") + + session.cancel() + + assert session.cancel_event.is_set() + assert agent.interrupted is True + + +def test_lru_eviction_caps_session_count(fakes): + store = ContextSessionStore(agent_factory=fakes.FakeAgent, max_sessions=2) + a = store.get_or_create("a") + store.get_or_create("b") + # Touch "a" so "b" becomes the least-recently-used entry. + store.get_or_create("a") + c = store.get_or_create("c") # over cap -> evicts LRU ("b") + + assert store.get("b") is None + assert store.get("a") is a + assert store.get("c") is c + + +def test_cancel_after_turn_does_not_poison_next_turn(): + """A cancel that arrives with no turn running must not abort the next turn. + + AIAgent.interrupt() sets a sticky flag that is only cleared by a turn that + runs to completion. If cancel() fires while idle (e.g. just after a turn + finished, or on a reused context), the next turn must start from a clean + slate instead of inheriting the stale interrupt and aborting immediately. + """ + agent = _InterruptibleAgent() + session = ContextSessionStore(agent_factory=lambda: agent).get_or_create("ctx") + + session.cancel() # no turn running; sets the sticky interrupt flag + assert agent._interrupt_requested is True + + result = session.run_turn("hello", task_id="t-new") + assert result.get("interrupted") is not True + assert result["final_response"] == "echo: hello" + + +def test_cancel_of_non_running_task_does_not_interrupt_agent(): + """Task-scoped cancel: cancelling a task that is not the running one must + not interrupt the agent (which would kill an unrelated in-flight turn on the + same context). The targeted task is instead skipped if it later starts.""" + agent = _InterruptibleAgent() + session = ContextSessionStore(agent_factory=lambda: agent).get_or_create("ctx") + + session.cancel(task_id="queued-task") # nothing running on this context + assert agent._interrupt_requested is False + + result = session.run_turn("hi", task_id="queued-task") + assert result.get("interrupted") is True + assert agent.runs == [] # the cancelled task never actually ran + + +def test_run_turn_applies_passed_callbacks_under_the_turn(): + """Per-turn callbacks must be the ones active during this turn's run, and be + cleared afterwards — so concurrent turns can't cross-wire the shared agent.""" + agent = _InterruptibleAgent() + seen = [] + agent.run_conversation = ( # type: ignore[method-assign] + lambda **kw: ( + agent.stream_delta_callback("d"), + {"final_response": "ok", "messages": []}, + )[1] + ) + + def cb(text): + seen.append(text) + + session = ContextSessionStore(agent_factory=lambda: agent).get_or_create("ctx") + session.run_turn("go", task_id="t1", callbacks={"stream_delta_callback": cb}) + + assert seen == ["d"] # the passed callback fired during the turn + assert agent.stream_delta_callback is None # cleared after the turn + + +def test_lru_does_not_evict_an_in_flight_session(fakes): + """An in-flight turn's session must survive eviction; an idle LRU session is + dropped instead. Evicting a busy session silently forks its history into a + fresh empty agent and orphans the running worker thread.""" + release = threading.Event() + started = threading.Event() + + class BlockingAgent(_InterruptibleAgent): + def run_conversation( + self, *, user_message, conversation_history=None, task_id=None, **kw + ): + started.set() + release.wait(5) + return {"final_response": "done", "messages": []} + + blocking = BlockingAgent() + agents = iter([blocking, fakes.FakeAgent(), fakes.FakeAgent()]) + store = ContextSessionStore(agent_factory=lambda: next(agents), max_sessions=2) + + busy = store.get_or_create("busy") + store.get_or_create("idle") # idle, least-recently-used after "busy" runs + + worker = threading.Thread(target=lambda: busy.run_turn("x", task_id="t-busy")) + worker.start() + assert started.wait(5) # turn is now in flight -> "busy" is active + + try: + third = store.get_or_create("third") # over cap -> must evict an idle one + assert store.get("busy") is busy # in-flight session preserved + assert store.get("idle") is None # idle LRU evicted instead + assert store.get("third") is third + finally: + release.set() + worker.join(5) + + +def test_run_turn_scopes_resources_to_context_and_denies_remote_approval(): + from tools import terminal_tool + + seen = {} + + class ApprovalAgent(_InterruptibleAgent): + def run_conversation(self, *, task_id=None, **kwargs): + callback = terminal_tool._get_approval_callback() + seen["task_id"] = task_id + seen["decision"] = callback("rm -rf /tmp/x", "destructive") + return {"final_response": "denied", "messages": []} + + sentinel = lambda *_args, **_kwargs: "once" + terminal_tool.set_approval_callback(sentinel) + try: + session = ContextSessionStore(agent_factory=ApprovalAgent).get_or_create( + "ctx-stable" + ) + session.run_turn("go", task_id="a2a-task-ephemeral") + assert terminal_tool._get_approval_callback() is sentinel + finally: + terminal_tool.set_approval_callback(None) + + assert seen == {"task_id": "ctx-stable", "decision": "deny"} + + +def test_real_agent_factory_honors_platform_and_global_tool_config(monkeypatch): + import hermes_cli.config as config_module + import hermes_cli.runtime_provider as runtime_module + import hermes_cli.tools_config as tools_config_module + import run_agent + + captured = {} + config = { + "model": {"default": "model-x", "provider": "provider-x"}, + "platform_toolsets": {"a2a": ["file", "no_mcp"]}, + "agent": {"disabled_toolsets": ["terminal"]}, + } + + class CapturingAgent: + def __init__(self, **kwargs): + captured.update(kwargs) + + def resolve_tools(user_config, platform): + assert user_config is config + assert platform == "a2a" + return {"file"} + + monkeypatch.setattr(config_module, "load_config", lambda: config) + monkeypatch.setattr(tools_config_module, "_get_platform_tools", resolve_tools) + monkeypatch.setattr( + runtime_module, + "resolve_runtime_provider", + lambda **_kwargs: {"provider": "provider-x"}, + ) + monkeypatch.setattr(run_agent, "AIAgent", CapturingAgent) + + ContextSessionStore()._make_agent("ctx-configured") + + assert captured["enabled_toolsets"] == ["file"] + assert captured["disabled_toolsets"] == ["terminal"] + assert captured["platform"] == "a2a" + assert captured["session_id"] == "ctx-configured" + + +def test_session_lease_blocks_eviction_and_eviction_closes_agent(): + class ClosableAgent(_InterruptibleAgent): + def __init__(self): + super().__init__() + self.closed = False + + def close(self): + self.closed = True + + agents = [] + + def factory(): + agent = ClosableAgent() + agents.append(agent) + return agent + + store = ContextSessionStore(agent_factory=factory, max_sessions=1) + leased = store.acquire("leased") + transient = store.get_or_create("transient") + + assert store.get("leased") is leased + assert store.get("transient") is None + assert transient.agent.closed is True + + store.release(leased) + replacement = store.get_or_create("replacement") + assert store.get("leased") is None + assert leased.agent.closed is True + assert store.get("replacement") is replacement + + +def test_store_close_releases_all_retained_agents(): + class ClosableAgent(_InterruptibleAgent): + def __init__(self): + super().__init__() + self.closed = False + + def close(self): + self.closed = True + + agents = [ClosableAgent(), ClosableAgent()] + iterator = iter(agents) + store = ContextSessionStore(agent_factory=lambda: next(iterator)) + store.get_or_create("one") + store.get_or_create("two") + + store.close() + + assert all(agent.closed for agent in agents) + + +def test_releasing_a_lease_repairs_temporary_session_cap_overshoot(): + store = ContextSessionStore(agent_factory=_InterruptibleAgent, max_sessions=1) + first = store.acquire("first") + second = store.acquire("second") + assert store.size() == 2 + + store.release(first) + assert store.size() == 1 + assert store.get("first") is None + assert store.get("second") is second + + store.release(second) + + +def test_begin_close_prevents_queued_turn_from_starting(): + started = threading.Event() + release = threading.Event() + + class BlockingAgent(_InterruptibleAgent): + def run_conversation(self, **_kwargs): + self.runs.append("run") + started.set() + release.wait(5) + return {"final_response": None, "interrupted": True, "messages": []} + + def interrupt(self, message=None): + super().interrupt(message) + release.set() + + session = ContextSessionStore(agent_factory=BlockingAgent).get_or_create( + "ctx-shutdown" + ) + results = [] + first = threading.Thread( + target=lambda: results.append(session.run_turn("one", task_id="one")) + ) + second = threading.Thread( + target=lambda: results.append(session.run_turn("two", task_id="two")) + ) + first.start() + assert started.wait(5) + second.start() + + session.begin_close() + first.join(5) + second.join(5) + + assert session.agent.runs == ["run"] + assert len(results) == 2 + assert all(result.get("interrupted") for result in results) diff --git a/tests/a2a/test_task_store.py b/tests/a2a/test_task_store.py new file mode 100644 index 000000000000..23426c936c26 --- /dev/null +++ b/tests/a2a/test_task_store.py @@ -0,0 +1,81 @@ +"""Bounded task persistence and terminal-state monotonicity.""" + +from __future__ import annotations + +import asyncio + +from a2a.types import TaskState, TaskStatus +from a2a.utils import new_task + +from plugins.platforms.a2a.task_store import BoundedTaskStore + + +def _task(fakes, task_id: str, state: TaskState): + task = new_task(fakes.make_user_message("work", context_id="ctx-store")) + task.id = task_id + task.status = TaskStatus(state=state) + return task + + +def test_terminal_state_cannot_be_resurrected_by_stale_working_event(fakes): + async def exercise(): + store = BoundedTaskStore(max_tasks=10, max_history_messages=10) + await store.save(_task(fakes, "task-1", TaskState.working)) + await store.save(_task(fakes, "task-1", TaskState.canceled)) + await store.save(_task(fakes, "task-1", TaskState.working)) + + persisted = await store.get("task-1") + assert persisted is not None + assert persisted.status.state == TaskState.canceled + + asyncio.run(exercise()) + + +def test_task_count_and_status_history_are_bounded(fakes): + async def exercise(): + store = BoundedTaskStore(max_tasks=2, max_history_messages=2) + first = _task(fakes, "task-1", TaskState.working) + first.history = [ + fakes.make_user_message(str(index), context_id="ctx-store") + for index in range(4) + ] + await store.save(first) + await store.save(_task(fakes, "task-2", TaskState.completed)) + await store.save(_task(fakes, "task-3", TaskState.working)) + + assert await store.size() == 2 + assert await store.get("task-1") is not None + assert await store.get("task-2") is None + + retained = _task(fakes, "task-1", TaskState.working) + retained.history = [ + fakes.make_user_message(str(index), context_id="ctx-store") + for index in range(4) + ] + await store.save(retained) + persisted = await store.get("task-1") + assert persisted is not None + assert [message.parts[0].root.text for message in persisted.history] == [ + "2", + "3", + ] + + asyncio.run(exercise()) + + +def test_all_active_tasks_temporarily_overshoot_then_repair(fakes): + async def exercise(): + store = BoundedTaskStore(max_tasks=2, max_history_messages=2) + for task_id in ("task-1", "task-2", "task-3"): + await store.save(_task(fakes, task_id, TaskState.working)) + + assert await store.size() == 3 + assert await store.get("task-1") is not None + + await store.save(_task(fakes, "task-1", TaskState.completed)) + assert await store.size() == 2 + assert await store.get("task-1") is None + assert await store.get("task-2") is not None + assert await store.get("task-3") is not None + + asyncio.run(exercise()) diff --git a/tests/test_packaging_metadata.py b/tests/test_packaging_metadata.py index f1ccee4773b0..dd16b42e03c9 100644 --- a/tests/test_packaging_metadata.py +++ b/tests/test_packaging_metadata.py @@ -198,8 +198,8 @@ def test_starlette_pinned_above_cve_2026_48710_floor_in_pyproject(): ver = spec.split("==", 1)[1].split(";", 1)[0].strip() found[extra] = ver - # The four server-surface extras must each carry the direct pin. - for extra in ("web", "mcp", "computer-use", "dev"): + # The server-surface extras must each carry the direct pin. + for extra in ("web", "mcp", "computer-use", "dev", "a2a"): assert extra in found, ( f"[{extra}] no longer pins starlette directly — CVE-2026-48710 " f"regression risk (mcp/fastapi pull it transitively with no upper bound)" diff --git a/uv.lock b/uv.lock index 59f9d8e2628b..591a3b7470fd 100644 --- a/uv.lock +++ b/uv.lock @@ -7,6 +7,33 @@ resolution-markers = [ "python_full_version < '3.12'", ] +[options] +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. +exclude-newer-span = "P7D" + +[[package]] +name = "a2a-sdk" +version = "0.3.26" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-api-core" }, + { name = "httpx" }, + { name = "httpx-sse" }, + { name = "protobuf" }, + { name = "pydantic" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/be/97/a6840e01795b182ce751ca165430d46459927cde9bfab838087cbb24aef7/a2a_sdk-0.3.26.tar.gz", hash = "sha256:44068e2d037afbb07ab899267439e9bc7eaa7ac2af94f1e8b239933c993ad52d", size = 274598, upload-time = "2026-04-09T15:21:13.902Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/dd/d5/51f4ee1bf3b736add42a542d3c8a3fd3fa85f3d36c17972127defc46c26f/a2a_sdk-0.3.26-py3-none-any.whl", hash = "sha256:754e0573f6d33b225c1d8d51f640efa69cbbed7bdfb06ce9c3540ea9f58d4a91", size = 151016, upload-time = "2026-04-09T15:21:12.35Z" }, +] + +[package.optional-dependencies] +http-server = [ + { name = "fastapi" }, + { name = "sse-starlette" }, + { name = "starlette" }, +] + [[package]] name = "agent-client-protocol" version = "0.9.0" @@ -1552,10 +1579,15 @@ dependencies = [ ] [package.optional-dependencies] +a2a = [ + { name = "a2a-sdk", extra = ["http-server"] }, + { name = "starlette" }, +] acp = [ { name = "agent-client-protocol" }, ] all = [ + { name = "a2a-sdk", extra = ["http-server"] }, { name = "agent-client-protocol" }, { name = "aiohttp" }, { name = "fastapi" }, @@ -1732,6 +1764,7 @@ youtube = [ [package.metadata] requires-dist = [ + { name = "a2a-sdk", extras = ["http-server"], marker = "extra == 'a2a'", specifier = "==0.3.26" }, { name = "agent-client-protocol", marker = "extra == 'acp'", specifier = "==0.9.0" }, { name = "aiohttp", marker = "extra == 'homeassistant'", specifier = "==3.14.1" }, { name = "aiohttp", marker = "extra == 'matrix'", specifier = "==3.14.1" }, @@ -1769,6 +1802,7 @@ requires-dist = [ { name = "google-auth", marker = "extra == 'vertex'", specifier = "==2.55.1" }, { name = "google-auth-httplib2", marker = "extra == 'google'", specifier = "==0.3.1" }, { name = "google-auth-oauthlib", marker = "extra == 'google'", specifier = "==1.3.1" }, + { name = "hermes-agent", extras = ["a2a"], marker = "extra == 'all'" }, { name = "hermes-agent", extras = ["acp"], marker = "extra == 'all'" }, { name = "hermes-agent", extras = ["acp"], marker = "extra == 'termux'" }, { name = "hermes-agent", extras = ["cli"], marker = "extra == 'all'" }, @@ -1839,6 +1873,7 @@ requires-dist = [ { name = "slack-sdk", marker = "extra == 'messaging'", specifier = "==3.43.0" }, { name = "slack-sdk", marker = "extra == 'slack'", specifier = "==3.43.0" }, { name = "sounddevice", marker = "extra == 'voice'", specifier = "==0.5.5" }, + { name = "starlette", marker = "extra == 'a2a'", specifier = "==1.0.1" }, { name = "starlette", marker = "extra == 'computer-use'", specifier = "==1.0.1" }, { name = "starlette", marker = "extra == 'dev'", specifier = "==1.0.1" }, { name = "starlette", marker = "extra == 'mcp'", specifier = "==1.0.1" }, @@ -1853,7 +1888,7 @@ requires-dist = [ { name = "websockets", specifier = "==15.0.1" }, { name = "youtube-transcript-api", marker = "extra == 'youtube'", specifier = "==1.2.4" }, ] -provides-extras = ["anthropic", "exa", "firecrawl", "parallel-web", "fal", "edge-tts", "modal", "daytona", "hindsight", "dev", "messaging", "cron", "slack", "matrix", "wecom", "cli", "tts-premium", "voice", "pty", "honcho", "supermemory", "mem0", "vision", "mcp", "nemo-relay", "homeassistant", "sms", "teams", "computer-use", "acp", "mistral", "bedrock", "vertex", "azure-identity", "termux", "termux-all", "dingtalk", "feishu", "google", "youtube", "web", "all"] +provides-extras = ["anthropic", "exa", "firecrawl", "parallel-web", "fal", "edge-tts", "modal", "daytona", "hindsight", "dev", "messaging", "cron", "slack", "matrix", "wecom", "cli", "tts-premium", "voice", "pty", "honcho", "supermemory", "mem0", "vision", "mcp", "nemo-relay", "homeassistant", "sms", "teams", "computer-use", "acp", "a2a", "mistral", "bedrock", "vertex", "azure-identity", "termux", "termux-all", "dingtalk", "feishu", "google", "youtube", "web", "all"] [[package]] name = "hf-xet" diff --git a/website/docs/developer-guide/a2a-internals.md b/website/docs/developer-guide/a2a-internals.md new file mode 100644 index 000000000000..eac4fd64c39e --- /dev/null +++ b/website/docs/developer-guide/a2a-internals.md @@ -0,0 +1,134 @@ +--- +sidebar_position: 3 +title: "A2A Internals" +description: "How the A2A adapter works: lifecycle, context sessions, event bridge, and the Agent Card" +--- + +# A2A Internals + +The A2A adapter wraps Hermes' synchronous `AIAgent` in an async JSON-RPC + SSE +HTTP server built on the [`a2a-sdk`](https://a2a-protocol.org). It mirrors the +[ACP adapter](./acp-internals.md): a protocol server class drives the same +`AIAgent` callback seam, just translating events to A2A instead of ACP. + +Key implementation files: + +- `plugins/platforms/a2a/adapter.py` +- `plugins/platforms/a2a/entry.py` +- `plugins/platforms/a2a/card.py` +- `plugins/platforms/a2a/executor.py` +- `plugins/platforms/a2a/sessions.py` +- `plugins/platforms/a2a/events.py` + +## Boot flow + +```text +standalone: hermes-a2a / python -m plugins.platforms.a2a + -> plugins.platforms.a2a.entry.main() + -> parse --version / --check before server startup + -> load ~/.hermes/.env + -> discover MCP tools (tools.mcp_tool.discover_mcp_tools) + -> build_app(): AgentCard + DefaultRequestHandler + BoundedTaskStore + -> A2AStarletteApplication(...).build() + -> uvicorn.run(app) + +gateway: discover plugins -> ctx.register_platform(name="a2a", ...) + -> A2AAdapter.connect() -> uvicorn.Server.serve() +``` + +The Agent Card is served at `/.well-known/agent-card.json`; the JSON-RPC +endpoint is at `/`. + +## Major components + +### `HermesAgentExecutor` + +`plugins/platforms/a2a/executor.py` implements the a2a-sdk `AgentExecutor` +interface (`execute` / `cancel`). + +`execute()`: + +- reads the user text and resolves (or creates) the task via `new_task` +- creates a `TaskUpdater`, marks the task `working` +- resolves the Hermes session for `contextId` +- wires AIAgent callbacks to A2A events +- runs `AIAgent.run_conversation` in a dedicated bounded worker pool +- emits the final response as an artifact, then marks the task `completed` + +`cancel()` signals the session and emits a `canceled` status. + +### `ContextSessionStore` + +`plugins/platforms/a2a/sessions.py` maps `contextId` to a `HermesSession` (an +`AIAgent`, its rolling history, and a cancel event). It is thread-safe, creates +agents lazily, and accepts an `agent_factory` so tests can inject a fake. The +real build mirrors `acp_adapter.session._make_agent` and resolves tools through +the generic `platform_toolsets.a2a` surface instead of adding an A2A-specific +core profile. Global `agent.disabled_toolsets` restrictions are authoritative. + +### Event bridge + +`plugins/platforms/a2a/events.py` converts AIAgent callbacks into `TaskUpdater` +events: + +- `stream_delta_callback` -> `working` status with the text chunk +- `tool_progress_callback` -> `working` status tagged `hermes/kind=tool-call` +- `step_callback` -> `working` status tagged `hermes/kind=tool-result` +- `reasoning_callback` -> `working` status tagged `hermes/kind=reasoning` +- final response -> `add_artifact(...)` + `complete()` + +Because `AIAgent` runs in a worker thread while the A2A event queue lives on the +server event loop, the bridge marshals each async update with: + +```python +asyncio.run_coroutine_threadsafe(...) +``` + +and blocks briefly on it so updates preserve order relative to the agent's own +progress (and all working updates land before the final artifact). A failed +update is logged and swallowed — it never aborts the turn. + +### Agent Card + +`plugins/platforms/a2a/card.py` builds the `AgentCard` dynamically from the +Hermes version plus a curated skill list (general agent, research). Unlike ACP's +checked-in `acp_registry/agent.json`, the A2A card is built at server startup +(no static JSON manifest to keep in sync) and re-serialized on each +`/.well-known/agent-card.json` request. + +## Task lifecycle + +```text +message/send | message/stream + -> DefaultRequestHandler -> HermesAgentExecutor.execute() + -> new_task() (if no current task) -> enqueue Task + -> TaskUpdater.start_work() [status: working] + -> to_thread(AIAgent.run_conversation) + stream_delta / tool_progress / step -> working status updates + -> add_artifact(final_response) [artifact-update] + -> complete() [status: completed] +``` + +## Cancelation + +`cancel()` sets the session cancel event and calls `agent.interrupt()` when +available, then emits a terminal `canceled` status. + +## Current limitations + +- Bounded in-memory task store and sessions: both are lost on process restart. + Terminal states are monotonic so stale working events cannot undo a cancel. +- The endpoint is served unauthenticated; bind `127.0.0.1` or front it with a + proxy/auth layer (see the security note in the user guide). +- Push notifications, persistent task stores, and gRPC / HTTP+JSON transports + are not part of this cut. (`tasks/resubscribe` is routable via the SDK's + default handler but is not exercised by this adapter's tests.) +- Input is text-only; the seam accepts richer parts later. + +## Related files + +- `tests/a2a/` — A2A test suite +- `plugins/platforms/a2a/plugin.yaml` — bundled platform manifest +- `hermes_cli/config.py` — default `a2a` configuration values +- `pyproject.toml` — `[a2a]` optional dependency + `hermes-a2a` script +- `.plans/a2a-protocol.md` — design + protocol-to-Hermes mapping diff --git a/website/docs/integrations/index.md b/website/docs/integrations/index.md index 7f4e1f17679e..04f5cb89c0af 100644 --- a/website/docs/integrations/index.md +++ b/website/docs/integrations/index.md @@ -78,6 +78,10 @@ Speech-to-text supports six providers: local faster-whisper (free, runs on-devic - **[IDE Integration (ACP)](/user-guide/features/acp)** — Use Hermes Agent inside ACP-compatible editors such as VS Code, Zed, and JetBrains. Hermes runs as an ACP server, rendering chat messages, tool activity, file diffs, and terminal commands inside your editor. +## Agent-to-Agent (A2A) + +- **[A2A Server](/user-guide/features/a2a)** — Run Hermes Agent as an [Agent2Agent](https://a2a-protocol.org) server so other agents can discover it (via an Agent Card) and delegate tasks over JSON-RPC + SSE. Turns Hermes into a callable worker for orchestrators like LangGraph, CrewAI, and Google ADK. + ## Programmatic Access - **[API Server](/user-guide/features/api-server)** — Expose Hermes as an OpenAI-compatible HTTP endpoint. Any frontend that speaks the OpenAI format — Open WebUI, LobeChat, LibreChat, NextChat, ChatBox — can connect and use Hermes as a backend with its full toolset. diff --git a/website/docs/user-guide/features/a2a.md b/website/docs/user-guide/features/a2a.md new file mode 100644 index 000000000000..632e4a934bfb --- /dev/null +++ b/website/docs/user-guide/features/a2a.md @@ -0,0 +1,180 @@ +--- +sidebar_position: 12 +title: "A2A (Agent2Agent) Server" +description: "Run Hermes Agent as an A2A server so other agents can discover it and delegate tasks over JSON-RPC + SSE" +--- + +# A2A (Agent2Agent) Server + +Hermes Agent can run as an [A2A](https://a2a-protocol.org) server, letting any +A2A-compatible client or peer agent discover it and delegate tasks over +HTTP(S). Where MCP connects an agent to _tools_, A2A connects an agent to +_other agents_ — so A2A turns Hermes into a callable worker for orchestrators +like LangGraph, CrewAI, Google ADK, the `a2a-inspector`, or another Hermes. + +It is the sibling of [ACP](./acp.md) (editor integration over stdio) and the +MCP server (tools over MCP): A2A is **Hermes as a remote agent for other +agents**. + +## What Hermes exposes in A2A mode + +- An **Agent Card** at `/.well-known/agent-card.json` describing Hermes' name, + version, capabilities (streaming), and skills. +- `message/send` — synchronous request/response (returns a completed task). +- `message/stream` — Server-Sent Events streaming of task status updates and + artifacts. +- `tasks/get` and `tasks/cancel`. + +Each turn composes Hermes' existing coding/research toolsets (shell, +filesystem, web/browser, memory, todo, skills, `execute_code`, and +`delegate_task`) without adding an A2A-specific core toolset or interactive +messaging/audio surfaces. + +## Installation + +Install Hermes normally, then add the A2A extra: + +```bash +pip install -e '.[a2a]' +``` + +This installs the `a2a-sdk[http-server]` dependency and enables: + +- `hermes-a2a` +- `python -m plugins.platforms.a2a` +- the bundled A2A gateway platform plugin + +## Launching the A2A server + +```bash +hermes-a2a +``` + +```bash +python -m plugins.platforms.a2a +``` + +By default the server binds `127.0.0.1:9100`. The Agent Card is served at +`/.well-known/agent-card.json` and the JSON-RPC endpoint at `/`. + +```bash +hermes-a2a --host 127.0.0.1 --port 9100 +hermes-a2a --public-url https://agents.example.com/hermes/ # URL advertised in the card +``` + +For non-interactive checks: + +```bash +hermes-a2a --version +hermes-a2a --check +``` + +:::warning Exposing to the network +The A2A endpoint is **unauthenticated**. Binding `--host 0.0.0.0` exposes +Hermes — and its shell/filesystem tools — to anything that can reach the port. +Only do so behind a reverse proxy or auth layer you control. The server logs a +warning when started on `0.0.0.0`. +::: + +## Talking to the server + +Fetch the card, then send a message. With `curl`: + +```bash +curl http://127.0.0.1:9100/.well-known/agent-card.json + +curl http://127.0.0.1:9100/ -H 'Content-Type: application/json' -d '{ + "jsonrpc": "2.0", "id": "1", "method": "message/send", + "params": {"message": {"role": "user", "kind": "message", "messageId": "m1", + "parts": [{"kind": "text", "text": "Summarize what this repo does."}]}} +}' +``` + +`message/stream` uses the same body with `"method": "message/stream"` (the +method name is what selects streaming; an `Accept: text/event-stream` header is +the conventional client courtesy). The response is an SSE stream of +`TaskStatusUpdateEvent` (working) and `TaskArtifactUpdateEvent` (the result), +ending in a `completed` status. + +## Conversation continuity + +A2A `contextId` maps to a persistent Hermes session: one `AIAgent` plus its +rolling history per context. Follow-up messages that reuse the same `contextId` +continue the same conversation. Each `taskId` is one turn within a context. +Sessions are held in memory for the lifetime of the server process. + +## Configuration and credentials + +A2A mode uses the same Hermes profile configuration as the CLI. Behavioral +settings live in `~/.hermes/config.yaml`: + +```yaml +a2a: + enabled: false # true starts A2A with `hermes gateway` + host: 127.0.0.1 + port: 9100 + public_url: null # externally advertised base URL + max_concurrency: 16 # simultaneous blocking agent turns + max_sessions: 512 # in-memory context LRU cap + max_tasks: 2048 # retained protocol task cap + max_task_history: 100 # retained status messages per task + tool_io: preview # preview | none | full + +platform_toolsets: + a2a: [web, terminal, file, vision, skills, browser, todo, memory, + session_search, code_execution, delegation] +``` + +`tool_io: preview` bounds peer-visible tool arguments/results, `none` sends +tool names only, and `full` sends unbounded values. Command-line host, port, +and public-URL flags override the file for standalone launches. + +`platform_toolsets.a2a` uses the same tool configuration surface as every +gateway platform. `agent.disabled_toolsets` remains authoritative, so an +operator can globally remove sensitive capabilities such as `terminal` or +`file`. Because A2A has no interactive approval round trip, dangerous commands +that require confirmation are denied instead of reading from server stdin. + +The plugin can also be managed by the existing gateway lifecycle: + +```bash +hermes config set a2a.enabled true +hermes gateway run +``` + +Credentials remain in the normal secret store: + +- `~/.hermes/.env` +- `~/.hermes/config.yaml` +- `~/.hermes/skills/` + +Provider resolution uses Hermes' normal runtime resolver, so A2A inherits the +currently configured provider and credentials. Configure credentials with +`hermes model` or by editing `~/.hermes/.env`. + +## Troubleshooting + +### Server starts but tasks fail immediately + +Verify dependencies and provider setup: + +```bash +hermes-a2a --check +hermes model +hermes doctor +``` + +### A client cannot discover the agent + +Confirm the card is reachable and the client points at the base URL (not the +card URL): + +```bash +curl -fsS http://127.0.0.1:9100/.well-known/agent-card.json +``` + +## See also + +- [A2A Internals](../../developer-guide/a2a-internals.md) +- [ACP Editor Integration](./acp.md) +- [Provider Runtime Resolution](../../developer-guide/provider-runtime.md) diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/a2a-internals.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/a2a-internals.md new file mode 100644 index 000000000000..ee17cff42ccb --- /dev/null +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/a2a-internals.md @@ -0,0 +1,111 @@ +--- +sidebar_position: 3 +title: "A2A 内部原理" +description: "A2A 适配器的工作方式：生命周期、context 会话、事件桥接与 Agent Card" +--- + +# A2A 内部原理 + +A2A 适配器将 Hermes 的同步 `AIAgent` 包装成一个基于 [`a2a-sdk`](https://a2a-protocol.org) 的异步 JSON-RPC + SSE HTTP 服务器。它与 [ACP 适配器](./acp-internals.md) 同构：一个协议服务器类驱动相同的 `AIAgent` 回调接缝，只是把事件翻译为 A2A 而非 ACP。 + +关键实现文件： + +- `plugins/platforms/a2a/adapter.py` +- `plugins/platforms/a2a/entry.py` +- `plugins/platforms/a2a/card.py` +- `plugins/platforms/a2a/executor.py` +- `plugins/platforms/a2a/sessions.py` +- `plugins/platforms/a2a/events.py` + +## 启动流程 + +```text +独立：hermes-a2a / python -m plugins.platforms.a2a + -> plugins.platforms.a2a.entry.main() + -> 在服务器启动前解析 --version / --check + -> 加载 ~/.hermes/.env + -> 发现 MCP 工具（tools.mcp_tool.discover_mcp_tools） + -> build_app()：AgentCard + DefaultRequestHandler + BoundedTaskStore + -> A2AStarletteApplication(...).build() + -> uvicorn.run(app) + +网关：发现插件 -> ctx.register_platform(name="a2a", ...) + -> A2AAdapter.connect() -> uvicorn.Server.serve() +``` + +Agent Card 位于 `/.well-known/agent-card.json`；JSON-RPC 端点位于 `/`。 + +## 主要组件 + +### `HermesAgentExecutor` + +`plugins/platforms/a2a/executor.py` 实现 a2a-sdk 的 `AgentExecutor` 接口（`execute` / `cancel`）。 + +`execute()`： + +- 读取用户文本，并通过 `new_task` 解析（或创建）task +- 创建 `TaskUpdater`，将 task 标记为 `working` +- 解析 `contextId` 对应的 Hermes 会话 +- 将 AIAgent 回调连接到 A2A 事件 +- 在专用的有界工作线程池中运行 `AIAgent.run_conversation` +- 把最终响应作为产物发出，然后将 task 标记为 `completed` + +`cancel()` 向会话发出信号并发出 `canceled` 状态。 + +### `ContextSessionStore` + +`plugins/platforms/a2a/sessions.py` 将 `contextId` 映射到一个 `HermesSession`（一个 `AIAgent`、其滚动历史和一个取消事件）。它是线程安全的，惰性创建 agent，并接受 `agent_factory` 以便测试注入伪实现。真实构建会组合现有的编码/研究工具集，而不是增加 A2A 专用的核心 profile。 + +### 事件桥接 + +`plugins/platforms/a2a/events.py` 将 AIAgent 回调转换为 `TaskUpdater` 事件： + +- `stream_delta_callback` -> 带文本块的 `working` 状态 +- `tool_progress_callback` -> 标记 `hermes/kind=tool-call` 的 `working` 状态 +- `step_callback` -> 标记 `hermes/kind=tool-result` 的 `working` 状态 +- `reasoning_callback` -> 标记 `hermes/kind=reasoning` 的 `working` 状态 +- 最终响应 -> `add_artifact(...)` + `complete()` + +由于 `AIAgent` 运行在工作线程中，而 A2A 事件队列存在于服务器事件循环上，桥接使用以下方式编排每个异步更新： + +```python +asyncio.run_coroutine_threadsafe(...) +``` + +并在其上短暂阻塞，从而使更新相对于 agent 自身进度保持有序（且所有 working 更新都在最终产物之前送达）。失败的更新会被记录并吞掉 —— 它绝不会中止该回合。 + +### Agent Card + +`plugins/platforms/a2a/card.py` 根据 Hermes 版本加上精选的技能列表（通用 agent、研究）动态构建 `AgentCard`。与 ACP 签入的 `acp_registry/agent.json` 不同，A2A 的 card 在服务器启动时构建（没有需要保持同步的静态 JSON 清单），并在每次 `/.well-known/agent-card.json` 请求时重新序列化。 + +## 任务生命周期 + +```text +message/send | message/stream + -> DefaultRequestHandler -> HermesAgentExecutor.execute() + -> new_task()（若无当前 task）-> 入队 Task + -> TaskUpdater.start_work() [状态: working] + -> to_thread(AIAgent.run_conversation) + stream_delta / tool_progress / step -> working 状态更新 + -> add_artifact(final_response) [artifact-update] + -> complete() [状态: completed] +``` + +## 取消 + +`cancel()` 设置会话取消事件，并在可用时调用 `agent.interrupt()`，然后发出终态 `canceled` 状态。 + +## 当前限制 + +- 内存内的 task 存储和会话：两者在进程重启时都会丢失。 +- 端点以未认证方式提供；请绑定 `127.0.0.1` 或在代理/认证层之后提供（见用户指南中的安全提示）。 +- 推送通知、持久化 task 存储以及 gRPC / HTTP+JSON 传输不在本次范围内。（`tasks/resubscribe` 可通过 SDK 默认处理器路由，但本适配器的测试未覆盖。） +- 输入为纯文本；该接缝后续可接受更丰富的 part。 + +## 相关文件 + +- `tests/a2a/` —— A2A 测试套件 +- `plugins/platforms/a2a/plugin.yaml` —— 内置平台清单 +- `hermes_cli/config.py` —— 默认 `a2a` 配置值 +- `pyproject.toml` —— `[a2a]` 可选依赖 + `hermes-a2a` 脚本 +- `.plans/a2a-protocol.md` —— 设计与协议到 Hermes 的映射 diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/a2a.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/a2a.md new file mode 100644 index 000000000000..e4c98bdd371b --- /dev/null +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/a2a.md @@ -0,0 +1,148 @@ +--- +sidebar_position: 12 +title: "A2A（Agent2Agent）服务器" +description: "将 Hermes Agent 作为 A2A 服务器运行，让其他 agent 通过 JSON-RPC + SSE 发现它并委派任务" +--- + +# A2A（Agent2Agent）服务器 + +Hermes Agent 可作为 [A2A](https://a2a-protocol.org) 服务器运行，让任何兼容 A2A 的客户端或对等 agent 通过 HTTP(S) 发现它并委派任务。MCP 把 agent 连接到 _工具_，而 A2A 把 agent 连接到 _其他 agent_ —— 因此 A2A 让 Hermes 成为 LangGraph、CrewAI、Google ADK、`a2a-inspector` 乃至另一个 Hermes 等编排器可调用的工作单元。 + +它是 [ACP](./acp.md)（通过 stdio 的编辑器集成）和 MCP 服务器（通过 MCP 暴露工具）的姊妹能力：A2A 是 **作为其他 agent 的远程 agent 的 Hermes**。 + +## Hermes 在 A2A 模式下暴露的内容 + +- 位于 `/.well-known/agent-card.json` 的 **Agent Card**，描述 Hermes 的名称、版本、能力（流式）和技能。 +- `message/send` —— 同步请求/响应（返回已完成的 task）。 +- `message/stream` —— 通过 Server-Sent Events 流式传输 task 状态更新和产物（artifact）。 +- `tasks/get` 和 `tasks/cancel`。 + +每个回合会组合 Hermes 现有的编码/研究工具集（shell、文件系统、网页/浏览器、记忆、待办、skills、`execute_code`、`delegate_task`），无需增加 A2A 专用的核心工具集，也不会引入交互式消息或音频功能。 + +## 安装 + +正常安装 Hermes 后，添加 A2A 扩展： + +```bash +pip install -e '.[a2a]' +``` + +这将安装 `a2a-sdk[http-server]` 依赖并启用： + +- `hermes-a2a` +- `python -m plugins.platforms.a2a` +- 内置的 A2A 网关平台插件 + +## 启动 A2A 服务器 + +```bash +hermes-a2a +``` + +```bash +python -m plugins.platforms.a2a +``` + +默认情况下，服务器绑定 `127.0.0.1:9100`。Agent Card 位于 `/.well-known/agent-card.json`，JSON-RPC 端点位于 `/`。 + +```bash +hermes-a2a --host 127.0.0.1 --port 9100 +hermes-a2a --public-url https://agents.example.com/hermes/ # 在 card 中公布的 URL +``` + +非交互式检查： + +```bash +hermes-a2a --version +hermes-a2a --check +``` + +:::warning 暴露到网络 +A2A 端点是 **未认证的**。绑定 `--host 0.0.0.0` 会把 Hermes —— 及其 shell/文件系统工具 —— 暴露给任何能访问该端口的对象。只能在你掌控的反向代理或认证层之后这样做。服务器在以 `0.0.0.0` 启动时会记录一条警告。 +::: + +## 与服务器通信 + +先获取 card，再发送消息。使用 `curl`： + +```bash +curl http://127.0.0.1:9100/.well-known/agent-card.json + +curl http://127.0.0.1:9100/ -H 'Content-Type: application/json' -d '{ + "jsonrpc": "2.0", "id": "1", "method": "message/send", + "params": {"message": {"role": "user", "kind": "message", "messageId": "m1", + "parts": [{"kind": "text", "text": "总结这个仓库做了什么。"}]}} +}' +``` + +`message/stream` 使用相同的请求体，但 `"method": "message/stream"`（决定流式的是方法名；`Accept: text/event-stream` 头是客户端的惯例性礼貌）。响应是 `TaskStatusUpdateEvent`（working）和 `TaskArtifactUpdateEvent`（结果）的 SSE 流，最终以 `completed` 状态结束。 + +## 会话连续性 + +A2A 的 `contextId` 映射到一个持久的 Hermes 会话：每个 context 对应一个 `AIAgent` 及其滚动历史。复用同一 `contextId` 的后续消息会延续同一段对话。每个 `taskId` 是 context 内的一个回合。会话在服务器进程的生命周期内保存于内存中。 + +## 配置与凭据 + +A2A 模式使用与 CLI 相同的 Hermes profile 配置。行为设置位于 `~/.hermes/config.yaml`： + +```yaml +a2a: + enabled: false # true 时随 `hermes gateway` 启动 + host: 127.0.0.1 + port: 9100 + public_url: null + max_concurrency: 16 + max_sessions: 512 + max_tasks: 2048 + max_task_history: 100 + tool_io: preview # preview | none | full + +platform_toolsets: + a2a: [web, terminal, file, vision, skills, browser, todo, memory, + session_search, code_execution, delegation] +``` + +`tool_io: preview` 会限制对等 agent 可见的工具参数/结果，`none` 仅发送工具名，`full` 发送未截断值。独立启动时，命令行的 host、port 和 public URL 标志优先于配置文件。 + +`platform_toolsets.a2a` 使用与其他网关平台相同的工具配置机制。`agent.disabled_toolsets` 始终优先，因此运维人员可以全局移除 `terminal` 或 `file` 等敏感能力。由于 A2A 没有交互式审批往返，需要确认的危险命令会被拒绝，而不会从服务器标准输入读取。 + +也可以通过现有网关生命周期启动插件： + +```bash +hermes config set a2a.enabled true +hermes gateway run +``` + +凭据仍保存在正常的密钥存储中： + +- `~/.hermes/.env` +- `~/.hermes/config.yaml` +- `~/.hermes/skills/` + +provider 解析使用 Hermes 正常的运行时解析器，因此 A2A 继承当前配置的 provider 和凭据。使用 `hermes model` 或编辑 `~/.hermes/.env` 来配置凭据。 + +## 故障排查 + +### 服务器启动但任务立即失败 + +验证依赖和 provider 设置： + +```bash +hermes-a2a --check +hermes model +hermes doctor +``` + +### 客户端无法发现 agent + +确认 card 可访问，且客户端指向基础 URL（而非 card URL）： + +```bash +curl -fsS http://127.0.0.1:9100/.well-known/agent-card.json +``` + +## 另见 + +- [A2A 内部原理](../../developer-guide/a2a-internals.md) +- [ACP 编辑器集成](./acp.md) +- [Provider 运行时解析](../../developer-guide/provider-runtime.md) diff --git a/website/scripts/generate-llms-txt.py b/website/scripts/generate-llms-txt.py index a34c57792a3d..e7bc0ceb8ecf 100644 --- a/website/scripts/generate-llms-txt.py +++ b/website/scripts/generate-llms-txt.py @@ -106,6 +106,7 @@ ("integrations/providers", "Providers", None), ("user-guide/features/mcp", "MCP (Model Context Protocol)", None), ("user-guide/features/acp", "ACP (Agent Context Protocol)", None), + ("user-guide/features/a2a", "A2A (Agent2Agent Protocol)", None), ("user-guide/features/api-server", "API Server", None), ("user-guide/features/honcho", "Honcho Memory", None), ("user-guide/features/provider-routing", "Provider Routing", None), diff --git a/website/sidebars.ts b/website/sidebars.ts index a6c9d28fd521..47d00ad50f5a 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -672,6 +672,7 @@ const sidebars: SidebarsConfig = { 'integrations/providers', 'user-guide/features/mcp', 'user-guide/features/acp', + 'user-guide/features/a2a', 'user-guide/features/provider-routing', 'user-guide/features/fallback-providers', 'user-guide/features/credential-pools', @@ -765,6 +766,7 @@ const sidebars: SidebarsConfig = { 'developer-guide/tools-runtime', 'developer-guide/browser-supervisor', 'developer-guide/acp-internals', + 'developer-guide/a2a-internals', 'developer-guide/cron-internals', 'developer-guide/trajectory-format', ],