1321 lines
60 KiB
Python
1321 lines
60 KiB
Python
"""
|
|
A2A inbound platform adapter — exposes Hermes as an A2A-discoverable agent.
|
|
|
|
Design (the #11025 insight, done as a plugin with zero core edits):
|
|
- Runs a stdlib http.server in a daemon thread (no a2a-sdk, no asyncio loop
|
|
dependency at register() time — avoids the a2a_fleet "register outside a
|
|
loop" bug class).
|
|
- Serves the A2A v1.0 Agent Card at GET /.well-known/agent-card.json (and legacy agent.json).
|
|
- JSON-RPC at POST /: message/send, message/stream (SSE), tasks/get,
|
|
tasks/list, tasks/cancel, tasks/subscribe, tasks/pushNotificationConfig/create,
|
|
tasks/pushNotificationConfig/get, tasks/pushNotificationConfig/list,
|
|
tasks/pushNotificationConfig/delete.
|
|
- Push notifications: config accepted inline in message/send
|
|
(configuration.taskPushNotificationConfig) or via the create method;
|
|
payloads are v1.0 StreamResponse objects, HMAC-signed.
|
|
- Metrics at GET /metrics.
|
|
- Each inbound task is filtered + framed (security.wrap_inbound) and routed
|
|
into the agent's LIVE gateway session via the normal MessageEvent path, so
|
|
the agent that replies is the same one talking to its user — full memory
|
|
and context, not a throwaway clone.
|
|
- The agent's reply comes back through ``adapter.send()``; we override that to
|
|
fulfil a per-task Future the HTTP handler is blocked on, turning the
|
|
async gateway into a synchronous request/response for the A2A caller.
|
|
``on_processing_complete`` resolves failures/cancellations promptly.
|
|
- Every exchange is persisted to disk and audit-logged.
|
|
|
|
Bind safety: with no token configured, the server binds 127.0.0.1 only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections import deque
|
|
from concurrent.futures import Future
|
|
from concurrent.futures import TimeoutError as FuturesTimeout
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Any, Dict, Optional
|
|
|
|
from gateway.platforms.base import (
|
|
BasePlatformAdapter,
|
|
MessageEvent,
|
|
MessageType,
|
|
ProcessingOutcome,
|
|
SendResult,
|
|
)
|
|
from gateway.config import Platform
|
|
|
|
from . import protocol, security
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_DEFAULT_PORT = 9900
|
|
_ORPHAN_TIMEOUT = 300 # seconds before a pending task is considered orphaned
|
|
_WATCHDOG_INTERVAL = 60 # seconds between orphaned task watchdog runs
|
|
_MAX_BODY = 1_048_576 # 1MB max request body — prevents DoS via memory exhaustion
|
|
_SSE_KEEPALIVE = 5 # seconds between SSE keepalive comments
|
|
|
|
|
|
def _reply_timeout() -> float:
|
|
"""Seconds to wait for the agent to answer an inbound task."""
|
|
try:
|
|
return max(1.0, float(os.getenv("A2A_REPLY_TIMEOUT", "300")))
|
|
except (ValueError, TypeError):
|
|
return 300.0
|
|
|
|
|
|
def _profile_scoped() -> bool:
|
|
"""True when running inside a multiplexed secondary profile's scope.
|
|
|
|
Secondary-profile adapters are constructed inside ``_profile_runtime_scope``
|
|
(secret scope installed + multiplex active) — the same discriminator the
|
|
Buzz/SimpleX adapters use for this bug class (#98738). The DEFAULT profile
|
|
under multiplexing runs unscoped: ``os.environ`` holds its own bridge
|
|
output there and keeps its legacy precedence.
|
|
"""
|
|
try:
|
|
from agent.secret_scope import current_secret_scope, is_multiplex_active
|
|
|
|
return bool(is_multiplex_active() and current_secret_scope() is not None)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _default_agent_name() -> str:
|
|
# Scope-aware: inside a secondary multiplex profile, os.environ holds the
|
|
# DEFAULT profile's bridged A2A_AGENT_NAME — borrowing it would brand a
|
|
# secondary profile's Agent Card with another profile's identity. There
|
|
# is no per-profile config.yaml equivalent yet, so a scoped profile just
|
|
# falls through to the hostname-based default below instead.
|
|
name = "" if _profile_scoped() else os.getenv("A2A_AGENT_NAME", "").strip()
|
|
if name:
|
|
return name
|
|
try:
|
|
import socket
|
|
return f"hermes-{socket.gethostname()}"
|
|
except Exception:
|
|
return "hermes-agent"
|
|
|
|
|
|
def _clean_slug(value: str) -> str:
|
|
"""Return a URL-safe-ish single-segment slug for a served agent."""
|
|
slug = str(value or "").strip().strip("/")
|
|
return "" if slug in ("", "default", "root") else slug.split("/")[0]
|
|
|
|
|
|
def _join_url(base: str, prefix: str) -> str:
|
|
base = (base or "").strip() or "/"
|
|
if not base.endswith("/"):
|
|
base += "/"
|
|
prefix = (prefix or "").strip("/")
|
|
if not prefix:
|
|
return base
|
|
return urllib.parse.urljoin(base, prefix + "/")
|
|
|
|
|
|
def _active_profile_name() -> str:
|
|
try:
|
|
from hermes_cli.profiles import get_active_profile_name
|
|
return get_active_profile_name() or "default"
|
|
except Exception:
|
|
return os.getenv("HERMES_PROFILE", "default") or "default"
|
|
|
|
|
|
def _profile_home(profile: str) -> Optional[str]:
|
|
try:
|
|
from hermes_cli.profiles import get_profile_dir
|
|
return str(get_profile_dir(profile))
|
|
except Exception:
|
|
if not profile or profile == "default":
|
|
try:
|
|
from hermes_cli.config import get_hermes_home
|
|
return str(get_hermes_home())
|
|
except Exception:
|
|
return None
|
|
return os.path.expanduser(f"~/.hermes/profiles/{profile}")
|
|
|
|
def _safe_context_slug(value: str, max_len: int = 96) -> str:
|
|
"""Sanitize attacker-provided context ids before using in session titles."""
|
|
slug = re.sub(r"[^A-Za-z0-9_.-]+", "-", str(value or "")).strip("-._")
|
|
return (slug or "ctx")[:max_len]
|
|
|
|
|
|
def _method_info(method: str) -> tuple[str, bool]:
|
|
"""Return (canonical_operation, is_v1_method).
|
|
|
|
Canonical operation names are lowercase internal labels. v1 methods use the
|
|
PascalCase names from A2A v1.0 §5.3/§9.4; legacy aliases remain accepted.
|
|
"""
|
|
mapping = {
|
|
"SendMessage": ("send", True),
|
|
"message/send": ("send", False),
|
|
"SendStreamingMessage": ("stream", True),
|
|
"message/stream": ("stream", False),
|
|
"GetTask": ("get", True),
|
|
"tasks/get": ("get", False),
|
|
"ListTasks": ("list", True),
|
|
"tasks/list": ("list", False),
|
|
"CancelTask": ("cancel", True),
|
|
"tasks/cancel": ("cancel", False),
|
|
"SubscribeToTask": ("subscribe", True),
|
|
"tasks/subscribe": ("subscribe", False),
|
|
"CreateTaskPushNotificationConfig": ("push_create", True),
|
|
"tasks/pushNotificationConfig/create": ("push_create", False),
|
|
"tasks/pushNotificationConfig/set": ("push_create", False),
|
|
"tasks/pushNotification/set": ("push_create", False),
|
|
"GetTaskPushNotificationConfig": ("push_get", True),
|
|
"tasks/pushNotificationConfig/get": ("push_get", False),
|
|
"ListTaskPushNotificationConfigs": ("push_list", True),
|
|
"tasks/pushNotificationConfig/list": ("push_list", False),
|
|
"DeleteTaskPushNotificationConfig": ("push_delete", True),
|
|
"tasks/pushNotificationConfig/delete": ("push_delete", False),
|
|
}
|
|
return mapping.get(method, ("", False))
|
|
|
|
|
|
class _A2AServer(ThreadingHTTPServer):
|
|
"""ThreadingHTTPServer that carries a reference to its adapter."""
|
|
|
|
daemon_threads = True
|
|
|
|
def __init__(self, addr, handler_cls, adapter: "A2AAdapter"):
|
|
super().__init__(addr, handler_cls)
|
|
self.adapter = adapter
|
|
|
|
|
|
class A2ARequestHandler(BaseHTTPRequestHandler):
|
|
"""HTTP handler for the A2A JSON-RPC surface.
|
|
|
|
Module-level (not a closure) so request routing is unit-testable; all
|
|
state lives on ``self.server.adapter``.
|
|
"""
|
|
|
|
@property
|
|
def adapter(self) -> "A2AAdapter":
|
|
return self.server.adapter # type: ignore[attr-defined]
|
|
|
|
# Silence the default stderr access log.
|
|
def log_message(self, format, *args): # noqa: A002,N802
|
|
logger.debug("A2A http: " + format, *args)
|
|
|
|
def _json(self, code: int, payload: dict):
|
|
body = json.dumps(payload).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _request_public_url(self) -> str:
|
|
"""Derive the routable URL for this request.
|
|
|
|
Priority: A2A_PUBLIC_URL env > X-Forwarded-Host / Host header (with
|
|
scheme from X-Forwarded-Proto) > empty. Empty means "caller has no
|
|
info, fall back to bind host". See gfdsa's k8s bind-host bug report
|
|
(PR #41711).
|
|
"""
|
|
explicit = os.getenv("A2A_PUBLIC_URL", "").strip()
|
|
if explicit:
|
|
return explicit
|
|
host = self.headers.get("X-Forwarded-Host", "") or self.headers.get("Host", "")
|
|
if not host:
|
|
return ""
|
|
host = host.split(",")[0].strip()
|
|
scheme = (self.headers.get("X-Forwarded-Proto", "") or "http").split(",")[0].strip()
|
|
return f"{scheme}://{host}/"
|
|
|
|
def do_GET(self): # noqa: N802
|
|
route = self.adapter._route_for_path(self.path)
|
|
agent = route["agent"]
|
|
subpath = route["subpath"].rstrip("/") or "/"
|
|
if subpath in ("/.well-known/agent.json", "/.well-known/agent-card.json"):
|
|
public_url = self._request_public_url() or None
|
|
self._json(200, self.adapter._build_card(public_url, agent=agent))
|
|
return
|
|
if subpath in ("/", "/health"):
|
|
payload = {
|
|
"status": "ok",
|
|
"agent": agent.get("name") or self.adapter.agent_name,
|
|
}
|
|
# Do not leak profile/tenant topology on remote unauthenticated GETs.
|
|
# Agent Cards are intentionally public; health topology is not.
|
|
if self.adapter._security_context.localhost_only() or self.adapter._security_context.authenticate(
|
|
self.headers.get("Authorization"),
|
|
self.client_address[0] if self.client_address else "",
|
|
) is not None:
|
|
payload["served_agents"] = self.adapter._served_agent_summary(
|
|
public_url=self._request_public_url() or None)
|
|
self._json(200, payload)
|
|
return
|
|
if subpath == "/metrics":
|
|
self._json(200, protocol.metrics.snapshot())
|
|
return
|
|
self._json(404, {"error": "not found"})
|
|
|
|
def do_POST(self): # noqa: N802
|
|
adapter = self.adapter
|
|
client_ip = self.client_address[0] if self.client_address else ""
|
|
|
|
# Identity comes from the presented credential (or the socket in
|
|
# localhost-only mode) — never from the request body.
|
|
identity = adapter._security_context.authenticate(
|
|
self.headers.get("Authorization"), client_ip
|
|
)
|
|
if identity is None:
|
|
self._json(401, protocol.jsonrpc_error(None, protocol.ERR_UNAUTHORIZED, "unauthorized"))
|
|
return
|
|
|
|
try:
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length > _MAX_BODY:
|
|
self._json(413, protocol.jsonrpc_error(None, protocol.ERR_PARSE, "payload too large"))
|
|
return
|
|
raw = self.rfile.read(length) if length else b"{}"
|
|
req = json.loads(raw.decode("utf-8"))
|
|
except Exception:
|
|
self._json(400, protocol.jsonrpc_error(None, protocol.ERR_PARSE, "parse error"))
|
|
return
|
|
|
|
if not isinstance(req, dict):
|
|
self._json(400, protocol.jsonrpc_error(None, protocol.ERR_INVALID_PARAMS, "JSON-RPC request must be an object"))
|
|
return
|
|
|
|
req_id = req.get("id")
|
|
method = str(req.get("method", ""))
|
|
params = req.get("params", {})
|
|
if params is None:
|
|
params = {}
|
|
if not isinstance(params, dict):
|
|
self._json(200, protocol.jsonrpc_error(req_id, protocol.ERR_INVALID_PARAMS, "params must be an object"))
|
|
return
|
|
|
|
version = (self.headers.get("A2A-Version") or "").strip()
|
|
if version and version not in {"1.0", "1.0.0"}:
|
|
self._json(200, protocol.jsonrpc_error(req_id, protocol.ERR_INVALID_PARAMS, f"unsupported A2A-Version: {version}"))
|
|
return
|
|
|
|
operation, is_v1 = _method_info(method)
|
|
route = adapter._route_for_request(self.path, params)
|
|
if route.get("error"):
|
|
self._json(400, protocol.jsonrpc_error(req_id, protocol.ERR_INVALID_PARAMS, route["error"]))
|
|
return
|
|
agent = route["agent"]
|
|
|
|
if not adapter._rate_limiter.allow(identity):
|
|
protocol.metrics.rate_limit_triggers += 1
|
|
self._json(429, protocol.jsonrpc_error(req_id, protocol.ERR_RATE_LIMITED, "rate limit exceeded"))
|
|
return
|
|
|
|
if not adapter._security_context.is_trusted_peer(identity):
|
|
self._json(403, protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_UNTRUSTED_PEER, f"peer '{identity}' not trusted"))
|
|
return
|
|
|
|
if not operation:
|
|
self._json(200, protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_METHOD_NOT_FOUND, f"method not found: {method}"))
|
|
return
|
|
|
|
if operation == "send":
|
|
self._json(200, adapter._rpc_message_send(req_id, params, identity, agent=agent, v1_response=is_v1))
|
|
return
|
|
if operation == "stream":
|
|
adapter._rpc_message_stream(self, req_id, params, identity, agent=agent)
|
|
return
|
|
if operation == "get":
|
|
self._json(200, adapter._rpc_tasks_get(req_id, params, agent=agent))
|
|
return
|
|
if operation == "list":
|
|
self._json(200, adapter._rpc_tasks_list(req_id, params, agent=agent))
|
|
return
|
|
if operation == "cancel":
|
|
self._json(200, adapter._rpc_tasks_cancel(req_id, params, agent=agent))
|
|
return
|
|
if operation == "subscribe":
|
|
adapter._rpc_tasks_subscribe(self, req_id, params, agent=agent)
|
|
return
|
|
if operation == "push_create":
|
|
self._json(200, adapter._rpc_push_config_create(req_id, params, agent=agent))
|
|
return
|
|
if operation == "push_get":
|
|
self._json(200, adapter._rpc_push_config_get(req_id, params, agent=agent))
|
|
return
|
|
if operation == "push_list":
|
|
self._json(200, adapter._rpc_push_config_list(req_id, params, agent=agent))
|
|
return
|
|
if operation == "push_delete":
|
|
self._json(200, adapter._rpc_push_config_delete(req_id, params, agent=agent))
|
|
return
|
|
|
|
|
|
|
|
class A2AAdapter(BasePlatformAdapter):
|
|
"""Inbound A2A server adapter."""
|
|
|
|
def __init__(self, config, **kwargs):
|
|
platform = Platform("a2a")
|
|
super().__init__(config=config, platform=platform)
|
|
|
|
extra = getattr(config, "extra", {}) or {}
|
|
# Scope-aware: a secondary multiplex profile must not borrow the
|
|
# default profile's bridged A2A_PORT (mirrors the Buzz/SimpleX fix
|
|
# for #98738) — an unconfigured profile falls closed to the module
|
|
# default port instead. (advertised_toolsets has the same env-leak
|
|
# shape but is left unscoped here — see the "Scope note" in this
|
|
# fix's PR description: open PR #98937 is actively rewriting this
|
|
# field's None-vs-empty-list semantics.)
|
|
self._security_context = security.A2ASecurityContext.capture()
|
|
_port_env = None if _profile_scoped() else os.getenv("A2A_PORT")
|
|
self.port = int(_port_env or extra.get("port", _DEFAULT_PORT))
|
|
self.host = self._security_context.resolve_bind_host()
|
|
self.agent_name = _default_agent_name()
|
|
self._advertised_toolsets = [
|
|
t.strip() for t in (
|
|
list(extra.get("advertised_toolsets") or [])
|
|
or os.getenv("A2A_ADVERTISED_TOOLSETS", "").split(",")
|
|
) if str(t).strip()
|
|
]
|
|
self._active_profile = _active_profile_name()
|
|
self._agents = self._load_served_agents(extra)
|
|
|
|
self._httpd: Optional[_A2AServer] = None
|
|
self._server_thread: Optional[threading.Thread] = None
|
|
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
|
|
|
# Per-adapter protocol state (not module-global): task store, anti-loop
|
|
# turn tracking, and rate limiting.
|
|
self.tasks = protocol.TaskStore()
|
|
self._turns = protocol.TurnTracker()
|
|
self._rate_limiter = protocol.RateLimiter()
|
|
|
|
# Forwarded profile sessions: map (profile, agent_slug, context_id) -> session_id.
|
|
self._profile_sessions: Dict[tuple[str, str, str], str] = {}
|
|
self._profile_session_locks: Dict[tuple[str, str, str], threading.Lock] = {}
|
|
self._profile_session_locks_guard = threading.Lock()
|
|
|
|
# Pending reply futures, keyed by task_id. Each future resolves to a
|
|
# (state, text) tuple. _pending_order keeps per-context FIFO order so
|
|
# adapter.send() — which only knows the context — resolves the oldest
|
|
# outstanding task for that context (no cross-talk between concurrent
|
|
# requests sharing a context).
|
|
self._pending: Dict[str, tuple[str, Future]] = {}
|
|
self._pending_order: Dict[str, deque[str]] = {}
|
|
self._pending_lock = threading.Lock()
|
|
|
|
# Orphaned task watchdog
|
|
self._watchdog_stop = threading.Event()
|
|
self._watchdog_thread: Optional[threading.Thread] = None
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "A2A"
|
|
|
|
@property
|
|
def authorization_is_upstream(self) -> bool:
|
|
"""A2A authenticates every inbound request via bearer token (or
|
|
localhost-only binding) in ``do_POST`` before dispatch — the identity
|
|
is already authorized upstream. Without this override, the gateway's
|
|
per-platform user allow-list (``{PLATFORM}_ALLOWED_USERS``) rejects
|
|
A2A peers because their identity is a token-derived name or pod IP,
|
|
not a platform account the operator configures in an env allow-list.
|
|
|
|
This is authorization delegated to the A2A bearer-token transport,
|
|
not a fail-open: every request is 401'd if the credential is wrong.
|
|
Reported by kuangmi-bit (PR #41711 comment, Jun 27).
|
|
"""
|
|
return True
|
|
|
|
# ── Lifecycle ─────────────────────────────────────────────────────────
|
|
|
|
async def connect(self, **_kwargs) -> bool:
|
|
# Gateway reconnection plumbing passes adapter-agnostic kwargs such as
|
|
# ``is_reconnect``. A2A does not need them, but accepting them keeps the
|
|
# plugin compatible with the BasePlatformAdapter lifecycle contract.
|
|
# Capture the running gateway loop so the HTTP thread can marshal
|
|
# events onto it via run_coroutine_threadsafe.
|
|
try:
|
|
self._loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
self._loop = None
|
|
|
|
try:
|
|
self._httpd = _A2AServer((self.host, self.port), A2ARequestHandler, self)
|
|
except OSError as e:
|
|
logger.error("A2A: could not bind %s:%s — %s", self.host, self.port, e)
|
|
self._set_fatal_error("bind_failed", f"A2A bind failed: {e}", retryable=True)
|
|
return False
|
|
|
|
self._server_thread = threading.Thread(
|
|
target=self._httpd.serve_forever,
|
|
name="a2a-http",
|
|
daemon=True,
|
|
)
|
|
self._server_thread.start()
|
|
|
|
# Reset watchdog state for reconnection (disconnect sets the event)
|
|
self._watchdog_stop.clear()
|
|
self._watchdog_thread = threading.Thread(
|
|
target=self._watchdog_loop,
|
|
name="a2a-watchdog",
|
|
daemon=True,
|
|
)
|
|
self._watchdog_thread.start()
|
|
|
|
self._mark_connected()
|
|
|
|
exposure = (
|
|
"localhost-only"
|
|
if self._security_context.localhost_only()
|
|
else "REMOTE (bearer auth)"
|
|
)
|
|
logger.info(
|
|
"A2A: serving Agent Card + JSON-RPC on http://%s:%s (%s) as %r; %d routed agent(s)",
|
|
self.host, self.port, exposure, self.agent_name, len(self._agents),
|
|
)
|
|
# Plugin-registered native handlers (ctx.register_platform_handler).
|
|
self._wire_plugin_handlers(None)
|
|
return True
|
|
|
|
async def disconnect(self) -> None:
|
|
self._mark_disconnected()
|
|
self._watchdog_stop.set()
|
|
if self._httpd is not None:
|
|
try:
|
|
self._httpd.shutdown()
|
|
self._httpd.server_close()
|
|
except Exception:
|
|
pass
|
|
self._httpd = None
|
|
# Fail any in-flight replies so blocked HTTP threads don't hang.
|
|
with self._pending_lock:
|
|
for _ctx, fut in self._pending.values():
|
|
if not fut.done():
|
|
fut.set_result((protocol.STATE_FAILED, "[agent shutting down]"))
|
|
self._pending.clear()
|
|
self._pending_order.clear()
|
|
|
|
# ── Orphaned task watchdog ─────────────────────────────────────────────
|
|
|
|
def _watchdog_loop(self) -> None:
|
|
"""Background thread that fails orphaned tasks (keeps them queryable)."""
|
|
while not self._watchdog_stop.wait(_WATCHDOG_INTERVAL):
|
|
try:
|
|
for tid in self.tasks.fail_orphans(_ORPHAN_TIMEOUT):
|
|
logger.warning("A2A: orphaned task %s marked failed (timeout %ds)", tid, _ORPHAN_TIMEOUT)
|
|
protocol.metrics.tasks_failed += 1
|
|
except Exception:
|
|
logger.debug("A2A: watchdog error", exc_info=True)
|
|
|
|
# ── Agent routing + Agent Cards ───────────────────────────────────────
|
|
|
|
def _load_global_a2a_config(self) -> dict:
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
cfg = load_config() or {}
|
|
return cfg if isinstance(cfg, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
def _load_served_agents(self, extra: dict) -> dict[str, dict]:
|
|
"""Load served-agent routing config.
|
|
|
|
Preferred config location is ``platforms.a2a.extra.agents``. A top-level
|
|
``a2a_served_agents`` fallback is accepted for scripts/tests. Root/default
|
|
always maps to the live gateway session for backward compatibility.
|
|
"""
|
|
raw = extra.get("agents") or extra.get("served_agents")
|
|
if raw is None:
|
|
cfg = self._load_global_a2a_config()
|
|
raw = cfg.get("a2a_served_agents") or (cfg.get("a2a") or {}).get("served_agents")
|
|
|
|
agents: dict[str, dict] = {}
|
|
# Scope-aware for the same reason as port/toolsets above: a secondary
|
|
# profile must not inherit the default profile's A2A_AGENT_DESCRIPTION.
|
|
default_desc = (
|
|
"Hermes Agent — a general-purpose agent reachable over A2A."
|
|
if _profile_scoped()
|
|
else os.getenv(
|
|
"A2A_AGENT_DESCRIPTION",
|
|
"Hermes Agent — a general-purpose agent reachable over A2A.",
|
|
)
|
|
)
|
|
agents[""] = {
|
|
"slug": "",
|
|
"path": "",
|
|
"tenant": "",
|
|
"profile": self._active_profile,
|
|
"local": True,
|
|
"name": self.agent_name,
|
|
"description": default_desc,
|
|
"advertised_toolsets": self._advertised_toolsets,
|
|
}
|
|
|
|
reserved = {"health", "metrics", ".well-known"}
|
|
tenants: dict[str, str] = {}
|
|
items = raw.items() if isinstance(raw, dict) else enumerate(raw or []) if isinstance(raw, list) else []
|
|
for key, val in items:
|
|
if not isinstance(val, dict):
|
|
continue
|
|
slug = _clean_slug(str(val.get("slug") or val.get("id") or key))
|
|
if not slug:
|
|
continue
|
|
path_segment = _clean_slug(str(val.get("path") or slug))
|
|
if not path_segment or path_segment in reserved:
|
|
logger.warning("A2A: ignoring served agent %r with reserved/invalid path %r", slug, path_segment)
|
|
continue
|
|
profile = str(val.get("profile") or slug).strip()
|
|
path = "/" + path_segment
|
|
toolsets = val.get("advertised_toolsets") or val.get("toolsets") or val.get("capabilities") or []
|
|
if isinstance(toolsets, str):
|
|
toolsets = [t.strip() for t in toolsets.split(",") if t.strip()]
|
|
local = bool(val.get("local")) or profile in ("", "default", self._active_profile)
|
|
tenant = str(val.get("tenant") or slug).strip()
|
|
if tenant:
|
|
if tenant in tenants:
|
|
logger.warning(
|
|
"A2A: ignoring served agent %r with duplicate tenant %r already used by %r",
|
|
slug, tenant, tenants[tenant],
|
|
)
|
|
continue
|
|
tenants[tenant] = slug
|
|
agents[slug] = {
|
|
"slug": slug,
|
|
"path": path,
|
|
"tenant": tenant,
|
|
"profile": profile or slug,
|
|
"local": local,
|
|
"name": str(val.get("name") or f"Hermes {slug}"),
|
|
"description": str(val.get("description") or f"Hermes profile '{profile or slug}' exposed over A2A."),
|
|
"advertised_toolsets": list(toolsets or []),
|
|
"timeout": int(val.get("timeout") or _reply_timeout()),
|
|
}
|
|
return agents
|
|
|
|
def _served_agent_summary(self, public_url: Optional[str] = None) -> list[dict]:
|
|
base = (public_url or "").strip() or f"http://{self.host}:{self.port}/"
|
|
return [
|
|
{
|
|
"slug": a["slug"] or "default",
|
|
"name": a.get("name"),
|
|
"url": _join_url(base, a.get("path", "")),
|
|
"tenant": a.get("tenant") or None,
|
|
"profile": a.get("profile"),
|
|
"local": bool(a.get("local")),
|
|
}
|
|
for a in self._agents.values()
|
|
]
|
|
|
|
def _route_for_path(self, raw_path: str) -> dict:
|
|
path = urllib.parse.urlsplit(raw_path or "/").path or "/"
|
|
# Longest prefix wins. Default/root agent is the fallback.
|
|
for agent in sorted(self._agents.values(), key=lambda a: len(a.get("path", "")), reverse=True):
|
|
prefix = agent.get("path", "") or ""
|
|
if prefix and (path == prefix or path.startswith(prefix + "/")):
|
|
subpath = path[len(prefix):] or "/"
|
|
if not subpath.startswith("/"):
|
|
subpath = "/" + subpath
|
|
return {"agent": agent, "subpath": subpath}
|
|
return {"agent": self._agents[""], "subpath": path}
|
|
|
|
def _route_for_request(self, raw_path: str, params: dict) -> dict:
|
|
route = self._route_for_path(raw_path)
|
|
agent = route["agent"]
|
|
tenant = str((params or {}).get("tenant") or "")
|
|
# If no URL prefix chose a non-default agent, allow v1.0 tenant routing.
|
|
if agent.get("slug") == "" and tenant:
|
|
matches = [a for a in self._agents.values() if a.get("tenant") == tenant]
|
|
if matches:
|
|
route = {"agent": matches[0], "subpath": route["subpath"]}
|
|
agent = matches[0]
|
|
expected = str(agent.get("tenant") or "")
|
|
if tenant and expected and tenant != expected:
|
|
return {"error": f"tenant {tenant!r} does not match routed agent {agent.get('slug') or 'default'}"}
|
|
return route
|
|
|
|
def _build_card(self, public_url: Optional[str] = None, agent: Optional[dict] = None) -> dict:
|
|
# Prefer per-request public URL (from X-Forwarded-Host / Host /
|
|
# A2A_PUBLIC_URL) over bind host, so peers can call back when we're
|
|
# behind a reverse proxy.
|
|
agent = agent or self._agents[""]
|
|
base = (public_url or "").strip() or f"http://{self.host}:{self.port}/"
|
|
url = _join_url(base, agent.get("path", ""))
|
|
return protocol.build_agent_card(
|
|
name=agent.get("name") or self.agent_name,
|
|
url=url,
|
|
description=agent.get("description") or "Hermes Agent — a general-purpose agent reachable over A2A.",
|
|
skills=self._advertised_skills(agent),
|
|
streaming=bool(agent.get("local", True)),
|
|
push_notifications=True,
|
|
auth_required=not self._security_context.localhost_only(),
|
|
tenant=str(agent.get("tenant") or ""),
|
|
)
|
|
|
|
def _advertised_skills(self, agent: Optional[dict] = None) -> list[dict]:
|
|
"""Dynamic Agent Card skills from the live tool registry.
|
|
|
|
The card reflects what the agent can actually do right now. An
|
|
explicit ``advertised_toolsets`` config (or A2A_ADVERTISED_TOOLSETS)
|
|
restricts what we advertise; without a registry we fall back to that
|
|
static list.
|
|
"""
|
|
try:
|
|
from tools.registry import registry as tool_registry
|
|
names = tool_registry.get_registered_toolset_names()
|
|
configured = (agent or {}).get("advertised_toolsets") if agent else self._advertised_toolsets
|
|
allowed = set(configured or []) or None
|
|
mapping = {
|
|
n: tool_registry.get_tool_names_for_toolset(n)
|
|
for n in names
|
|
if allowed is None or n in allowed
|
|
}
|
|
if mapping:
|
|
return protocol.skills_from_toolsets(mapping)
|
|
except Exception:
|
|
logger.debug("A2A: tool registry unavailable for Agent Card", exc_info=True)
|
|
configured = (agent or {}).get("advertised_toolsets") if agent else self._advertised_toolsets
|
|
return protocol.skills_from_toolsets(configured or [])
|
|
|
|
# ── Pending reply plumbing ────────────────────────────────────────────
|
|
|
|
def _add_pending(self, task_id: str, context_id: str) -> Future:
|
|
fut: Future = Future()
|
|
with self._pending_lock:
|
|
self._pending[task_id] = (context_id, fut)
|
|
self._pending_order.setdefault(context_id, deque()).append(task_id)
|
|
return fut
|
|
|
|
def _pop_pending(self, task_id: str) -> None:
|
|
with self._pending_lock:
|
|
entry = self._pending.pop(task_id, None)
|
|
if entry:
|
|
order = self._pending_order.get(entry[0])
|
|
if order:
|
|
try:
|
|
order.remove(task_id)
|
|
except ValueError:
|
|
pass
|
|
if not order:
|
|
self._pending_order.pop(entry[0], None)
|
|
|
|
def _resolve_task(self, task_id: str, state: str, text: str) -> bool:
|
|
with self._pending_lock:
|
|
entry = self._pending.get(task_id)
|
|
if entry and not entry[1].done():
|
|
entry[1].set_result((state, text))
|
|
return True
|
|
return False
|
|
|
|
def _resolve_oldest_for_context(self, context_id: str, state: str, text: str) -> bool:
|
|
with self._pending_lock:
|
|
for task_id in self._pending_order.get(context_id, ()):
|
|
entry = self._pending.get(task_id)
|
|
if entry and not entry[1].done():
|
|
entry[1].set_result((state, text))
|
|
return True
|
|
return False
|
|
|
|
def _scope_for_agent(self, agent: Optional[dict]) -> tuple[str, str]:
|
|
agent = agent or self._agents[""]
|
|
return str(agent.get("slug") or ""), str(agent.get("tenant") or "")
|
|
|
|
def _forward_lock(self, key: tuple[str, str, str]) -> threading.Lock:
|
|
with self._profile_session_locks_guard:
|
|
lock = self._profile_session_locks.get(key)
|
|
if lock is None:
|
|
lock = threading.Lock()
|
|
self._profile_session_locks[key] = lock
|
|
return lock
|
|
|
|
# ── Inbound task handling ─────────────────────────────────────────────
|
|
|
|
def _prepare_task(self, params: dict, peer: str, agent: Optional[dict] = None) -> tuple[Optional[dict], Optional[dict]]:
|
|
"""Validate, register, and dispatch an inbound message.
|
|
|
|
Returns (terminal_task, None) when the task ends immediately
|
|
(rejected / not ready), else (None, pending) where pending carries
|
|
the future the caller must wait on. Runs on an HTTP worker thread.
|
|
"""
|
|
agent = agent or self._agents[""]
|
|
text = protocol.extract_text(params)
|
|
context_id = protocol.extract_context_id(params) or protocol.new_context_id()
|
|
task_id = protocol.new_task_id()
|
|
|
|
# Anti-loop ping-pong protection
|
|
turn = self._turns.track(context_id)
|
|
if turn > protocol.max_pingpong_turns():
|
|
protocol.metrics.anti_loop_triggers += 1
|
|
logger.warning("A2A: anti-loop triggered for context %s (turn %d > %d)",
|
|
context_id, turn, protocol.max_pingpong_turns())
|
|
rec = self.tasks.create(task_id, context_id, peer, *self._scope_for_agent(agent))
|
|
self.tasks.complete(task_id, protocol.STATE_REJECTED, "")
|
|
return protocol.build_task(
|
|
task_id, context_id, protocol.STATE_REJECTED,
|
|
f"Anti-loop protection: context {context_id} exceeded "
|
|
f"{protocol.max_pingpong_turns()} turns. Start a new context or "
|
|
f"increase A2A_MAX_PINGPONG_TURNS.",
|
|
created_at=rec["created_iso"],
|
|
), None
|
|
|
|
if not text:
|
|
rec = self.tasks.create(task_id, context_id, peer, *self._scope_for_agent(agent))
|
|
self.tasks.complete(task_id, protocol.STATE_REJECTED, "")
|
|
return protocol.build_task(
|
|
task_id, context_id, protocol.STATE_REJECTED,
|
|
"Empty task — nothing to do.", created_at=rec["created_iso"],
|
|
), None
|
|
|
|
framed = security.wrap_inbound(peer, text)
|
|
security.audit("inbound", peer, task_id, text)
|
|
protocol.persist_message(context_id, "user", text, task_id)
|
|
protocol.metrics.inbound_total += 1
|
|
|
|
rec = self.tasks.create(task_id, context_id, peer, *self._scope_for_agent(agent))
|
|
self._register_inline_push(task_id, params, agent=agent)
|
|
|
|
if not agent.get("local", True):
|
|
reply, state = self._forward_to_profile(agent, peer, context_id, framed)
|
|
self.tasks.complete(task_id, state, reply)
|
|
protocol.persist_message(context_id, "agent", reply, task_id)
|
|
security.audit("outbound", peer, task_id, reply)
|
|
if state == protocol.STATE_COMPLETED:
|
|
protocol.metrics.outbound_total += 1
|
|
protocol.metrics.tasks_completed += 1
|
|
else:
|
|
protocol.metrics.tasks_failed += 1
|
|
self._send_push_notification(task_id, context_id, reply, state)
|
|
return protocol.build_task(task_id, context_id, state, reply, created_at=rec["created_iso"]), None
|
|
|
|
if self._loop is None or self._message_handler is None:
|
|
self.tasks.complete(task_id, protocol.STATE_FAILED, "")
|
|
protocol.metrics.tasks_failed += 1
|
|
return protocol.build_task(
|
|
task_id, context_id, protocol.STATE_FAILED,
|
|
"Agent gateway not ready to accept A2A tasks.",
|
|
created_at=rec["created_iso"],
|
|
), None
|
|
|
|
fut = self._add_pending(task_id, context_id)
|
|
|
|
event = MessageEvent(
|
|
text=framed,
|
|
message_type=MessageType.TEXT,
|
|
source=self.build_source(
|
|
chat_id=context_id,
|
|
chat_name=f"a2a:{peer}",
|
|
chat_type="dm",
|
|
user_id=peer,
|
|
user_name=peer,
|
|
),
|
|
message_id=task_id,
|
|
)
|
|
|
|
try:
|
|
asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
|
|
except Exception as e:
|
|
self._pop_pending(task_id)
|
|
msg = security.redact_outbound(f"Dispatch failed: {e}")
|
|
self.tasks.complete(task_id, protocol.STATE_FAILED, msg)
|
|
protocol.metrics.tasks_failed += 1
|
|
return protocol.build_task(
|
|
task_id, context_id, protocol.STATE_FAILED, msg,
|
|
created_at=rec["created_iso"],
|
|
), None
|
|
|
|
self.tasks.set_state(task_id, protocol.STATE_WORKING)
|
|
return None, {
|
|
"task_id": task_id,
|
|
"context_id": context_id,
|
|
"peer": peer,
|
|
"future": fut,
|
|
"created_iso": rec["created_iso"],
|
|
"started": time.time(),
|
|
}
|
|
|
|
def _profile_state_db(self, profile: str) -> Optional[str]:
|
|
home = _profile_home(profile)
|
|
if not home:
|
|
return None
|
|
return os.path.join(home, "state.db")
|
|
|
|
def _lookup_forward_session(self, profile: str, title: str) -> str:
|
|
db = self._profile_state_db(profile)
|
|
if not db or not os.path.exists(db):
|
|
return ""
|
|
try:
|
|
con = sqlite3.connect(db, timeout=5)
|
|
row = con.execute(
|
|
"SELECT id FROM sessions WHERE title = ? ORDER BY started_at DESC LIMIT 1",
|
|
(title,),
|
|
).fetchone()
|
|
con.close()
|
|
return str(row[0]) if row else ""
|
|
except Exception:
|
|
logger.debug("A2A: could not lookup forwarded session", exc_info=True)
|
|
return ""
|
|
|
|
def _latest_a2a_session(self, profile: str, started_after: float) -> str:
|
|
db = self._profile_state_db(profile)
|
|
if not db or not os.path.exists(db):
|
|
return ""
|
|
try:
|
|
con = sqlite3.connect(db, timeout=5)
|
|
row = con.execute(
|
|
"SELECT id FROM sessions WHERE source = 'a2a' AND started_at >= ? ORDER BY started_at DESC LIMIT 1",
|
|
(started_after - 2.0,),
|
|
).fetchone()
|
|
con.close()
|
|
return str(row[0]) if row else ""
|
|
except Exception:
|
|
logger.debug("A2A: could not find latest forwarded session", exc_info=True)
|
|
return ""
|
|
|
|
def _title_forward_session(self, profile: str, session_id: str, title: str) -> None:
|
|
db = self._profile_state_db(profile)
|
|
if not db or not os.path.exists(db) or not session_id:
|
|
return
|
|
try:
|
|
con = sqlite3.connect(db, timeout=5)
|
|
con.execute("UPDATE sessions SET title = ? WHERE id = ?", (title, session_id))
|
|
con.commit()
|
|
con.close()
|
|
except Exception:
|
|
logger.debug("A2A: could not title forwarded session", exc_info=True)
|
|
|
|
def _forward_to_profile(self, agent: dict, peer: str, context_id: str, framed_text: str) -> tuple[str, str]:
|
|
"""Forward a routed A2A task to another local Hermes profile.
|
|
|
|
First contact creates a normal ``source=a2a`` CLI session, records its
|
|
session id, and titles it deterministically. Later turns resume by the
|
|
concrete session id, not by a non-existent name. The public CLI boundary
|
|
is preserved while giving A2A contexts stable multi-turn continuity.
|
|
"""
|
|
profile = str(agent.get("profile") or agent.get("slug") or "").strip()
|
|
slug = str(agent.get("slug") or profile or "agent")
|
|
safe_ctx = _safe_context_slug(context_id)
|
|
session_title = f"a2a-{slug}-{safe_ctx}"
|
|
key = (profile or "default", slug, safe_ctx)
|
|
timeout = int(agent.get("timeout") or _reply_timeout())
|
|
|
|
lock = self._forward_lock(key)
|
|
with lock:
|
|
session_id = self._profile_sessions.get(key) or self._lookup_forward_session(profile, session_title)
|
|
cmd = ["hermes", "chat", "-q", framed_text, "-Q", "--source", "a2a"]
|
|
if session_id:
|
|
cmd.extend(["--resume", session_id])
|
|
|
|
env = os.environ.copy()
|
|
home = _profile_home(profile)
|
|
if home:
|
|
env["HERMES_HOME"] = home
|
|
env["HERMES_A2A_PEER"] = peer
|
|
start = time.time()
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd, capture_output=True, text=True, timeout=timeout,
|
|
env=env, check=False, stdin=subprocess.DEVNULL,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return "[profile did not reply in time]", protocol.STATE_FAILED
|
|
except Exception as e:
|
|
return security.redact_outbound(f"Profile dispatch failed: {e}"), protocol.STATE_FAILED
|
|
if proc.returncode != 0:
|
|
msg = (proc.stderr or proc.stdout or f"profile exited {proc.returncode}").strip()
|
|
return security.redact_outbound(msg[-2000:]), protocol.STATE_FAILED
|
|
if not session_id:
|
|
session_id = self._latest_a2a_session(profile, start)
|
|
if session_id:
|
|
self._profile_sessions[key] = session_id
|
|
self._title_forward_session(profile, session_id, session_title)
|
|
return security.redact_outbound((proc.stdout or "").strip()), protocol.STATE_COMPLETED
|
|
|
|
def _finalize_task(self, pending: dict, state: str, reply: str) -> tuple[str, str]:
|
|
"""Record the outcome of a dispatched task. Returns (state, reply) after
|
|
redaction and input-required detection."""
|
|
task_id = pending["task_id"]
|
|
context_id = pending["context_id"]
|
|
peer = pending["peer"]
|
|
self._pop_pending(task_id)
|
|
|
|
reply = security.redact_outbound(reply or "")
|
|
|
|
# The agent flags clarification requests with a leading marker; map
|
|
# them to the A2A input-required state so the peer knows to answer.
|
|
if state == protocol.STATE_COMPLETED:
|
|
stripped = reply.lstrip()
|
|
if stripped.upper().startswith(protocol.INPUT_REQUIRED_MARKER):
|
|
state = protocol.STATE_INPUT_REQUIRED
|
|
reply = stripped[len(protocol.INPUT_REQUIRED_MARKER):].strip()
|
|
|
|
protocol.persist_message(context_id, "agent", reply, task_id)
|
|
security.audit("outbound", peer, task_id, reply)
|
|
|
|
if state in (protocol.STATE_COMPLETED, protocol.STATE_INPUT_REQUIRED):
|
|
protocol.metrics.outbound_total += 1
|
|
protocol.metrics.tasks_completed += 1
|
|
protocol.metrics.record_latency(time.time() - pending["started"])
|
|
else:
|
|
protocol.metrics.tasks_failed += 1
|
|
|
|
self.tasks.complete(task_id, state, reply)
|
|
self._send_push_notification(task_id, context_id, reply, state)
|
|
return state, reply
|
|
|
|
def _await_reply(self, pending: dict, keepalive=None) -> tuple[str, str]:
|
|
"""Block until the task's future resolves (or times out).
|
|
|
|
``keepalive`` is an optional zero-arg callable invoked every
|
|
_SSE_KEEPALIVE seconds while waiting (used by the SSE paths); if it
|
|
raises, the client is gone and we stop waiting.
|
|
"""
|
|
fut: Future = pending["future"]
|
|
deadline = pending["started"] + _reply_timeout()
|
|
while True:
|
|
try:
|
|
return fut.result(timeout=_SSE_KEEPALIVE if keepalive else max(0.0, deadline - time.time()))
|
|
except FuturesTimeout:
|
|
if time.time() >= deadline:
|
|
return (protocol.STATE_FAILED, "[agent did not reply in time]")
|
|
if keepalive:
|
|
try:
|
|
keepalive()
|
|
except Exception:
|
|
return (protocol.STATE_FAILED, "[client disconnected]")
|
|
except Exception:
|
|
return (protocol.STATE_FAILED, "[agent did not reply in time]")
|
|
|
|
def _rpc_message_send(self, req_id: Any, params: dict, peer: str, agent: Optional[dict] = None, v1_response: bool = False) -> dict:
|
|
terminal, pending = self._prepare_task(params, peer, agent=agent)
|
|
if terminal is not None:
|
|
result = protocol.send_message_response(terminal) if v1_response else terminal
|
|
return protocol.jsonrpc_result(req_id, result)
|
|
state, reply = self._await_reply(pending)
|
|
state, reply = self._finalize_task(pending, state, reply)
|
|
task = protocol.build_task(
|
|
pending["task_id"], pending["context_id"], state, reply,
|
|
created_at=pending["created_iso"],
|
|
)
|
|
result = protocol.send_message_response(task) if v1_response else task
|
|
return protocol.jsonrpc_result(req_id, result)
|
|
|
|
# ── Streaming (SSE) ───────────────────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def _sse_headers(handler) -> None:
|
|
handler.send_response(200)
|
|
handler.send_header("Content-Type", "text/event-stream")
|
|
handler.send_header("Cache-Control", "no-cache")
|
|
handler.end_headers()
|
|
# v1.0: closing the stream signals the terminal state, so the socket
|
|
# must actually close once we emit the done event.
|
|
handler.close_connection = True
|
|
|
|
@staticmethod
|
|
def _sse_write(handler, chunk: str) -> None:
|
|
handler.wfile.write(chunk.encode("utf-8"))
|
|
handler.wfile.flush()
|
|
|
|
def _emit_terminal(self, handler, task_id: str, context_id: str, state: str, reply: str,
|
|
req_id: Any = None) -> None:
|
|
"""Emit the final artifact/status events and close the stream (v1.0:
|
|
closure signals terminal state, no ``final`` field).
|
|
|
|
``req_id`` is threaded into JSON-RPC-wrapped SSE frames per §9.4."""
|
|
if reply and state == protocol.STATE_COMPLETED:
|
|
self._sse_write(handler, protocol.sse_data(
|
|
protocol.artifact_update(task_id, context_id, reply), req_id))
|
|
self._sse_write(handler, protocol.sse_data(
|
|
protocol.status_update(task_id, context_id, state), req_id))
|
|
else:
|
|
self._sse_write(handler, protocol.sse_data(
|
|
protocol.status_update(task_id, context_id, state, reply), req_id))
|
|
self._sse_write(handler, protocol.sse_done())
|
|
|
|
def _rpc_message_stream(self, handler, req_id: Any, params: dict, peer: str, agent: Optional[dict] = None) -> None:
|
|
"""Handle message/stream as an SSE response of JSON-RPC-wrapped
|
|
StreamResponse events (A2A v1.0 §9.4)."""
|
|
protocol.metrics.streams_started += 1
|
|
self._sse_headers(handler)
|
|
|
|
try:
|
|
terminal, pending = self._prepare_task(params, peer, agent=agent)
|
|
if terminal is not None:
|
|
self._emit_terminal(
|
|
handler, terminal["id"], terminal["contextId"],
|
|
terminal["status"]["state"],
|
|
protocol.extract_text(terminal.get("status", {}).get("message", {}) or {}),
|
|
req_id=req_id,
|
|
)
|
|
return
|
|
|
|
task_id, context_id = pending["task_id"], pending["context_id"]
|
|
self._sse_write(handler, protocol.sse_data(protocol.stream_task(
|
|
protocol.build_task(task_id, context_id, protocol.STATE_SUBMITTED, created_at=pending["created_iso"])),
|
|
req_id))
|
|
self._sse_write(handler, protocol.sse_data(
|
|
protocol.status_update(task_id, context_id, protocol.STATE_WORKING), req_id))
|
|
|
|
state, reply = self._await_reply(
|
|
pending, keepalive=lambda: self._sse_write(handler, ": keepalive\n\n"))
|
|
state, reply = self._finalize_task(pending, state, reply)
|
|
self._emit_terminal(handler, task_id, context_id, state, reply, req_id=req_id)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
logger.debug("A2A: stream client disconnected")
|
|
|
|
def _rpc_tasks_subscribe(self, handler, req_id: Any, params: dict, agent: Optional[dict] = None) -> None:
|
|
"""Reconnect to an existing task's stream (v1.0 SubscribeToTask)."""
|
|
task_id = str(params.get("taskId") or params.get("id") or "")
|
|
rec = self.tasks.get(task_id, *self._scope_for_agent(agent))
|
|
if not rec:
|
|
handler._json(200, protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_FOUND, f"task not found: {task_id}"))
|
|
return
|
|
|
|
self._sse_headers(handler)
|
|
try:
|
|
fut = self.tasks.watch(task_id, *self._scope_for_agent(agent))
|
|
if fut is None:
|
|
self._sse_write(handler, protocol.sse_done())
|
|
return
|
|
deadline = time.time() + _reply_timeout()
|
|
while True:
|
|
try:
|
|
state, reply = fut.result(timeout=_SSE_KEEPALIVE)
|
|
break
|
|
except FuturesTimeout:
|
|
if time.time() >= deadline:
|
|
state, reply = rec["state"], rec.get("reply", "")
|
|
break
|
|
self._sse_write(handler, ": keepalive\n\n")
|
|
self._emit_terminal(handler, task_id, rec["context_id"], state, reply, req_id=req_id)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
logger.debug("A2A: subscribe client disconnected")
|
|
|
|
# ── Task queries ──────────────────────────────────────────────────────
|
|
|
|
def _rpc_tasks_get(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
task_id = str(params.get("taskId") or params.get("id") or "")
|
|
rec = self.tasks.get(task_id, *self._scope_for_agent(agent))
|
|
if not rec:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_FOUND, f"task not found: {task_id}")
|
|
history_len = params.get("historyLength")
|
|
try:
|
|
history_len = int(history_len) if history_len is not None else None
|
|
except (TypeError, ValueError):
|
|
history_len = None
|
|
return protocol.jsonrpc_result(req_id, protocol.TaskStore.to_task(rec, history_length=history_len))
|
|
|
|
def _rpc_tasks_list(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
try:
|
|
offset = int(params.get("pageToken") or 0)
|
|
except (ValueError, TypeError):
|
|
offset = 0
|
|
try:
|
|
page_size = int(params.get("pageSize") or 50)
|
|
except (ValueError, TypeError):
|
|
page_size = 50
|
|
recs, next_offset, total = self.tasks.list(
|
|
context_id=str(params.get("contextId") or ""),
|
|
state=str(params.get("status") or params.get("state") or ""),
|
|
page_size=page_size,
|
|
offset=max(0, offset),
|
|
agent_slug=self._scope_for_agent(agent)[0],
|
|
tenant=self._scope_for_agent(agent)[1],
|
|
with_total=True,
|
|
)
|
|
include_artifacts = bool(params.get("includeArtifacts", False))
|
|
history_len = params.get("historyLength")
|
|
try:
|
|
history_len = int(history_len) if history_len is not None else None
|
|
except (TypeError, ValueError):
|
|
history_len = None
|
|
return protocol.jsonrpc_result(req_id, {
|
|
"tasks": [protocol.TaskStore.to_task(r, history_length=history_len, include_artifacts=include_artifacts) for r in recs],
|
|
"nextPageToken": str(next_offset) if next_offset else "",
|
|
"pageSize": max(1, min(page_size, 100)),
|
|
"totalSize": total,
|
|
})
|
|
|
|
def _rpc_tasks_cancel(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
task_id = str(params.get("taskId") or params.get("id") or "")
|
|
rec = self.tasks.get(task_id, *self._scope_for_agent(agent))
|
|
if not rec:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_FOUND, f"task not found: {task_id}")
|
|
if rec["state"] in protocol.TERMINAL_STATES:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_CANCELABLE,
|
|
f"task {task_id} already {rec['state']}")
|
|
self.tasks.complete(task_id, protocol.STATE_CANCELED, "")
|
|
self._turns.reset(rec["context_id"])
|
|
self._resolve_task(task_id, protocol.STATE_CANCELED, "")
|
|
rec = self.tasks.get(task_id, *self._scope_for_agent(agent)) or rec
|
|
return protocol.jsonrpc_result(req_id, protocol.TaskStore.to_task(rec))
|
|
|
|
# ── Push notifications ────────────────────────────────────────────────
|
|
|
|
def _register_inline_push(self, task_id: str, params: dict, agent: Optional[dict] = None) -> None:
|
|
"""v1.0: message/send can carry configuration.taskPushNotificationConfig."""
|
|
cfg = (params.get("configuration") or {}).get("taskPushNotificationConfig") or {}
|
|
if not isinstance(cfg, dict):
|
|
return
|
|
url = cfg.get("url") or (cfg.get("pushNotificationConfig") or {}).get("url") or ""
|
|
if url:
|
|
self.tasks.set_push_config(task_id, str(url), *self._scope_for_agent(agent))
|
|
|
|
def _rpc_push_config_create(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
task_id = str(params.get("taskId") or "")
|
|
cfg = params.get("pushNotificationConfig") or params.get("config") or {}
|
|
url = str((cfg or {}).get("url") or "")
|
|
if not task_id or not url:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_INVALID_PARAMS,
|
|
"taskId and pushNotificationConfig.url required")
|
|
stored = self.tasks.set_push_config(task_id, url, *self._scope_for_agent(agent))
|
|
if stored is None:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_FOUND, f"task not found: {task_id}")
|
|
return protocol.jsonrpc_result(req_id, stored)
|
|
|
|
def _rpc_push_config_get(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
"""GetTaskPushNotificationConfig — retrieve a push config by task id."""
|
|
task_id = str(params.get("taskId") or "")
|
|
config_id = str(params.get("id") or params.get("configId") or "")
|
|
if not task_id:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_INVALID_PARAMS, "taskId required")
|
|
cfg = self.tasks.get_push_config(task_id, config_id, *self._scope_for_agent(agent))
|
|
if cfg is None:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_FOUND,
|
|
f"push config not found for task: {task_id}")
|
|
return protocol.jsonrpc_result(req_id, cfg)
|
|
|
|
def _rpc_push_config_list(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
"""ListTaskPushNotificationConfigs — list push configs for a task."""
|
|
task_id = str(params.get("taskId") or "")
|
|
if not task_id:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_INVALID_PARAMS, "taskId required")
|
|
configs = self.tasks.list_push_configs(task_id, *self._scope_for_agent(agent))
|
|
return protocol.jsonrpc_result(req_id, {"configs": configs, "nextPageToken": ""})
|
|
|
|
def _rpc_push_config_delete(self, req_id: Any, params: dict, agent: Optional[dict] = None) -> dict:
|
|
"""DeleteTaskPushNotificationConfig — remove a push config."""
|
|
task_id = str(params.get("taskId") or "")
|
|
config_id = str(params.get("id") or params.get("configId") or "")
|
|
if not task_id:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_INVALID_PARAMS, "taskId required")
|
|
deleted = self.tasks.delete_push_config(task_id, config_id, *self._scope_for_agent(agent))
|
|
if not deleted:
|
|
return protocol.jsonrpc_error(
|
|
req_id, protocol.ERR_TASK_NOT_FOUND,
|
|
f"push config not found for task: {task_id}")
|
|
return protocol.jsonrpc_result(req_id, {"deleted": True})
|
|
|
|
def _send_push_notification(self, task_id: str, context_id: str, reply: str, state: str) -> None:
|
|
"""POST a v1.0 StreamResponse payload to the task's registered callback.
|
|
|
|
Validates the callback URL to prevent SSRF — blocks internal/private
|
|
addresses (169.254.x.x metadata, loopback, RFC1918 private ranges)
|
|
unless we're in localhost-only mode (where internal access is expected).
|
|
"""
|
|
callback_url = self.tasks.pop_push_url(task_id)
|
|
if not callback_url:
|
|
return
|
|
|
|
if not security.is_safe_callback_url(
|
|
callback_url,
|
|
localhost_mode=self._security_context.localhost_only(),
|
|
):
|
|
logger.warning("A2A: push notification for task %s blocked — unsafe callback URL: %s",
|
|
task_id, callback_url)
|
|
protocol.metrics.push_failed += 1
|
|
return
|
|
|
|
# Push payload uses the StreamResponse format (same as streaming).
|
|
payload = protocol.status_update(task_id, context_id, state, (reply or "")[:2000])
|
|
|
|
signature = self._security_context.sign_push_payload(payload)
|
|
headers = {"Content-Type": "application/json"}
|
|
if signature:
|
|
headers["X-A2A-Signature"] = signature
|
|
|
|
try:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(callback_url, data=data, headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, timeout=10) as resp: # noqa: S310
|
|
if 200 <= resp.status < 300:
|
|
protocol.metrics.push_sent += 1
|
|
logger.debug("A2A: push notification sent for task %s", task_id)
|
|
else:
|
|
protocol.metrics.push_failed += 1
|
|
logger.warning("A2A: push notification for task %s got HTTP %d", task_id, resp.status)
|
|
except Exception as e:
|
|
protocol.metrics.push_failed += 1
|
|
logger.warning("A2A: push notification for task %s failed: %s", task_id, e)
|
|
|
|
# ── Sending (the agent's reply path) ──────────────────────────────────
|
|
|
|
async def send(
|
|
self,
|
|
chat_id: str,
|
|
content: str,
|
|
reply_to: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Fulfil the pending reply Future for this context.
|
|
|
|
``chat_id`` is the A2A context id we set as the source chat_id; the
|
|
oldest outstanding task for that context receives the reply (the
|
|
gateway session processes messages in order).
|
|
|
|
The gateway marks final user-visible replies with ``metadata['notify']``
|
|
(see ``_mark_notify_metadata`` in gateway.platforms.base — this is the
|
|
base adapter's documented reply marker, not an incidental field).
|
|
Progress, status, and editable preview sends intentionally lack the
|
|
marker; those must not satisfy the JSON-RPC caller.
|
|
"""
|
|
message_id = str(int(time.time() * 1000))
|
|
if not (metadata or {}).get("notify"):
|
|
logger.debug("A2A: ignoring non-final send for context %s", chat_id)
|
|
return SendResult(success=True, message_id=message_id)
|
|
if not self._resolve_oldest_for_context(chat_id, protocol.STATE_COMPLETED, content or ""):
|
|
# No waiter (e.g. a late chunk or out-of-band send) — drop it.
|
|
logger.debug("A2A: send() for context %s had no pending waiter", chat_id)
|
|
return SendResult(success=True, message_id=message_id)
|
|
|
|
async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None:
|
|
"""Resolve the task future when processing ends without a reply send.
|
|
|
|
The success path resolves via send(); this hook catches failures,
|
|
cancellations, and empty runs so the HTTP thread returns promptly
|
|
instead of waiting out the reply timeout.
|
|
"""
|
|
task_id = str(getattr(event, "message_id", "") or "")
|
|
if not task_id:
|
|
return
|
|
if outcome == ProcessingOutcome.FAILURE:
|
|
self._resolve_task(task_id, protocol.STATE_FAILED, "[agent processing failed]")
|
|
elif outcome == ProcessingOutcome.CANCELLED:
|
|
self._resolve_task(task_id, protocol.STATE_CANCELED, "")
|
|
else:
|
|
self._resolve_task(task_id, protocol.STATE_COMPLETED, "")
|
|
|
|
async def send_typing(self, chat_id: str, metadata=None) -> None:
|
|
return None
|
|
|
|
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
|
|
return {"name": f"a2a:{chat_id}", "type": "dm"}
|