diff --git a/a2a_adapter/__init__.py b/a2a_adapter/__init__.py new file mode 100644 index 0000000000000..abd36263c4082 --- /dev/null +++ b/a2a_adapter/__init__.py @@ -0,0 +1,5 @@ +"""A2A (Agent-to-Agent) protocol server adapter for hermes-agent. + +Exposes Hermes as an A2A-discoverable agent that can receive tasks from +any A2A-compliant client or agent, regardless of framework. +""" diff --git a/a2a_adapter/__main__.py b/a2a_adapter/__main__.py new file mode 100644 index 0000000000000..0a3b33aef88ad --- /dev/null +++ b/a2a_adapter/__main__.py @@ -0,0 +1,6 @@ +"""Allow running with ``python -m a2a_adapter``.""" + +from a2a_adapter.entry import main + +if __name__ == "__main__": + main() diff --git a/a2a_adapter/entry.py b/a2a_adapter/entry.py new file mode 100644 index 0000000000000..27da9ab020d01 --- /dev/null +++ b/a2a_adapter/entry.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +"""Entry point for the Hermes A2A server. + +Starts Hermes as an A2A-discoverable agent on HTTP, serving: + - /.well-known/agent.json — Agent Card (discovery) + - / — JSON-RPC endpoint (task execution) + +Usage: + hermes-a2a # default port 9990 + hermes-a2a --port 8080 # custom port + hermes-a2a --host 127.0.0.1 --port 8080 # bind to localhost only + hermes-a2a --name "My Agent" # custom agent name +""" + +import argparse +import logging +import os +import sys + + +def _setup_logging(): + """Configure logging to stderr so stdout stays clean for HTTP.""" + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter( + logging.Formatter("%(asctime)s [%(name)s] %(levelname)s: %(message)s") + ) + root = logging.getLogger() + root.addHandler(handler) + root.setLevel(logging.INFO) + + +def _load_env(): + """Load environment from ~/.hermes/.env if available.""" + try: + from hermes_cli.env_loader import load_dotenv + load_dotenv() + except ImportError: + pass + + +def main(): + """Start the Hermes A2A server.""" + parser = argparse.ArgumentParser( + description="Run Hermes as an A2A-discoverable agent server" + ) + parser.add_argument( + "--host", default="127.0.0.1", + help="Host to bind to (default: 127.0.0.1). Use 0.0.0.0 for remote access (requires --bearer-token)." + ) + parser.add_argument( + "--port", type=int, default=9990, + help="Port to listen on (default: 9990)" + ) + parser.add_argument( + "--name", default="Hermes Agent", + help="Agent name in the Agent Card (default: 'Hermes Agent')" + ) + parser.add_argument( + "--bearer-token", + default=os.environ.get("HERMES_A2A_BEARER_TOKEN"), + help="Require this bearer token for all requests (env: HERMES_A2A_BEARER_TOKEN). " + "Strongly recommended when binding to 0.0.0.0.", + ) + parser.add_argument( + "--toolset", default="hermes-cli", + help="Toolset for agent sessions (default: hermes-cli). Use 'hermes-acp' for restricted access.", + ) + args = parser.parse_args() + + _setup_logging() + _load_env() + + logger = logging.getLogger(__name__) + + # Security: warn loudly if binding to all interfaces without auth + if args.host == "0.0.0.0" and not args.bearer_token: + logger.warning( + "WARNING: Binding to 0.0.0.0 WITHOUT --bearer-token. " + "This exposes full agent access (terminal, files, code execution) " + "to anyone who can reach this port. Set --bearer-token or use " + "HERMES_A2A_BEARER_TOKEN env var to require authentication." + ) + + try: + from a2a_adapter.server import build_application, is_server_available + except ImportError: + logger.error( + "A2A server dependencies not installed. " + "Install with: pip install 'hermes-agent[a2a]' or pip install 'a2a-sdk[http-server]'" + ) + sys.exit(1) + + if not is_server_available(): + logger.error( + "A2A SDK server components not available. " + "Install with: pip install 'a2a-sdk[http-server]'" + ) + sys.exit(1) + + try: + import uvicorn + except ImportError: + logger.error( + "uvicorn not installed. Install with: pip install uvicorn[standard]" + ) + sys.exit(1) + + logger.info("Starting Hermes A2A server on %s:%d", args.host, args.port) + if args.bearer_token: + logger.info("Bearer token authentication ENABLED") + logger.info("Agent Card: http://%s:%d/.well-known/agent.json", args.host, args.port) + + app = build_application( + host=args.host, port=args.port, name=args.name, + bearer_token=args.bearer_token, toolset=args.toolset, + ) + uvicorn.run(app.build(), host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/a2a_adapter/server.py b/a2a_adapter/server.py new file mode 100644 index 0000000000000..9c22d66fe1331 --- /dev/null +++ b/a2a_adapter/server.py @@ -0,0 +1,403 @@ +"""A2A Server — Expose Hermes as an A2A-discoverable agent. + +Implements the A2A AgentExecutor interface to handle incoming tasks from +remote A2A clients. Each task gets its own AIAgent session, and results +are streamed back via the A2A event queue. + +Architecture: + - A2AStarletteApplication serves the Agent Card and JSON-RPC endpoints + - HermesAgentExecutor wraps AIAgent to handle A2A task execution + - AIAgent runs synchronously in a thread pool (same pattern as ACP adapter) + - Task updates are streamed via EventQueue +""" + +import asyncio +import logging +from concurrent.futures import ThreadPoolExecutor +from typing import Optional + +from a2a_adapter.session import SessionManager + +logger = logging.getLogger(__name__) + +# Thread pool for running sync AIAgent (same pattern as ACP adapter) +_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="a2a-agent") + +# --------------------------------------------------------------------------- +# Graceful imports — a2a-sdk[http-server] is an optional dependency +# --------------------------------------------------------------------------- + +try: + from a2a.server.agent_execution import AgentExecutor, RequestContext + from a2a.server.events import EventQueue + from a2a.server.apps import A2AStarletteApplication + from a2a.server.request_handlers import DefaultRequestHandler + from a2a.server.tasks import InMemoryTaskStore + from a2a.types import ( + AgentCapabilities, + AgentCard, + AgentSkill, + TaskArtifactUpdateEvent, + TaskState, + TaskStatus, + TaskStatusUpdateEvent, + ) + from a2a.utils.artifact import new_text_artifact + from a2a.utils.message import new_agent_text_message + from a2a.utils.task import new_task + + _A2A_SERVER_AVAILABLE = True +except ImportError as e: + _A2A_SERVER_AVAILABLE = False + logger.debug("A2A server dependencies not available: %s", e) + + +def is_server_available() -> bool: + """Check if A2A server dependencies are installed.""" + return _A2A_SERVER_AVAILABLE + + +# --------------------------------------------------------------------------- +# Agent Card builder +# --------------------------------------------------------------------------- + +def build_agent_card( + host: str = "127.0.0.1", + port: int = 9990, + name: str = "Hermes Agent", + version: str = "0.5.0", +) -> "AgentCard": + """Build an A2A Agent Card describing Hermes' capabilities. + + The card is served at /.well-known/agent.json and tells remote agents + what Hermes can do. + """ + if not _A2A_SERVER_AVAILABLE: + raise ImportError("a2a-sdk[http-server] is required for A2A server mode") + + # 0.0.0.0 is a bind address, not reachable — resolve to localhost for the card + card_host = "127.0.0.1" if host == "0.0.0.0" else host + url = f"http://{card_host}:{port}" + + skills = [ + AgentSkill( + id="general_assistant", + name="General Assistant", + description=( + "General-purpose AI agent with access to terminal, file system, " + "web search, browser automation, code execution, and more. " + "Can handle software engineering, research, analysis, and automation tasks." + ), + tags=["coding", "research", "automation", "analysis", "devops"], + examples=[ + "Write a Python script that fetches weather data", + "Search the web for recent news about AI agents", + "Read and analyze the files in /tmp/project", + "Debug this error message: ...", + ], + ), + AgentSkill( + id="code_execution", + name="Code Execution", + description="Execute Python code in a sandboxed environment and return results.", + tags=["python", "code", "execution"], + examples=[ + "Run this Python code: print('hello')", + "Calculate the factorial of 100", + ], + ), + AgentSkill( + id="web_research", + name="Web Research", + description="Search the web and extract content from URLs for research tasks.", + tags=["search", "web", "research"], + examples=[ + "Find the latest documentation for FastAPI", + "What are the top GitHub repos for agent frameworks?", + ], + ), + AgentSkill( + id="file_operations", + name="File Operations", + description="Read, write, search, and patch files on the local filesystem.", + tags=["files", "filesystem", "read", "write"], + examples=[ + "Read the file at /tmp/data.json", + "Search for files containing 'TODO' in the current directory", + ], + ), + ] + + return AgentCard( + name=name, + description=( + "Hermes Agent — a self-improving AI agent that creates skills from experience, " + "improves them during use, and runs anywhere. Supports terminal, file system, " + "web search, browser automation, code execution, delegation, and more." + ), + url=url, + version=version, + default_input_modes=["text"], + default_output_modes=["text"], + capabilities=AgentCapabilities(streaming=True), + skills=skills, + ) + + +# --------------------------------------------------------------------------- +# Agent Executor — bridges A2A tasks to Hermes AIAgent +# --------------------------------------------------------------------------- + +if _A2A_SERVER_AVAILABLE: + + class HermesAgentExecutor(AgentExecutor): + """Execute A2A tasks using Hermes AIAgent. + + Each task creates or reuses a session with its own AIAgent instance. + The agent runs synchronously in a thread pool, and results are + streamed back via the A2A event queue. + """ + + def __init__(self, session_manager: Optional[SessionManager] = None): + self.session_manager = session_manager or SessionManager() + + async def execute( + self, + context: RequestContext, + event_queue: EventQueue, + ) -> None: + """Handle an incoming A2A task.""" + # Extract user message text + user_text = self._extract_text(context) + if not user_text: + await self._send_error(context, event_queue, "No text content in message") + return + + # Create or get session for this task (supports multi-turn) + task_id = context.task_id + session = self.session_manager.get_or_create_session(task_id) + + # Emit task + WORKING status + task = context.current_task or new_task(context.message) + await event_queue.enqueue_event(task) + + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=context.task_id, + context_id=context.context_id, + status=TaskStatus( + state=TaskState.working, + message=new_agent_text_message("Processing request..."), + ), + ) + ) + + # Run AIAgent in thread pool (sync agent, async server) + loop = asyncio.get_running_loop() + try: + result = await loop.run_in_executor( + _executor, + lambda: self._run_agent(session, user_text, task_id=task_id), + ) + + # Extract response text + response_text = result.get("final_response", "") + if not response_text: + response_text = "(Agent produced no response)" + + # Emit artifact with result + await event_queue.enqueue_event( + TaskArtifactUpdateEvent( + task_id=context.task_id, + context_id=context.context_id, + artifact=new_text_artifact( + name="response", + text=response_text, + ), + ) + ) + + # Mark completed + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=context.task_id, + context_id=context.context_id, + status=TaskStatus( + state=TaskState.completed, + ), + ) + ) + + except Exception as e: + logger.exception("A2A task execution failed: %s", e) + await self._send_error(context, event_queue, str(e)) + + async def cancel( + self, + context: RequestContext, + event_queue: EventQueue, + ) -> None: + """Cancel a running task. + + Sets the cancel event AND calls agent.interrupt() to actually + stop in-progress terminal/code execution (matching ACP pattern). + """ + task_id = context.task_id + # Signal cancellation + cancelled = self.session_manager.cancel_session(task_id) + + # Also interrupt the agent to stop running tools + session = self.session_manager.get_session(task_id) + if session and getattr(session, "agent", None): + try: + if hasattr(session.agent, "interrupt"): + session.agent.interrupt() + except Exception: + logger.debug("Failed to interrupt A2A session %s", task_id, exc_info=True) + + state = ( + TaskState.canceled + if cancelled + else TaskState.failed + ) + msg = "Task cancelled" if cancelled else "Task not found or already completed" + + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=context.task_id, + context_id=context.context_id, + status=TaskStatus( + state=state, + message=new_agent_text_message(msg), + ), + ) + ) + + def _extract_text(self, context: RequestContext) -> str: + """Extract text from the incoming A2A message.""" + message = context.message + if not message: + return "" + + parts = getattr(message, "parts", []) + texts = [] + for part in parts: + # Handle protobuf Part (has .text field directly) + text = getattr(part, "text", None) + if text: + texts.append(text) + continue + # Handle pydantic Part (has .root.text) + root = getattr(part, "root", None) + if root: + text = getattr(root, "text", None) + if text: + texts.append(text) + + return "\n".join(texts) + + def _run_agent(self, session: "SessionState", user_text: str, task_id: str = "") -> dict: + """Run AIAgent synchronously (called from thread pool).""" + agent = session.agent + result = agent.run_conversation( + user_message=user_text, + conversation_history=session.history, + task_id=task_id, + ) + + # Replace history (not extend) to avoid duplicating prior context + if result.get("messages"): + session.history = result["messages"] + + return result + + async def _send_error( + self, + context: RequestContext, + event_queue: EventQueue, + error_msg: str, + ) -> None: + """Send a FAILED status update.""" + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=context.task_id, + context_id=context.context_id, + status=TaskStatus( + state=TaskState.failed, + message=new_agent_text_message(f"Error: {error_msg}"), + ), + ) + ) + + +# --------------------------------------------------------------------------- +# Application builder +# --------------------------------------------------------------------------- + +def _make_bearer_middleware(bearer_token: str): + """Create Starlette middleware that validates bearer tokens. + + Agent Card discovery (/.well-known/agent.json) is always open so + remote clients can discover the agent before authenticating. + """ + from starlette.middleware.base import BaseHTTPMiddleware + from starlette.responses import JSONResponse + + class BearerAuthMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request, call_next): + # Allow unauthenticated Agent Card discovery + if request.url.path == "/.well-known/agent.json": + return await call_next(request) + + auth = request.headers.get("authorization", "") + if not auth.startswith("Bearer ") or auth[7:] != bearer_token: + return JSONResponse( + {"error": "Unauthorized — provide a valid Bearer token"}, + status_code=401, + ) + return await call_next(request) + + return BearerAuthMiddleware + + +def build_application( + host: str = "127.0.0.1", + port: int = 9990, + name: str = "Hermes Agent", + bearer_token: Optional[str] = None, + toolset: str = "hermes-cli", +) -> "A2AStarletteApplication": + """Build the A2A Starlette application ready to run with uvicorn.""" + if not _A2A_SERVER_AVAILABLE: + raise ImportError( + "A2A server dependencies not installed. " + "Install with: pip install 'hermes-agent[a2a]' or pip install 'a2a-sdk[http-server]'" + ) + + card = build_agent_card(host=host, port=port, name=name) + session_manager = SessionManager(toolset=toolset) + executor = HermesAgentExecutor(session_manager=session_manager) + handler = DefaultRequestHandler( + agent_executor=executor, + task_store=InMemoryTaskStore(), + ) + + app = A2AStarletteApplication( + agent_card=card, + http_handler=handler, + ) + + if bearer_token: + middleware_cls = _make_bearer_middleware(bearer_token) + # Add middleware before build() so it's included in the middleware + # stack compilation. Patch build() to inject the middleware. + original_build = app.build + + def _build_with_auth(): + starlette_app = original_build() + starlette_app.add_middleware(middleware_cls) + return starlette_app + + app.build = _build_with_auth + + return app diff --git a/a2a_adapter/session.py b/a2a_adapter/session.py new file mode 100644 index 0000000000000..0b0cc9af4769b --- /dev/null +++ b/a2a_adapter/session.py @@ -0,0 +1,143 @@ +"""Session management for the A2A server adapter. + +Each A2A task maps to a Hermes session with its own AIAgent instance. +Thread-safe via threading.Lock, matching the ACP adapter pattern. +""" + +import logging +import threading +import uuid +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class SessionState: + """State for a single A2A task/session.""" + + session_id: str + agent: Any # AIAgent instance + cwd: str = "." + model: str = "" + history: List[Dict[str, Any]] = field(default_factory=list) + cancel_event: Any = None # threading.Event + + +class SessionManager: + """Thread-safe session lifecycle management for A2A tasks.""" + + def __init__(self, toolset: str = "hermes-cli"): + self._sessions: Dict[str, SessionState] = {} + self._lock = threading.Lock() + self._toolset = toolset + + def create_session(self, cwd: str = ".", model: str = "") -> SessionState: + """Create a new session with a fresh AIAgent.""" + session_id = uuid.uuid4().hex + cancel_event = threading.Event() + agent = self._make_agent( + session_id=session_id, cwd=cwd, model=model, + ) + state = SessionState( + session_id=session_id, + agent=agent, + cwd=cwd, + model=model, + cancel_event=cancel_event, + ) + with self._lock: + self._sessions[session_id] = state + logger.info("Created A2A session %s", session_id) + return state + + def get_session(self, session_id: str) -> Optional[SessionState]: + """Thread-safe session lookup.""" + with self._lock: + return self._sessions.get(session_id) + + def get_or_create_session(self, task_id: str, cwd: str = ".") -> SessionState: + """Get existing session for a task or create a new one. + + Sessions are keyed by task_id only (not also by session_id) to + avoid dual-storage where list_sessions() returns duplicates. + """ + with self._lock: + if task_id in self._sessions: + return self._sessions[task_id] + + # Build session outside lock (_make_agent is slow) + cancel_event = threading.Event() + agent = self._make_agent(session_id=task_id, cwd=cwd) + state = SessionState( + session_id=task_id, + agent=agent, + cwd=cwd, + cancel_event=cancel_event, + ) + + with self._lock: + # Double-check: another thread may have created it while we were + # building the agent (TOCTOU guard) + if task_id in self._sessions: + # Discard the agent we just built to avoid resource leaks + logger.debug("Discarding duplicate agent for task %s (race)", task_id) + return self._sessions[task_id] + self._sessions[task_id] = state + logger.info("Created A2A session for task %s", task_id) + return state + + def cancel_session(self, session_id: str) -> bool: + """Signal a session to cancel.""" + state = self.get_session(session_id) + if state and state.cancel_event: + state.cancel_event.set() + logger.info("Cancelled A2A session %s", session_id) + return True + return False + + def list_sessions(self) -> List[dict]: + """Return lightweight session info dicts.""" + with self._lock: + return [ + { + "session_id": s.session_id, + "cwd": s.cwd, + "model": s.model, + "history_length": len(s.history), + } + for s in self._sessions.values() + ] + + def _make_agent(self, *, session_id: str, cwd: str, model: str = "") -> Any: + """Factory: create an AIAgent for a session.""" + from run_agent import AIAgent + from hermes_cli.config import load_config + + config = load_config() + model_cfg = config.get("model") + default_model = model_cfg if model_cfg else "anthropic/claude-sonnet-4-20250514" + + kwargs = { + "platform": "a2a", + "enabled_toolsets": [self._toolset], + "quiet_mode": True, + "session_id": session_id, + "model": model or default_model, + } + + # Resolve runtime provider for API credentials + try: + from hermes_cli.runtime_provider import resolve_runtime_provider + runtime = resolve_runtime_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"), + }) + except Exception as e: + logger.warning("Failed to resolve runtime provider: %s", e) + + return AIAgent(**kwargs) diff --git a/pyproject.toml b/pyproject.toml index a58e172795e67..ce7854fff6d9c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,6 +65,7 @@ mcp = ["mcp>=1.2.0,<2"] homeassistant = ["aiohttp>=3.9.0,<4"] sms = ["aiohttp>=3.9.0,<4"] acp = ["agent-client-protocol>=0.9.0,<1.0"] +a2a = ["a2a-sdk[http-server]>=0.3.24"] mistral = ["mistralai>=2.3.0,<3"] bedrock = ["boto3>=1.35.0,<2"] termux = [ @@ -120,6 +121,7 @@ all = [ "hermes-agent[homeassistant]", "hermes-agent[sms]", "hermes-agent[acp]", + "hermes-agent[a2a]", "hermes-agent[voice]", "hermes-agent[dingtalk]", "hermes-agent[feishu]", @@ -133,6 +135,7 @@ all = [ hermes = "hermes_cli.main:main" hermes-agent = "run_agent:main" hermes-acp = "acp_adapter.entry:main" +hermes-a2a = "a2a_adapter.entry:main" [tool.setuptools] py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajectory_compressor", "toolset_distributions", "cli", "hermes_constants", "hermes_state", "hermes_time", "hermes_logging", "rl_cli", "utils"] @@ -141,7 +144,7 @@ py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajector hermes_cli = ["web_dist/**/*"] [tool.setuptools.packages.find] -include = ["agent", "agent.*", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "tui_gateway", "tui_gateway.*", "cron", "acp_adapter", "plugins", "plugins.*"] +include = ["agent", "agent.*", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "tui_gateway", "tui_gateway.*", "cron", "acp_adapter", "a2a_adapter", "plugins", "plugins.*"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/tools/test_a2a_integration.py b/tests/tools/test_a2a_integration.py new file mode 100644 index 0000000000000..21b413e64b382 --- /dev/null +++ b/tests/tools/test_a2a_integration.py @@ -0,0 +1,166 @@ +"""Integration tests for the A2A protocol — real HTTP roundtrip. + +Starts a real uvicorn server in a background thread, then uses the actual +client tools (a2a_discover, a2a_call) to hit it over HTTP. The AIAgent +is mocked so no LLM keys are needed, but the HTTP transport is real. + +All tests are marked ``@pytest.mark.integration`` and skipped by default. +""" + +import json +import socket +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _free_port() -> int: + """Find an available TCP port.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _server_deps_available() -> bool: + """Check if both server and client deps are installed.""" + try: + from a2a_adapter.server import is_server_available + if not is_server_available(): + return False + import uvicorn # noqa: F401 + from tools.a2a_tool import check_a2a_available + return check_a2a_available() + except ImportError: + return False + + +_skip_reason = "a2a-sdk[http-server], uvicorn, or a2a client deps not installed" + + +def _make_mock_agent(): + """Create a mock AIAgent whose run_conversation returns a canned response.""" + agent = MagicMock() + agent.run_conversation.return_value = { + "final_response": "Integration test response from Hermes", + "messages": [{"role": "assistant", "content": "Integration test response from Hermes"}], + } + return agent + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="module") +def a2a_server(): + """Start a real A2A server on a random port with a mocked AIAgent. + + Yields the base URL (e.g. ``http://127.0.0.1:PORT``). + """ + if not _server_deps_available(): + pytest.skip(_skip_reason) + + import uvicorn + from a2a_adapter.server import build_application + + port = _free_port() + host = "127.0.0.1" + + # Patch SessionManager._make_agent so no real LLM keys are needed + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = _make_mock_agent() + + app = build_application(host=host, port=port) + starlette_app = app.build() + + config = uvicorn.Config(starlette_app, host=host, port=port, log_level="warning") + server = uvicorn.Server(config) + + thread = threading.Thread(target=server.run, daemon=True, name="a2a-test-server") + thread.start() + + # Wait for the server to be ready + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + with socket.create_connection((host, port), timeout=0.5): + break + except OSError: + time.sleep(0.1) + else: + pytest.fail(f"A2A test server did not start on {host}:{port} within 10s") + + yield f"http://{host}:{port}" + + server.should_exit = True + thread.join(timeout=5) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +@pytest.mark.integration +@pytest.mark.skipif(not _server_deps_available(), reason=_skip_reason) +class TestA2AIntegration: + """Full HTTP roundtrip: start server -> discover -> call.""" + + def test_discover_roundtrip(self, a2a_server): + """Discover the live A2A server and verify the Agent Card fields.""" + import tools.a2a_tool as mod + + # Clear any cached cards so we actually hit the network + mod._agent_cards.clear() + mod._config_cache = None + + result = json.loads(mod.a2a_discover({"agent": a2a_server})) + + assert "error" not in result, f"Discovery failed: {result}" + assert result["name"] == "Hermes Agent" + assert "skills" in result + assert len(result["skills"]) >= 1 + assert result["capabilities"]["streaming"] is True + + def test_call_roundtrip(self, a2a_server): + """Send a message to the live A2A server and get a response.""" + import tools.a2a_tool as mod + + # Ensure the agent card is cached (discover first) + mod._agent_cards.clear() + mod._config_cache = None + mod.a2a_discover({"agent": a2a_server}) + + raw = mod.a2a_call({ + "agent": a2a_server, + "message": "Say hello for the integration test", + }) + result = json.loads(raw) + + assert "error" not in result, f"Call failed: {result}" + # The response should contain our mocked text or a valid status + assert result.get("status") in ("completed", "unknown", None) or "response" in result + + def test_discover_then_call(self, a2a_server): + """Full flow: discover agent, then call it — verifies card caching works.""" + import tools.a2a_tool as mod + + mod._agent_cards.clear() + mod._config_cache = None + + # Step 1: discover + card = json.loads(mod.a2a_discover({"agent": a2a_server})) + assert card["name"] == "Hermes Agent" + + # Step 2: call (should reuse cached card, no extra discovery request) + raw = mod.a2a_call({ + "agent": a2a_server, + "message": "Integration roundtrip test", + }) + result = json.loads(raw) + assert "error" not in result, f"Call after discover failed: {result}" diff --git a/tests/tools/test_a2a_server.py b/tests/tools/test_a2a_server.py new file mode 100644 index 0000000000000..40c5f7536f904 --- /dev/null +++ b/tests/tools/test_a2a_server.py @@ -0,0 +1,290 @@ +"""Tests for the A2A server adapter. + +All tests use mocks -- no real servers are started. +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helper to check if server deps are available +# --------------------------------------------------------------------------- + +def _server_available() -> bool: + try: + from a2a_adapter.server import is_server_available + return is_server_available() + except ImportError: + return False + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_mock_message(text="Hello agent"): + """Create a mock A2A message with text parts.""" + part = SimpleNamespace(text=text, root=None) + return SimpleNamespace(parts=[part]) + + +def _make_mock_context(task_id="task-123", context_id="ctx-456", text="Hello agent"): + """Create a mock RequestContext.""" + message = _make_mock_message(text) + return SimpleNamespace( + task_id=task_id, + context_id=context_id, + message=message, + current_task=None, + ) + + +# --------------------------------------------------------------------------- +# Server availability +# --------------------------------------------------------------------------- + +class TestServerAvailability: + def test_is_server_available_returns_bool(self): + """is_server_available returns a boolean.""" + from a2a_adapter.server import is_server_available + result = is_server_available() + assert isinstance(result, bool) + + +# --------------------------------------------------------------------------- +# Agent Card builder +# --------------------------------------------------------------------------- + +class TestBuildAgentCard: + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_card_has_required_fields(self): + """Agent Card contains name, description, skills, capabilities.""" + from a2a_adapter.server import build_agent_card + card = build_agent_card(host="localhost", port=9990) + assert card.name == "Hermes Agent" + assert "self-improving" in card.description.lower() or "hermes" in card.description.lower() + assert len(card.skills) >= 1 + assert card.capabilities.streaming is True + + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_card_custom_name(self): + """Custom name is reflected in the card.""" + from a2a_adapter.server import build_agent_card + card = build_agent_card(name="My Custom Agent") + assert card.name == "My Custom Agent" + + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_card_url_uses_host_port(self): + """Card URL reflects host:port.""" + from a2a_adapter.server import build_agent_card + card = build_agent_card(host="127.0.0.1", port=8080) + assert "127.0.0.1:8080" in card.url + + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_card_url_resolves_wildcard(self): + """0.0.0.0 bind address is resolved to 127.0.0.1 in the Agent Card.""" + from a2a_adapter.server import build_agent_card + card = build_agent_card(host="0.0.0.0", port=9990) + assert "0.0.0.0" not in card.url + assert "127.0.0.1:9990" in card.url + + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_card_skills_have_ids(self): + """Each skill has an id, name, and description.""" + from a2a_adapter.server import build_agent_card + card = build_agent_card() + for skill in card.skills: + assert skill.id + assert skill.name + assert skill.description + + +# --------------------------------------------------------------------------- +# Session Manager +# --------------------------------------------------------------------------- + +class TestSessionManager: + def test_create_session(self): + """Creating a session returns a SessionState with an agent.""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager() + state = mgr.create_session(cwd="/tmp") + assert state.session_id + assert state.agent is not None + assert state.cwd == "/tmp" + + def test_get_session(self): + """Can retrieve a session by ID.""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager() + state = mgr.create_session() + retrieved = mgr.get_session(state.session_id) + assert retrieved is state + + def test_get_nonexistent_session(self): + """Getting a nonexistent session returns None.""" + from a2a_adapter.session import SessionManager + mgr = SessionManager() + assert mgr.get_session("nonexistent") is None + + def test_get_or_create_session(self): + """get_or_create creates on first call, returns same on second.""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager() + s1 = mgr.get_or_create_session("task-1") + s2 = mgr.get_or_create_session("task-1") + assert s1.session_id == s2.session_id + assert mock_make.call_count == 1 # Only created once + + def test_cancel_session(self): + """Cancelling a session sets the cancel event.""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager() + state = mgr.create_session() + assert not state.cancel_event.is_set() + result = mgr.cancel_session(state.session_id) + assert result is True + assert state.cancel_event.is_set() + + def test_cancel_nonexistent(self): + """Cancelling nonexistent session returns False.""" + from a2a_adapter.session import SessionManager + mgr = SessionManager() + assert mgr.cancel_session("nonexistent") is False + + def test_list_sessions(self): + """list_sessions returns info dicts.""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager() + mgr.create_session(cwd="/tmp") + mgr.create_session(cwd="/home") + sessions = mgr.list_sessions() + assert len(sessions) == 2 + assert all("session_id" in s for s in sessions) + + def test_get_or_create_no_dual_storage(self): + """get_or_create_session stores exactly one entry per task (no duplicates).""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager() + mgr.get_or_create_session("task-123") + sessions = mgr.list_sessions() + assert len(sessions) == 1, f"Expected 1 session, got {len(sessions)}" + + def test_session_manager_accepts_toolset(self): + """SessionManager passes custom toolset to _make_agent.""" + from a2a_adapter.session import SessionManager + with patch("a2a_adapter.session.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + mgr = SessionManager(toolset="hermes-acp") + assert mgr._toolset == "hermes-acp" + + +# --------------------------------------------------------------------------- +# HermesAgentExecutor +# --------------------------------------------------------------------------- + +class TestHermesAgentExecutor: + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_extract_text_from_message(self): + """Text extraction works from protobuf-style parts.""" + from a2a_adapter.server import HermesAgentExecutor + executor = HermesAgentExecutor() + context = _make_mock_context(text="Hello world") + result = executor._extract_text(context) + assert result == "Hello world" + + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_extract_text_empty_message(self): + """Empty message returns empty string.""" + from a2a_adapter.server import HermesAgentExecutor + executor = HermesAgentExecutor() + context = SimpleNamespace( + task_id="t1", context_id="c1", message=None, current_task=None, + ) + result = executor._extract_text(context) + assert result == "" + + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_extract_text_pydantic_parts(self): + """Text extraction works from pydantic-style parts (root.text).""" + from a2a_adapter.server import HermesAgentExecutor + executor = HermesAgentExecutor() + part = SimpleNamespace(text=None, root=SimpleNamespace(text="Pydantic text")) + message = SimpleNamespace(parts=[part]) + context = SimpleNamespace( + task_id="t1", context_id="c1", message=message, current_task=None, + ) + result = executor._extract_text(context) + assert result == "Pydantic text" + + +# --------------------------------------------------------------------------- +# Application builder +# --------------------------------------------------------------------------- + +class TestBuildApplication: + @pytest.mark.skipif( + not _server_available(), + reason="a2a-sdk[http-server] not installed", + ) + def test_build_application_returns_app(self): + """build_application returns an A2AStarletteApplication.""" + with patch("a2a_adapter.server.SessionManager._make_agent") as mock_make: + mock_make.return_value = MagicMock() + from a2a_adapter.server import build_application + app = build_application(host="localhost", port=9990) + assert app is not None + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + +class TestEntryPoint: + def test_import_entry(self): + """Entry module is importable.""" + import a2a_adapter.entry + assert hasattr(a2a_adapter.entry, "main") + + def test_import_main(self): + """__main__ module is importable.""" + import a2a_adapter.__main__ diff --git a/tests/tools/test_a2a_tool.py b/tests/tools/test_a2a_tool.py new file mode 100644 index 0000000000000..fb7ebbb515bf2 --- /dev/null +++ b/tests/tools/test_a2a_tool.py @@ -0,0 +1,701 @@ +"""Tests for the A2A (Agent-to-Agent) protocol client support. + +All tests use mocks -- no real A2A servers are contacted. +""" + +import json +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_agent_card(name="test-agent", description="A test agent", url="http://localhost:9999"): + """Create a fake AgentCard object.""" + skill = SimpleNamespace( + id="skill-1", + name="test-skill", + description="A test skill", + ) + capabilities = SimpleNamespace( + streaming=True, + pushNotifications=False, + ) + card = SimpleNamespace( + name=name, + description=description, + url=url, + skills=[skill], + capabilities=capabilities, + securitySchemes={"bearer": {"type": "http", "scheme": "bearer"}}, + ) + return card + + +def _make_send_response(text="Hello from remote agent", status="completed"): + """Create a fake SendMessageResponse.""" + text_part = SimpleNamespace(text=text, root=SimpleNamespace(text=text)) + message = SimpleNamespace(parts=[text_part]) + task_status = SimpleNamespace(state=status, message=message) + artifact = SimpleNamespace(parts=[text_part]) + task = SimpleNamespace(status=task_status, artifacts=[artifact]) + root = SimpleNamespace(result=task, error=None) + return SimpleNamespace(root=root) + + +# --------------------------------------------------------------------------- +# Availability check +# --------------------------------------------------------------------------- + +class TestA2AAvailability: + def test_check_returns_bool(self): + """check_a2a_available returns a boolean.""" + from tools.a2a_tool import check_a2a_available + result = check_a2a_available() + assert isinstance(result, bool) + + +# --------------------------------------------------------------------------- +# Config resolution +# --------------------------------------------------------------------------- + +class TestResolveAgentURL: + def test_direct_url_http(self): + """Direct HTTP URL passes through.""" + from tools.a2a_tool import _resolve_agent_url + with patch("tools.a2a_tool._load_a2a_config", return_value={}): + url, config = _resolve_agent_url("http://localhost:9999") + assert url == "http://localhost:9999" + assert config == {} + + def test_direct_url_https(self): + """Direct HTTPS URL passes through.""" + from tools.a2a_tool import _resolve_agent_url + with patch("tools.a2a_tool._load_a2a_config", return_value={}): + url, config = _resolve_agent_url("https://agent.example.com") + assert url == "https://agent.example.com" + + def test_trailing_slash_stripped(self): + """Trailing slash is removed from URLs.""" + from tools.a2a_tool import _resolve_agent_url + with patch("tools.a2a_tool._load_a2a_config", return_value={}): + url, _ = _resolve_agent_url("http://localhost:9999/") + assert url == "http://localhost:9999" + + def test_config_name_resolved(self): + """Named agent from config resolves to its URL.""" + config = { + "researcher": { + "url": "http://researcher.local:8080", + "auth": {"type": "bearer", "token": "test-token"}, + } + } + from tools.a2a_tool import _resolve_agent_url + with patch("tools.a2a_tool._load_a2a_config", return_value=config): + url, agent_config = _resolve_agent_url("researcher") + assert url == "http://researcher.local:8080" + assert agent_config["auth"]["type"] == "bearer" + + def test_unknown_name_raises(self): + """Unknown agent name raises ValueError.""" + from tools.a2a_tool import _resolve_agent_url + with patch("tools.a2a_tool._load_a2a_config", return_value={}): + with pytest.raises(ValueError, match="Unknown A2A agent"): + _resolve_agent_url("nonexistent") + + def test_config_missing_url_raises(self): + """Config entry without URL raises ValueError.""" + config = {"broken": {"auth": {"type": "bearer"}}} + from tools.a2a_tool import _resolve_agent_url + with patch("tools.a2a_tool._load_a2a_config", return_value=config): + with pytest.raises(ValueError, match="no 'url'"): + _resolve_agent_url("broken") + + +# --------------------------------------------------------------------------- +# Agent Card formatting +# --------------------------------------------------------------------------- + +class TestFormatAgentCard: + def test_basic_card(self): + """Agent Card is formatted with all fields.""" + from tools.a2a_tool import _format_agent_card + card = _make_agent_card() + result = _format_agent_card(card) + assert result["name"] == "test-agent" + assert result["description"] == "A test agent" + assert len(result["skills"]) == 1 + assert result["skills"][0]["name"] == "test-skill" + assert result["capabilities"]["streaming"] is True + assert "bearer" in result["auth_schemes"] + + def test_card_without_skills(self): + """Card with no skills still formats correctly.""" + from tools.a2a_tool import _format_agent_card + card = SimpleNamespace( + name="minimal", + description="", + url="http://localhost", + skills=None, + capabilities=None, + securitySchemes=None, + ) + result = _format_agent_card(card) + assert result["name"] == "minimal" + assert "skills" not in result + + +# --------------------------------------------------------------------------- +# Response extraction +# --------------------------------------------------------------------------- + +class TestExtractResponse: + def test_successful_response(self): + """Completed task extracts text from artifacts.""" + from tools.a2a_tool import _extract_response + response = _make_send_response(text="Result text", status="completed") + result = _extract_response(response) + assert result["status"] == "completed" + assert "Result text" in result.get("response", "") + + def test_error_response(self): + """Error in response is captured.""" + from tools.a2a_tool import _extract_response + error_response = SimpleNamespace( + root=SimpleNamespace( + error=SimpleNamespace(message="Something went wrong"), + result=None, + ) + ) + result = _extract_response(error_response) + assert result["status"] == "error" + + def test_input_required(self): + """INPUT_REQUIRED status returns the agent's question.""" + from tools.a2a_tool import _extract_response + response = _make_send_response( + text="What file should I read?", + status="input-required", + ) + result = _extract_response(response) + assert result["status"] == "input-required" + assert "What file" in result["response"] + + +# --------------------------------------------------------------------------- +# Tool handler: a2a_discover +# --------------------------------------------------------------------------- + +class TestA2ADiscover: + def test_missing_agent_param(self): + """Missing agent parameter returns error.""" + from tools.a2a_tool import a2a_discover + result = json.loads(a2a_discover({})) + assert "error" in result + assert "Missing" in result["error"] + + def test_empty_agent_param(self): + """Empty agent string returns error.""" + from tools.a2a_tool import a2a_discover + result = json.loads(a2a_discover({"agent": " "})) + assert "error" in result + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_successful_discover(self, mock_resolve, mock_run): + """Successful discovery returns formatted card.""" + from tools.a2a_tool import a2a_discover + mock_resolve.return_value = ("http://localhost:9999", {}) + mock_run.return_value = { + "name": "test-agent", + "description": "A test agent", + "skills": [{"id": "1", "name": "test", "description": "test skill"}], + } + result = json.loads(a2a_discover({"agent": "http://localhost:9999"})) + assert result["name"] == "test-agent" + assert len(result["skills"]) == 1 + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_discover_connection_error(self, mock_resolve, mock_run): + """Connection error returns sanitized message.""" + from tools.a2a_tool import a2a_discover + mock_resolve.return_value = ("http://unreachable:9999", {}) + mock_run.side_effect = ConnectionError("Connection refused") + result = json.loads(a2a_discover({"agent": "http://unreachable:9999"})) + assert "error" in result + assert "Failed to discover" in result["error"] + + +# --------------------------------------------------------------------------- +# Tool handler: a2a_call +# --------------------------------------------------------------------------- + +class TestA2ACall: + def test_missing_agent(self): + """Missing agent returns error.""" + from tools.a2a_tool import a2a_call + result = json.loads(a2a_call({"message": "hello"})) + assert "error" in result + assert "agent" in result["error"].lower() + + def test_missing_message(self): + """Missing message returns error.""" + from tools.a2a_tool import a2a_call + result = json.loads(a2a_call({"agent": "http://localhost:9999"})) + assert "error" in result + assert "message" in result["error"].lower() + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_successful_call(self, mock_resolve, mock_run): + """Successful call returns agent response.""" + from tools.a2a_tool import a2a_call + mock_resolve.return_value = ("http://localhost:9999", {}) + mock_run.return_value = json.dumps({ + "status": "completed", + "response": "Hello from the agent!", + }) + result = json.loads(a2a_call({ + "agent": "http://localhost:9999", + "message": "Say hello", + })) + assert result["status"] == "completed" + assert "Hello" in result["response"] + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_call_with_config_name(self, mock_resolve, mock_run): + """Call using config name resolves correctly.""" + from tools.a2a_tool import a2a_call + mock_resolve.return_value = ("http://researcher.local:8080", {"timeout": 60}) + mock_run.return_value = json.dumps({"status": "completed", "response": "Done"}) + result = json.loads(a2a_call({ + "agent": "researcher", + "message": "Research topic X", + })) + assert result["status"] == "completed" + mock_resolve.assert_called_once_with("researcher") + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_call_error_sanitized(self, mock_resolve, mock_run): + """Credentials are stripped from error messages.""" + from tools.a2a_tool import a2a_call + mock_resolve.return_value = ("http://localhost:9999", {}) + mock_run.side_effect = Exception("Auth failed with Bearer sk-secret123abc") + result = json.loads(a2a_call({ + "agent": "http://localhost:9999", + "message": "hello", + })) + assert "error" in result + assert "sk-secret123abc" not in result["error"] + assert "[REDACTED]" in result["error"] + + +# --------------------------------------------------------------------------- +# Credential sanitization +# --------------------------------------------------------------------------- + +class TestSanitizeError: + def test_bearer_token(self): + """Bearer tokens are redacted.""" + from tools.a2a_tool import _sanitize_error + assert "[REDACTED]" in _sanitize_error("Bearer sk-abc123xyz") + + def test_api_key(self): + """sk- prefixed keys are redacted.""" + from tools.a2a_tool import _sanitize_error + result = _sanitize_error("Failed with key sk-mySecretKey123") + assert "sk-mySecretKey123" not in result + + def test_clean_text_unchanged(self): + """Text without credentials passes through.""" + from tools.a2a_tool import _sanitize_error + text = "Connection refused to localhost:9999" + assert _sanitize_error(text) == text + + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + +class TestRegistration: + def test_tools_registered(self): + """Both A2A tools are registered in the registry.""" + from tools.registry import registry + import tools.a2a_tool # noqa: F401 -- triggers registration + + names = registry.get_all_tool_names() + assert "a2a_discover" in names + assert "a2a_call" in names + + def test_toolset_assignment(self): + """Both tools belong to the 'a2a' toolset.""" + from tools.registry import registry + import tools.a2a_tool # noqa: F401 + + assert registry.get_toolset_for_tool("a2a_discover") == "a2a" + assert registry.get_toolset_for_tool("a2a_call") == "a2a" + + def test_schemas_valid(self): + """Tool schemas have required fields.""" + from tools.a2a_tool import A2A_DISCOVER_SCHEMA, A2A_CALL_SCHEMA + + assert A2A_DISCOVER_SCHEMA["name"] == "a2a_discover" + assert "parameters" in A2A_DISCOVER_SCHEMA + assert "agent" in A2A_DISCOVER_SCHEMA["parameters"]["properties"] + + assert A2A_CALL_SCHEMA["name"] == "a2a_call" + assert "agent" in A2A_CALL_SCHEMA["parameters"]["properties"] + assert "message" in A2A_CALL_SCHEMA["parameters"]["properties"] + assert "stream" in A2A_CALL_SCHEMA["parameters"]["properties"] + + +# --------------------------------------------------------------------------- +# Format helpers +# --------------------------------------------------------------------------- + +class TestFormatCallResponse: + def test_empty_results(self): + """Empty results return a default message.""" + from tools.a2a_tool import _format_call_response + result = json.loads(_format_call_response([])) + assert result["status"] == "completed" + assert "empty" in result["response"].lower() + + def test_multiple_chunks(self): + """Multiple streaming chunks are joined.""" + from tools.a2a_tool import _format_call_response + result = json.loads(_format_call_response(["chunk1", "chunk2", "chunk3"])) + assert "chunk1" in result["response"] + assert "chunk2" in result["response"] + assert "chunk3" in result["response"] + + +# --------------------------------------------------------------------------- +# Config loading +# --------------------------------------------------------------------------- + +class TestLoadConfig: + @patch("tools.a2a_tool.os.path.exists", return_value=False) + def test_no_config_file(self, mock_exists): + """Missing config file returns empty dict.""" + import tools.a2a_tool as mod + mod._config_cache = None # Reset cache + result = mod._load_a2a_config() + assert result == {} + + def test_config_caching(self): + """Config is cached after first load.""" + import tools.a2a_tool as mod + mod._config_cache = {"cached": {"url": "http://cached"}} + result = mod._load_a2a_config() + assert "cached" in result + mod._config_cache = None # Cleanup + + +# =========================================================================== +# Phase 3: Multi-Agent Orchestration +# =========================================================================== + +# --------------------------------------------------------------------------- +# Registry builder +# --------------------------------------------------------------------------- + +class TestBuildAgentRegistry: + def test_empty_registry(self): + """No config + no discovered cards = empty list.""" + from tools.a2a_tool import _build_agent_registry + with patch("tools.a2a_tool._load_a2a_config", return_value={}): + with patch("tools.a2a_tool._agent_cards", {}): + result = _build_agent_registry() + assert result == [] + + def test_config_only(self): + """Config agents appear even without discovery.""" + from tools.a2a_tool import _build_agent_registry + config = { + "researcher": {"url": "http://researcher:8080"}, + "coder": {"url": "http://coder:9090"}, + } + with patch("tools.a2a_tool._load_a2a_config", return_value=config): + with patch("tools.a2a_tool._agent_cards", {}): + result = _build_agent_registry() + assert len(result) == 2 + names = {a["name"] for a in result} + assert names == {"researcher", "coder"} + assert all(a["source"] == "config" for a in result) + assert all(a["status"] == "configured" for a in result) + + def test_discovered_only(self): + """Discovered agents not in config appear as source=discovered.""" + from tools.a2a_tool import _build_agent_registry + card = _make_agent_card(name="remote-agent", url="http://remote:5555") + with patch("tools.a2a_tool._load_a2a_config", return_value={}): + with patch("tools.a2a_tool._agent_cards", {"http://remote:5555": card}): + result = _build_agent_registry() + assert len(result) == 1 + assert result[0]["name"] == "remote-agent" + assert result[0]["source"] == "discovered" + assert len(result[0]["skills"]) == 1 + + def test_merged_config_and_discovered(self): + """Config agent that's been discovered shows status=discovered with skills.""" + from tools.a2a_tool import _build_agent_registry + card = _make_agent_card(name="researcher-card", url="http://researcher:8080") + config = {"researcher": {"url": "http://researcher:8080"}} + with patch("tools.a2a_tool._load_a2a_config", return_value=config): + with patch("tools.a2a_tool._agent_cards", {"http://researcher:8080": card}): + result = _build_agent_registry() + assert len(result) == 1 + assert result[0]["name"] == "researcher" + assert result[0]["source"] == "config" + assert result[0]["status"] == "discovered" + assert len(result[0]["skills"]) == 1 + + +# --------------------------------------------------------------------------- +# Skill matching +# --------------------------------------------------------------------------- + +class TestMatchSkillsToGoal: + def test_exact_keyword_match(self): + """Keywords in goal that match skill names score > 0.""" + from tools.a2a_tool import _match_skills_to_goal + info = { + "description": "Research assistant", + "skills": [{"name": "research", "description": "web research", "id": "s1"}], + } + score = _match_skills_to_goal("Do some research on AI", info) + assert score > 0.0 + + def test_no_match(self): + """Completely unrelated goal scores 0.""" + from tools.a2a_tool import _match_skills_to_goal + info = { + "description": "Music player", + "skills": [{"name": "play-music", "description": "plays songs", "id": "s1"}], + } + score = _match_skills_to_goal("Deploy kubernetes cluster", info) + assert score == 0.0 + + def test_empty_goal(self): + """Empty goal returns 0.""" + from tools.a2a_tool import _match_skills_to_goal + info = {"description": "anything", "skills": []} + assert _match_skills_to_goal("", info) == 0.0 + + def test_no_skills_no_description(self): + """Agent with empty skills/description scores 0.""" + from tools.a2a_tool import _match_skills_to_goal + info = {"description": "", "skills": []} + assert _match_skills_to_goal("research AI topics", info) == 0.0 + + def test_case_insensitive(self): + """Matching is case-insensitive.""" + from tools.a2a_tool import _match_skills_to_goal + info = { + "description": "RESEARCH Agent", + "skills": [{"name": "Research", "description": "Deep Research", "id": "s1"}], + } + score = _match_skills_to_goal("research", info) + assert score > 0.0 + + +# --------------------------------------------------------------------------- +# A2A List tool handler +# --------------------------------------------------------------------------- + +class TestA2AList: + def test_empty_registry(self): + """Empty registry returns total=0.""" + from tools.a2a_tool import a2a_list + with patch("tools.a2a_tool._build_agent_registry", return_value=[]): + result = json.loads(a2a_list({})) + assert result["total"] == 0 + assert result["agents"] == [] + + def test_populated_registry(self): + """Registry with agents returns them all.""" + from tools.a2a_tool import a2a_list + agents = [ + {"name": "a1", "url": "http://a1", "source": "config", + "status": "configured", "skills": [], "description": ""}, + {"name": "a2", "url": "http://a2", "source": "discovered", + "status": "discovered", "skills": [], "description": ""}, + ] + with patch("tools.a2a_tool._build_agent_registry", return_value=agents): + result = json.loads(a2a_list({})) + assert result["total"] == 2 + assert len(result["agents"]) == 2 + + def test_error_handling(self): + """Exceptions return error JSON.""" + from tools.a2a_tool import a2a_list + with patch("tools.a2a_tool._build_agent_registry", side_effect=RuntimeError("boom")): + result = json.loads(a2a_list({})) + assert "error" in result + + +# --------------------------------------------------------------------------- +# A2A Orchestrate tool handler +# --------------------------------------------------------------------------- + +class TestA2AOrchestrate: + def test_missing_goal(self): + """Missing goal returns error.""" + from tools.a2a_tool import a2a_orchestrate + result = json.loads(a2a_orchestrate({})) + assert "error" in result + assert "goal" in result["error"].lower() + + def test_empty_goal(self): + """Empty goal string returns error.""" + from tools.a2a_tool import a2a_orchestrate + result = json.loads(a2a_orchestrate({"goal": " "})) + assert "error" in result + + def test_invalid_mode(self): + """Invalid mode returns error.""" + from tools.a2a_tool import a2a_orchestrate + result = json.loads(a2a_orchestrate({"goal": "test", "mode": "invalid"})) + assert "error" in result + assert "invalid" in result["error"].lower() + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_explicit_agents_all_mode(self, mock_resolve, mock_run): + """Explicit agents with mode=all calls orchestrate correctly.""" + from tools.a2a_tool import a2a_orchestrate + mock_resolve.return_value = ("http://agent1:8080", {}) + mock_run.return_value = { + "mode": "all", + "agents_called": 1, + "results": [{"agent": "agent1", "url": "http://agent1:8080", + "status": "success", "response": "done", "duration_ms": 100}], + } + result = json.loads(a2a_orchestrate({ + "goal": "Research AI", + "agents": ["agent1"], + "mode": "all", + })) + assert result["mode"] == "all" + assert result["agents_called"] == 1 + assert result["results"][0]["status"] == "success" + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_first_mode(self, mock_resolve, mock_run): + """Mode=first returns first success.""" + from tools.a2a_tool import a2a_orchestrate + mock_resolve.return_value = ("http://a:8080", {}) + mock_run.return_value = { + "mode": "first", + "agents_called": 1, + "results": [{"agent": "a", "url": "http://a:8080", + "status": "success", "response": "fast", "duration_ms": 50}], + } + result = json.loads(a2a_orchestrate({ + "goal": "Quick task", + "agents": ["a"], + "mode": "first", + })) + assert result["mode"] == "first" + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._resolve_agent_url") + def test_best_mode(self, mock_resolve, mock_run): + """Mode=best is accepted (alias for all).""" + from tools.a2a_tool import a2a_orchestrate + mock_resolve.return_value = ("http://a:8080", {}) + mock_run.return_value = { + "mode": "best", + "agents_called": 1, + "results": [], + } + result = json.loads(a2a_orchestrate({ + "goal": "task", + "agents": ["a"], + "mode": "best", + })) + assert result["mode"] == "best" + + @patch("tools.a2a_tool._run_on_loop") + @patch("tools.a2a_tool._auto_select_agents") + def test_auto_select_agents(self, mock_auto, mock_run): + """No explicit agents triggers auto-select.""" + from tools.a2a_tool import a2a_orchestrate + mock_auto.return_value = [("http://auto:8080", {"_name": "auto"})] + mock_run.return_value = { + "mode": "all", + "agents_called": 1, + "results": [{"agent": "auto", "url": "http://auto:8080", + "status": "success", "response": "auto-done", "duration_ms": 200}], + } + result = json.loads(a2a_orchestrate({"goal": "research AI"})) + mock_auto.assert_called_once_with("research AI") + assert result["agents_called"] == 1 + + @patch("tools.a2a_tool._resolve_agent_url") + def test_resolve_failure(self, mock_resolve): + """Resolve failure returns error JSON.""" + from tools.a2a_tool import a2a_orchestrate + mock_resolve.side_effect = ValueError("Unknown agent 'bad'") + result = json.loads(a2a_orchestrate({ + "goal": "task", + "agents": ["bad"], + })) + assert "error" in result + + @patch("tools.a2a_tool._auto_select_agents") + def test_empty_auto_select(self, mock_auto): + """Empty auto-select raises and returns error.""" + from tools.a2a_tool import a2a_orchestrate + mock_auto.side_effect = ValueError("No agents in registry") + result = json.loads(a2a_orchestrate({"goal": "task"})) + assert "error" in result + + +# --------------------------------------------------------------------------- +# Phase 3 Registration +# --------------------------------------------------------------------------- + +class TestOrchestrateRegistration: + def test_list_tool_registered(self): + """a2a_list is registered in the registry.""" + from tools.registry import registry + import tools.a2a_tool # noqa: F401 + assert "a2a_list" in registry.get_all_tool_names() + + def test_orchestrate_tool_registered(self): + """a2a_orchestrate is registered in the registry.""" + from tools.registry import registry + import tools.a2a_tool # noqa: F401 + assert "a2a_orchestrate" in registry.get_all_tool_names() + + def test_toolset_includes_new_tools(self): + """a2a toolset includes all 4 tools.""" + from toolsets import resolve_toolset + tools = resolve_toolset("a2a") + assert "a2a_discover" in tools + assert "a2a_call" in tools + assert "a2a_list" in tools + assert "a2a_orchestrate" in tools + + def test_schemas_valid(self): + """New tool schemas have required fields.""" + from tools.a2a_tool import A2A_LIST_SCHEMA, A2A_ORCHESTRATE_SCHEMA + + assert A2A_LIST_SCHEMA["name"] == "a2a_list" + assert "parameters" in A2A_LIST_SCHEMA + + assert A2A_ORCHESTRATE_SCHEMA["name"] == "a2a_orchestrate" + assert "goal" in A2A_ORCHESTRATE_SCHEMA["parameters"]["properties"] + assert "agents" in A2A_ORCHESTRATE_SCHEMA["parameters"]["properties"] + assert "mode" in A2A_ORCHESTRATE_SCHEMA["parameters"]["properties"] + assert "goal" in A2A_ORCHESTRATE_SCHEMA["parameters"]["required"] diff --git a/tools/a2a_tool.py b/tools/a2a_tool.py new file mode 100644 index 0000000000000..d932a8c3e9a1d --- /dev/null +++ b/tools/a2a_tool.py @@ -0,0 +1,986 @@ +#!/usr/bin/env python3 +""" +A2A (Agent-to-Agent) Protocol Client Support + +Connects to remote A2A agents via HTTP, discovers their capabilities through +Agent Cards, and enables Hermes to delegate tasks to agents built on any +framework (LangChain, CrewAI, Google ADK, AutoGen, etc.). + +Configuration is read from ~/.hermes/config.yaml under the ``a2a_agents`` key. +The ``a2a-sdk`` Python package is optional -- if not installed, this module is a +no-op and logs a debug message. + +Example config:: + + a2a_agents: + researcher: + url: "http://localhost:9999" + auth: + type: "bearer" + token: "sk-..." + timeout: 120 + coder: + url: "http://remote-agent:8080" + +Features: + - Agent Card discovery from any A2A-compliant endpoint + - Synchronous and streaming message sending + - Multi-turn task support (INPUT_REQUIRED state) + - Config-driven named agents with direct URL fallback + - Thread-safe agent card caching + - Credential stripping in error messages + - Optional dependency -- graceful degradation when a2a-sdk not installed + +Architecture: + A dedicated background event loop (_a2a_loop) runs in a daemon thread, + mirroring the MCP tool pattern. Async A2A SDK calls are scheduled onto + this loop via ``run_coroutine_threadsafe()``. + +Thread safety: + _agent_cards cache and _a2a_loop/_a2a_thread are accessed from multiple + threads. All mutations are protected by _lock. +""" + +import asyncio +import json +import logging +import os +import re +import threading +from typing import Any, Dict, Optional +from uuid import uuid4 + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Graceful import -- A2A SDK is an optional dependency +# --------------------------------------------------------------------------- + +_A2A_AVAILABLE = False +try: + import httpx + from a2a.client import A2ACardResolver + from a2a.client.helpers import create_text_message_object + + # Try new ClientFactory pattern (v0.3.25+) + _A2A_CLIENT_FACTORY = False + try: + from a2a.client.client import ClientConfig + from a2a.client.client_factory import ClientFactory + _A2A_CLIENT_FACTORY = True + except ImportError: + pass + + # Fallback: legacy A2AClient (deprecated but functional) + _A2A_LEGACY_CLIENT = False + if not _A2A_CLIENT_FACTORY: + try: + from a2a.client.legacy import A2AClient + _A2A_LEGACY_CLIENT = True + except ImportError: + pass + + # Types for building requests + try: + from a2a.types import ( + MessageSendParams, + SendMessageRequest, + SendStreamingMessageRequest, + ) + _A2A_TYPES_AVAILABLE = True + except ImportError: + _A2A_TYPES_AVAILABLE = False + + _A2A_AVAILABLE = True + logger.debug("A2A SDK loaded (factory=%s, legacy=%s, types=%s)", + _A2A_CLIENT_FACTORY, _A2A_LEGACY_CLIENT, _A2A_TYPES_AVAILABLE) +except ImportError: + logger.debug("a2a-sdk package not installed -- A2A tool support disabled") + +from tools.registry import registry + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_DEFAULT_TIMEOUT = 120 # seconds for A2A calls + +# Credential patterns to strip from error messages +_CREDENTIAL_PATTERN = re.compile( + r"(?:" + r"sk-[A-Za-z0-9_]{1,255}" + r"|Bearer\s+\S+" + r"|token=[^\s&,;\"']{1,255}" + r"|key=[^\s&,;\"']{1,255}" + r")", + re.IGNORECASE, +) + +# --------------------------------------------------------------------------- +# Background event loop (mirrors mcp_tool.py pattern) +# --------------------------------------------------------------------------- + +_lock = threading.Lock() +_async_lock: Optional[asyncio.Lock] = None # Created lazily on the background loop +_a2a_loop: Optional[asyncio.AbstractEventLoop] = None +_a2a_thread: Optional[threading.Thread] = None +_agent_cards: Dict[str, Any] = {} # url -> AgentCard cache +_config_cache: Optional[Dict[str, Any]] = None + + +def _get_async_lock() -> asyncio.Lock: + """Get or create the async lock on the background event loop.""" + global _async_lock + if _async_lock is None: + _async_lock = asyncio.Lock() + return _async_lock + + +def _ensure_loop() -> asyncio.AbstractEventLoop: + """Start the background event loop if not already running.""" + global _a2a_loop, _a2a_thread + with _lock: + if _a2a_loop is not None and _a2a_loop.is_running(): + return _a2a_loop + + loop = asyncio.new_event_loop() + _a2a_loop = loop + + def _run(): + asyncio.set_event_loop(loop) + loop.run_forever() + + thread = threading.Thread(target=_run, daemon=True, name="a2a-loop") + thread.start() + _a2a_thread = thread + return loop + + +def _run_on_loop(coro, timeout: Optional[float] = None) -> Any: + """Schedule a coroutine on the background loop and wait for the result.""" + loop = _ensure_loop() + future = asyncio.run_coroutine_threadsafe(coro, loop) + return future.result(timeout=timeout or (_DEFAULT_TIMEOUT + 30)) + + +def _sanitize_error(text: str) -> str: + """Strip credential-like patterns from error text.""" + return _CREDENTIAL_PATTERN.sub("[REDACTED]", text) + + +# --------------------------------------------------------------------------- +# Config loading +# --------------------------------------------------------------------------- + +def _load_a2a_config() -> Dict[str, Any]: + """Load a2a_agents config from ~/.hermes/config.yaml.""" + global _config_cache + if _config_cache is not None: + return _config_cache + + try: + import yaml + config_path = os.path.join( + os.path.expanduser(os.getenv("HERMES_HOME", "~/.hermes")), + "config.yaml", + ) + if os.path.exists(config_path): + with open(config_path, "r") as f: + config = yaml.safe_load(f) or {} + _config_cache = config.get("a2a_agents", {}) + else: + _config_cache = {} + except Exception as e: + logger.debug("Failed to load A2A config: %s", e) + _config_cache = {} + + return _config_cache + + +def _resolve_agent_url(agent: str) -> tuple: + """Resolve an agent name or URL to (url, config_dict). + + Returns: + Tuple of (base_url, config_dict). config_dict may be empty for direct URLs. + """ + # Direct URL + if agent.startswith("http://") or agent.startswith("https://"): + return agent.rstrip("/"), {} + + # Config lookup + config = _load_a2a_config() + if agent in config: + entry = config[agent] + url = entry.get("url", "").rstrip("/") + if not url: + raise ValueError(f"A2A agent '{agent}' has no 'url' in config") + return url, entry + + raise ValueError( + f"Unknown A2A agent '{agent}'. " + f"Provide a direct URL or configure it in ~/.hermes/config.yaml under a2a_agents." + ) + + +# --------------------------------------------------------------------------- +# Registry & skill matching helpers +# --------------------------------------------------------------------------- + +def _build_agent_registry() -> list: + """Merge config agents + cached discovered cards into a unified list. + + Each entry: {name, url, source, status, skills, description}. + Thread-safe reads of _agent_cards via _lock. No network calls. + """ + agents = [] + + # Config agents + config = _load_a2a_config() + for name, entry in config.items(): + url = entry.get("url", "").rstrip("/") + if not url: + continue + + with _lock: + card = _agent_cards.get(url) + + info = { + "name": name, + "url": url, + "source": "config", + "status": "discovered" if card else "configured", + "skills": [], + "description": "", + } + if card: + formatted = _format_agent_card(card) + info["skills"] = formatted.get("skills", []) + info["description"] = formatted.get("description", "") + + agents.append(info) + + # Discovered agents not in config + config_urls = {e.get("url", "").rstrip("/") for e in config.values()} + with _lock: + discovered = dict(_agent_cards) + + for url, card in discovered.items(): + if url in config_urls: + continue + formatted = _format_agent_card(card) + agents.append({ + "name": formatted.get("name", url), + "url": url, + "source": "discovered", + "status": "discovered", + "skills": formatted.get("skills", []), + "description": formatted.get("description", ""), + }) + + return agents + + +def _match_skills_to_goal(goal: str, agent_info: dict) -> float: + """Simple keyword overlap scoring (0.0-1.0). + + Tokenizes goal into lowercase words (>2 chars), compares against + skill names/descriptions/ids + agent description. No LLM needed. + """ + if not goal: + return 0.0 + + goal_words = {w.lower() for w in goal.split() if len(w) > 2} + if not goal_words: + return 0.0 + + # Build corpus from agent info + corpus_parts = [agent_info.get("description", "")] + for skill in agent_info.get("skills", []): + corpus_parts.append(skill.get("name", "")) + corpus_parts.append(skill.get("description", "")) + corpus_parts.append(skill.get("id", "")) + + corpus = " ".join(corpus_parts).lower() + corpus_words = {w for w in corpus.split() if len(w) > 2} + + if not corpus_words: + return 0.0 + + overlap = goal_words & corpus_words + return len(overlap) / len(goal_words) + + +def _auto_select_agents(goal: str) -> list: + """Score all registry agents against goal, return those with positive scores. + + Falls back to all agents if no skill matches. + Raises ValueError if registry is empty. + + Returns: + List of (url, config_dict) tuples sorted by score descending. + """ + agents = _build_agent_registry() + if not agents: + raise ValueError("No agents in registry. Configure agents or discover them first.") + + config = _load_a2a_config() + + scored = [] + for info in agents: + score = _match_skills_to_goal(goal, info) + scored.append((score, info)) + + # Sort by score descending + scored.sort(key=lambda x: x[0], reverse=True) + + def _agent_config(info: dict) -> dict: + """Get config for an agent, preserving auth for discovered agents.""" + cfg = config.get(info["name"], {}) + if cfg: + return cfg + # For discovered agents not in config, build a minimal config + # from the registry entry so the URL and any cached state are preserved + return {"url": info["url"], "_name": info["name"]} + + # Filter to positive scores + selected = [(info["url"], _agent_config(info)) + for score, info in scored if score > 0.0] + + # Fall back to all agents if nothing matched + if not selected: + selected = [(info["url"], _agent_config(info)) + for _, info in scored] + + return selected + + +# --------------------------------------------------------------------------- +# Core async operations +# --------------------------------------------------------------------------- + +async def _async_discover(url: str, config: dict) -> dict: + """Fetch and cache an Agent Card from a remote A2A agent.""" + async_lock = _get_async_lock() + async with async_lock: + if url in _agent_cards: + return _format_agent_card(_agent_cards[url]) + + timeout = config.get("timeout", _DEFAULT_TIMEOUT) + headers = {} + auth = config.get("auth", {}) + if auth.get("type") == "bearer" and auth.get("token"): + headers["Authorization"] = f"Bearer {auth['token']}" + + async with httpx.AsyncClient(timeout=timeout, headers=headers) as client: + resolver = A2ACardResolver( + httpx_client=client, + base_url=url, + ) + card = await resolver.get_agent_card() + + async with async_lock: + _agent_cards[url] = card + + # Also update under threading lock for sync readers (_build_agent_registry) + with _lock: + _agent_cards[url] = card + + return _format_agent_card(card) + + +async def _async_call(url: str, config: dict, message: str, stream: bool = False) -> str: + """Send a message to a remote A2A agent and return the response.""" + # Ensure we have the agent card + with _lock: + card = _agent_cards.get(url) + + if not card: + await _async_discover(url, config) + with _lock: + card = _agent_cards.get(url) + + if not card: + return json.dumps({"error": f"Failed to discover agent at {url}"}) + + timeout = config.get("timeout", _DEFAULT_TIMEOUT) + headers = {} + auth = config.get("auth", {}) + if auth.get("type") == "bearer" and auth.get("token"): + headers["Authorization"] = f"Bearer {auth['token']}" + + async with httpx.AsyncClient(timeout=timeout, headers=headers) as http_client: + if _A2A_CLIENT_FACTORY: + return await _call_with_factory(http_client, card, message, stream) + elif _A2A_LEGACY_CLIENT: + return await _call_with_legacy(http_client, card, message, stream) + else: + return json.dumps({"error": "No A2A client implementation available. Update a2a-sdk."}) + + +async def _call_with_factory(http_client, card, message: str, stream: bool) -> str: + """Send message using the new ClientFactory pattern.""" + from a2a.types import Message, Part, TextPart, Role, SendMessageRequest, MessageSendParams + + # ClientConfig accepts streaming flag only; pass httpx_client to the + # factory so auth headers and timeout are preserved on the actual client. + cfg = ClientConfig(streaming=stream) + factory = ClientFactory(config=cfg) + client = factory.create(card, httpx_client=http_client) + + msg = Message( + role=Role.user, + parts=[Part(root=TextPart(text=message))], + message_id=uuid4().hex, + ) + params = MessageSendParams(message=msg) + request = SendMessageRequest(id=uuid4().hex, params=params) + + try: + if stream: + results = [] + async for event in client.send_message_streaming(request): + results.append(str(event)) + return _format_call_response(results) + else: + response = await client.send_message(request) + return json.dumps(_extract_response(response), indent=2) + finally: + try: + await client.close() + except Exception: + pass + + +async def _call_with_legacy(http_client, card, message: str, stream: bool) -> str: + """Send message using deprecated A2AClient.""" + import warnings + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + client = A2AClient(httpx_client=http_client, agent_card=card) + + msg_obj = create_text_message_object(content=message) + + if not _A2A_TYPES_AVAILABLE: + return json.dumps({"error": "A2A message types not available."}) + + params = MessageSendParams(message=msg_obj) + + if stream: + request = SendStreamingMessageRequest(id=uuid4().hex, params=params) + results = [] + async for event in client.send_message_streaming(request): + results.append(str(event)) + return _format_call_response(results) + else: + request = SendMessageRequest(id=uuid4().hex, params=params) + response = await client.send_message(request) + return json.dumps(_extract_response(response), indent=2) + + +# --------------------------------------------------------------------------- +# Response formatting +# --------------------------------------------------------------------------- + +def _format_agent_card(card) -> dict: + """Format an AgentCard into a clean dict for the LLM.""" + result = { + "name": getattr(card, "name", "Unknown"), + "description": getattr(card, "description", ""), + "url": getattr(card, "url", ""), + } + + # Skills + skills = getattr(card, "skills", None) + if skills: + result["skills"] = [] + for skill in skills: + skill_info = { + "id": getattr(skill, "id", ""), + "name": getattr(skill, "name", ""), + "description": getattr(skill, "description", ""), + } + result["skills"].append(skill_info) + + # Capabilities + caps = getattr(card, "capabilities", None) + if caps: + result["capabilities"] = { + "streaming": getattr(caps, "streaming", False), + "pushNotifications": getattr(caps, "pushNotifications", False), + } + + # Auth schemes + security_schemes = getattr(card, "securitySchemes", None) + if security_schemes: + result["auth_schemes"] = list(security_schemes.keys()) if isinstance(security_schemes, dict) else [] + + return result + + +def _format_call_response(results: list) -> str: + """Format collected streaming results into a response string.""" + if not results: + return json.dumps({"status": "completed", "response": "(empty response)"}) + + # Join all chunks + combined = "\n".join(results) + return json.dumps({"status": "completed", "response": combined}, indent=2) + + +def _extract_response(response) -> dict: + """Extract meaningful content from a SendMessageResponse.""" + result = {"status": "unknown", "response": ""} + + try: + # Navigate the response structure + root = getattr(response, "root", response) + + # Check for error + error = getattr(root, "error", None) + if error: + return {"status": "error", "error": str(error)} + + # Get the result (Task or Message) + task_or_msg = getattr(root, "result", None) + if task_or_msg is None: + return {"status": "completed", "response": str(response)} + + # If it's a Task, check status and extract artifacts + status = getattr(task_or_msg, "status", None) + if status: + state = getattr(status, "state", None) + result["status"] = str(state) if state else "unknown" + + # Check for INPUT_REQUIRED (multi-turn) + message = getattr(status, "message", None) + if message: + parts = getattr(message, "parts", []) + texts = [] + for part in parts: + text = getattr(part, "text", None) or getattr(getattr(part, "root", None), "text", None) + if text: + texts.append(text) + if texts: + result["response"] = "\n".join(texts) + + # Extract artifacts + artifacts = getattr(task_or_msg, "artifacts", None) + if artifacts: + artifact_texts = [] + for artifact in artifacts: + parts = getattr(artifact, "parts", []) + for part in parts: + text = getattr(part, "text", None) or getattr(getattr(part, "root", None), "text", None) + if text: + artifact_texts.append(text) + if artifact_texts: + result["artifacts"] = artifact_texts + if not result.get("response"): + result["response"] = "\n".join(artifact_texts) + + except Exception as e: + result = {"status": "error", "error": f"Failed to parse response: {e}"} + + return result + + +async def _async_orchestrate(goal: str, agent_targets: list, mode: str, + default_timeout: float = _DEFAULT_TIMEOUT) -> dict: + """Fan out a goal to multiple agents. + + Args: + goal: The task text to send to each agent. + agent_targets: List of (url, config) tuples. + mode: "all" (collect all), "first" (first success), or "best" (alias for all). + default_timeout: Per-agent timeout in seconds. + + Returns: + {mode, agents_called, results: [{agent, url, status, response, duration_ms}]} + """ + import time + + async def _call_one(url: str, config: dict) -> dict: + start = time.monotonic() + try: + agent_timeout = config.get("timeout", default_timeout) + response = await asyncio.wait_for( + _async_call(url, config, goal), + timeout=agent_timeout, + ) + duration = int((time.monotonic() - start) * 1000) + return { + "agent": config.get("_name", url), + "url": url, + "status": "success", + "response": response, + "duration_ms": duration, + } + except asyncio.TimeoutError: + duration = int((time.monotonic() - start) * 1000) + return { + "agent": config.get("_name", url), + "url": url, + "status": "timeout", + "response": None, + "duration_ms": duration, + } + except Exception as e: + duration = int((time.monotonic() - start) * 1000) + return { + "agent": config.get("_name", url), + "url": url, + "status": "error", + "response": _sanitize_error(str(e)), + "duration_ms": duration, + } + + results = [] + agents_called = len(agent_targets) + + if mode in ("all", "best"): + results = await asyncio.gather( + *[_call_one(url, cfg) for url, cfg in agent_targets], + return_exceptions=False, + ) + results = list(results) + elif mode == "first": + tasks = [asyncio.create_task(_call_one(url, cfg)) + for url, cfg in agent_targets] + done = set() + pending = set(tasks) + first_success = None + + while pending: + newly_done, pending = await asyncio.wait( + pending, return_when=asyncio.FIRST_COMPLETED + ) + done.update(newly_done) + for t in newly_done: + result = t.result() + if result["status"] == "success": + first_success = result + break + if first_success: + break + + # Cancel remaining + for t in pending: + t.cancel() + + if first_success: + results = [first_success] + else: + # All failed — return all results + results = [t.result() for t in done] + + return { + "mode": mode, + "agents_called": agents_called, + "results": results, + } + + +# --------------------------------------------------------------------------- +# Tool handlers (sync, bridged to async via background loop) +# --------------------------------------------------------------------------- + +def a2a_discover(args: dict, **kwargs) -> str: + """Discover a remote A2A agent by fetching its Agent Card. + + Args: + args: {"agent": "name_or_url"} - Config name or direct URL of the agent. + + Returns: + JSON string with agent name, description, skills, and capabilities. + """ + agent = args.get("agent", "").strip() + if not agent: + return json.dumps({"error": "Missing required 'agent' parameter (name or URL)"}) + + try: + url, config = _resolve_agent_url(agent) + result = _run_on_loop(_async_discover(url, config)) + return json.dumps(result, indent=2) + except Exception as e: + error_msg = _sanitize_error(str(e)) + logger.exception("A2A discover failed for '%s': %s", agent, error_msg) + return json.dumps({"error": f"Failed to discover agent: {error_msg}"}) + + +def a2a_call(args: dict, **kwargs) -> str: + """Send a task to a remote A2A agent. + + Args: + args: { + "agent": "name_or_url", - Config name or direct URL + "message": "text", - The task/message to send + "stream": false - Optional: use streaming mode + } + + Returns: + JSON string with the agent's response. + """ + agent = args.get("agent", "").strip() + message = args.get("message", "").strip() + stream = args.get("stream", False) + + if not agent: + return json.dumps({"error": "Missing required 'agent' parameter"}) + if not message: + return json.dumps({"error": "Missing required 'message' parameter"}) + + try: + url, config = _resolve_agent_url(agent) + result = _run_on_loop(_async_call(url, config, message, stream)) + return result + except Exception as e: + error_msg = _sanitize_error(str(e)) + logger.exception("A2A call failed for '%s': %s", agent, error_msg) + return json.dumps({"error": f"A2A call failed: {error_msg}"}) + + +def a2a_list(args: dict, **kwargs) -> str: + """List all known A2A agents from config and discovery cache. + + Returns: + JSON string with {agents: [...], total: int}. + """ + try: + agents = _build_agent_registry() + return json.dumps({"agents": agents, "total": len(agents)}, indent=2) + except Exception as e: + error_msg = _sanitize_error(str(e)) + logger.exception("A2A list failed: %s", error_msg) + return json.dumps({"error": f"Failed to list agents: {error_msg}"}) + + +def a2a_orchestrate(args: dict, **kwargs) -> str: + """Fan out a goal to multiple A2A agents in parallel. + + Args: + args: { + "goal": "text", - Required: task to send + "agents": ["name_or_url"], - Optional: explicit agent list + "mode": "all|first|best" - Optional: orchestration mode (default: all) + } + + Returns: + JSON string with orchestration results. + """ + goal = args.get("goal", "").strip() + if not goal: + return json.dumps({"error": "Missing required 'goal' parameter"}) + + mode = args.get("mode", "all").strip().lower() + if mode not in ("all", "first", "best"): + return json.dumps({"error": f"Invalid mode '{mode}'. Must be one of: all, first, best"}) + + try: + explicit_agents = args.get("agents", []) + if explicit_agents: + # Resolve each explicit agent + targets = [] + for agent in explicit_agents: + url, config = _resolve_agent_url(agent) + # Copy to avoid mutating cached config + config = dict(config) + config["_name"] = agent + targets.append((url, config)) + else: + # Auto-select from registry + targets = _auto_select_agents(goal) + # Tag with names + for idx, (url, cfg) in enumerate(targets): + cfg = dict(cfg) + cfg["_name"] = cfg.get("_name", url) + targets[idx] = (url, cfg) + + if not targets: + return json.dumps({"error": "No agents available for orchestration"}) + + # Compute total timeout (max agent timeout + buffer) + max_timeout = max( + (cfg.get("timeout", _DEFAULT_TIMEOUT) for _, cfg in targets), + default=_DEFAULT_TIMEOUT, + ) + result = _run_on_loop( + _async_orchestrate(goal, targets, mode), + timeout=max_timeout + 60, + ) + return json.dumps(result, indent=2) + + except Exception as e: + error_msg = _sanitize_error(str(e)) + logger.exception("A2A orchestrate failed: %s", error_msg) + return json.dumps({"error": f"Orchestration failed: {error_msg}"}) + + +# --------------------------------------------------------------------------- +# Availability check +# --------------------------------------------------------------------------- + +def check_a2a_available() -> bool: + """Return True if a2a-sdk is installed.""" + return _A2A_AVAILABLE + + +# --------------------------------------------------------------------------- +# Tool schemas (OpenAI function-calling format) +# --------------------------------------------------------------------------- + +A2A_DISCOVER_SCHEMA = { + "name": "a2a_discover", + "description": ( + "Discover a remote A2A (Agent-to-Agent) agent by fetching its Agent Card. " + "Returns the agent's name, description, skills, and capabilities. " + "Use this to learn what a remote agent can do before sending it tasks. " + "Accepts either a configured agent name or a direct URL." + ), + "parameters": { + "type": "object", + "properties": { + "agent": { + "type": "string", + "description": ( + "The agent to discover. Can be a name from config " + "(e.g. 'researcher') or a direct URL (e.g. 'http://localhost:9999')." + ), + }, + }, + "required": ["agent"], + }, +} + +A2A_CALL_SCHEMA = { + "name": "a2a_call", + "description": ( + "Send a task to a remote A2A agent and get its response. " + "The remote agent can be built on any framework (LangChain, CrewAI, " + "Google ADK, etc.) as long as it supports the A2A protocol. " + "Use a2a_discover first to see what the agent can do." + ), + "parameters": { + "type": "object", + "properties": { + "agent": { + "type": "string", + "description": ( + "The agent to call. Can be a name from config " + "or a direct URL." + ), + }, + "message": { + "type": "string", + "description": "The task or message to send to the remote agent.", + }, + "stream": { + "type": "boolean", + "description": "If true, use streaming mode for real-time responses. Default: false.", + "default": False, + }, + }, + "required": ["agent", "message"], + }, +} + +A2A_LIST_SCHEMA = { + "name": "a2a_list", + "description": ( + "List all known A2A agents from configuration and discovery cache. " + "Shows each agent's name, URL, status, skills, and whether it was " + "configured or dynamically discovered. Use this to see what agents " + "are available before orchestrating tasks." + ), + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, +} + +A2A_ORCHESTRATE_SCHEMA = { + "name": "a2a_orchestrate", + "description": ( + "Fan out a goal to multiple A2A agents in parallel and collect results. " + "Modes: 'all' sends to every agent and collects all responses, " + "'first' returns the first successful response and cancels the rest, " + "'best' is an alias for 'all'. If no agents are specified, auto-selects " + "agents whose skills match the goal." + ), + "parameters": { + "type": "object", + "properties": { + "goal": { + "type": "string", + "description": "The task or goal to send to the agents.", + }, + "agents": { + "type": "array", + "items": {"type": "string"}, + "description": ( + "Optional list of agent names or URLs. " + "If omitted, auto-selects agents based on skill matching." + ), + }, + "mode": { + "type": "string", + "enum": ["all", "first", "best"], + "description": ( + "Orchestration mode: 'all' collects every response, " + "'first' returns the first success, 'best' is an alias for 'all'. " + "Default: 'all'." + ), + "default": "all", + }, + }, + "required": ["goal"], + }, +} + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + +registry.register( + name="a2a_discover", + toolset="a2a", + schema=A2A_DISCOVER_SCHEMA, + handler=a2a_discover, + check_fn=check_a2a_available, + is_async=False, + description="Discover remote A2A agents and their capabilities", + emoji="🌐", +) + +registry.register( + name="a2a_call", + toolset="a2a", + schema=A2A_CALL_SCHEMA, + handler=a2a_call, + check_fn=check_a2a_available, + is_async=False, + description="Send tasks to remote A2A agents", + emoji="📡", +) + +registry.register( + name="a2a_list", + toolset="a2a", + schema=A2A_LIST_SCHEMA, + handler=a2a_list, + check_fn=check_a2a_available, + is_async=False, + description="List all known A2A agents", + emoji="📋", +) + +registry.register( + name="a2a_orchestrate", + toolset="a2a", + schema=A2A_ORCHESTRATE_SCHEMA, + handler=a2a_orchestrate, + check_fn=check_a2a_available, + is_async=False, + description="Fan out tasks to multiple A2A agents in parallel", + emoji="🎭", +) diff --git a/toolsets.py b/toolsets.py index 57e226d3c082e..3fda802123d5b 100644 --- a/toolsets.py +++ b/toolsets.py @@ -273,6 +273,12 @@ "includes": [] }, + "a2a": { + "description": "Discover and communicate with remote A2A (Agent-to-Agent) protocol agents", + "tools": ["a2a_discover", "a2a_call", "a2a_list", "a2a_orchestrate"], + "includes": [] + }, + # Scenario-specific toolsets