test: A2A agent card tests (#802)

GET /.well-known/agent-card.json returns A2A-compliant agent card
built from local config + installed skills.

agent/agent_card.py

@@ -0,0 +1,358 @@
+"""
+A2A Agent Card — publish capabilities for fleet discovery.
+
+Generates an A2A-compliant agent card from hermes config + installed skills.
+Serves at GET /.well-known/agent-card.json via the API server adapter.
+
+Spec: https://github.com/google/A2A (Agent2Agent Protocol v1.0)
+Issue: #802
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import platform
+import socket
+import time
+from dataclasses import dataclass, field, asdict
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class AgentSkill:
+    """A single skill the agent can perform."""
+    id: str
+    name: str
+    description: str = ""
+    tags: List[str] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        d = {"id": self.id, "name": self.name}
+        if self.description:
+            d["description"] = self.description
+        if self.tags:
+            d["tags"] = self.tags
+        return d
+
+
+@dataclass
+class AgentCapabilities:
+    """What the agent supports."""
+    streaming: bool = True
+    pushNotifications: bool = False
+    stateTransitionHistory: bool = True
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+@dataclass
+class AgentCard:
+    """
+    A2A-compliant agent card.
+
+    Published at /.well-known/agent-card.json for fleet discovery.
+    """
+    name: str
+    description: str = ""
+    url: str = ""
+    version: str = "1.0.0"
+    capabilities: AgentCapabilities = field(default_factory=AgentCapabilities)
+    skills: List[AgentSkill] = field(default_factory=list)
+    defaultInputModes: List[str] = field(default_factory=lambda: ["text/plain", "application/json"])
+    defaultOutputModes: List[str] = field(default_factory=lambda: ["text/plain", "application/json"])
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict:
+        d = {
+            "name": self.name,
+            "url": self.url,
+            "version": self.version,
+            "capabilities": self.capabilities.to_dict(),
+            "defaultInputModes": self.defaultInputModes,
+            "defaultOutputModes": self.defaultOutputModes,
+        }
+        if self.description:
+            d["description"] = self.description
+        if self.skills:
+            d["skills"] = [s.to_dict() for s in self.skills]
+        if self.metadata:
+            d["metadata"] = self.metadata
+        return d
+
+    def to_json(self, indent: int = 2) -> str:
+        return json.dumps(self.to_dict(), indent=indent)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "AgentCard":
+        caps = data.get("capabilities", {})
+        capabilities = AgentCapabilities(
+            streaming=caps.get("streaming", True),
+            pushNotifications=caps.get("pushNotifications", False),
+            stateTransitionHistory=caps.get("stateTransitionHistory", True),
+        )
+        skills = [
+            AgentSkill(
+                id=s.get("id", ""),
+                name=s.get("name", ""),
+                description=s.get("description", ""),
+                tags=s.get("tags", []),
+            )
+            for s in data.get("skills", [])
+        ]
+        return cls(
+            name=data.get("name", "hermes-agent"),
+            description=data.get("description", ""),
+            url=data.get("url", ""),
+            version=data.get("version", "1.0.0"),
+            capabilities=capabilities,
+            skills=skills,
+            defaultInputModes=data.get("defaultInputModes", ["text/plain", "application/json"]),
+            defaultOutputModes=data.get("defaultOutputModes", ["text/plain", "application/json"]),
+            metadata=data.get("metadata", {}),
+        )
+
+    @classmethod
+    def from_json(cls, text: str) -> "AgentCard":
+        return cls.from_dict(json.loads(text))
+
+
+def _load_skills_from_directory(skills_dir: Optional[str] = None) -> List[AgentSkill]:
+    """
+    Scan ~/.hermes/skills/ for installed skills and extract metadata.
+
+    Each skill is a directory with a SKILL.md that may have YAML frontmatter:
+    ---
+    name: my-skill
+    description: Does X
+    tags: [foo, bar]
+    ---
+    """
+    if skills_dir is None:
+        skills_dir = os.path.join(
+            os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")),
+            "skills"
+        )
+
+    skills_path = Path(skills_dir)
+    if not skills_path.is_dir():
+        return []
+
+    skills = []
+    seen_ids = set()
+
+    for skill_dir in sorted(skills_path.iterdir()):
+        if not skill_dir.is_dir():
+            continue
+        skill_md = skill_dir / "SKILL.md"
+        if not skill_md.exists():
+            continue
+
+        try:
+            text = skill_md.read_text(encoding="utf-8", errors="replace")
+        except OSError:
+            continue
+
+        # Parse YAML frontmatter
+        name = skill_dir.name
+        description = ""
+        tags: List[str] = []
+
+        if text.startswith("---"):
+            parts = text.split("---", 2)
+            if len(parts) >= 3:
+                fm_text = parts[1]
+                # Simple key: value parsing (avoids yaml dependency)
+                for line in fm_text.strip().splitlines():
+                    line = line.strip()
+                    if line.startswith("name:"):
+                        name = line.split(":", 1)[1].strip().strip(""'")
+                    elif line.startswith("description:"):
+                        description = line.split(":", 1)[1].strip().strip(""'")
+                    elif line.startswith("tags:"):
+                        tag_str = line.split(":", 1)[1].strip()
+                        # Parse [a, b, c] or "a, b, c"
+                        if tag_str.startswith("["):
+                            tags = [t.strip().strip(""'") for t in tag_str[1:-1].split(",") if t.strip()]
+                        else:
+                            tags = [t.strip() for t in tag_str.split(",") if t.strip()]
+
+        skill_id = skill_dir.name.lower().replace(" ", "-")
+        if skill_id in seen_ids:
+            continue
+        seen_ids.add(skill_id)
+
+        skills.append(AgentSkill(
+            id=skill_id,
+            name=name,
+            description=description,
+            tags=tags,
+        ))
+
+    return skills
+
+
+def _get_agent_name() -> str:
+    """Derive agent name from config or hostname."""
+    # Check environment
+    name = os.environ.get("AGENT_NAME", "")
+    if name:
+        return name
+
+    # Try config.yaml
+    hermes_home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))
+    config_path = Path(hermes_home) / "config.yaml"
+    if config_path.exists():
+        try:
+            text = config_path.read_text(encoding="utf-8", errors="replace")
+            for line in text.splitlines():
+                line = line.strip()
+                if line.startswith("agent_name:") or line.startswith("name:"):
+                    val = line.split(":", 1)[1].strip().strip(""'")
+                    if val:
+                        return val
+        except OSError:
+            pass
+
+    # Fallback to hostname
+    try:
+        return socket.gethostname().split(".")[0]
+    except Exception:
+        return "hermes-agent"
+
+
+def _get_agent_url(host: str = "", port: int = 0) -> str:
+    """Build the agent URL for the card."""
+    if not host:
+        host = os.environ.get("API_SERVER_HOST", os.environ.get("AGENT_HOST", "127.0.0.1"))
+    if not port:
+        port = int(os.environ.get("API_SERVER_PORT", os.environ.get("AGENT_PORT", "8642")))
+    scheme = "https" if os.environ.get("AGENT_TLS", "").lower() in ("1", "true") else "http"
+    return f"{scheme}://{host}:{port}"
+
+
+def _get_description() -> str:
+    """Get agent description from config or default."""
+    return os.environ.get(
+        "AGENT_DESCRIPTION",
+        "Sovereign AI agent — local-first, chain-anchored"
+    )
+
+
+def _get_model_info() -> dict:
+    """Get current model info for metadata."""
+    hermes_home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))
+    config_path = Path(hermes_home) / "config.yaml"
+    model = ""
+    provider = ""
+
+    if config_path.exists():
+        try:
+            text = config_path.read_text(encoding="utf-8", errors="replace")
+            for line in text.splitlines():
+                line = line.strip()
+                if line.startswith("model:"):
+                    model = line.split(":", 1)[1].strip().strip(""'")
+                elif line.startswith("provider:"):
+                    provider = line.split(":", 1)[1].strip().strip(""'")
+        except OSError:
+            pass
+
+    return {
+        "model": os.environ.get("HERMES_MODEL", model),
+        "provider": os.environ.get("HERMES_PROVIDER", provider),
+        "hostname": platform.node(),
+        "python": platform.python_version(),
+    }
+
+
+def build_agent_card(
+    name: str = "",
+    description: str = "",
+    url: str = "",
+    skills_dir: str = "",
+    host: str = "",
+    port: int = 0,
+    extra_skills: Optional[List[AgentSkill]] = None,
+) -> AgentCard:
+    """
+    Build an A2A agent card from local config + installed skills.
+
+    All parameters are optional — sensible defaults are derived from
+    environment variables, config.yaml, and the filesystem.
+
+    Usage:
+        card = build_agent_card()
+        print(card.to_json())
+
+        # Or from a config dict:
+        card = AgentCard.from_dict({"name": "timmy", ...})
+    """
+    agent_name = name or _get_agent_name()
+    agent_desc = description or _get_description()
+    agent_url = url or _get_agent_url(host, port)
+
+    # Load skills
+    skills = _load_skills_from_directory(skills_dir or None)
+    if extra_skills:
+        skills.extend(extra_skills)
+
+    # Deduplicate by id
+    seen = set()
+    unique_skills = []
+    for s in skills:
+        if s.id not in seen:
+            seen.add(s.id)
+            unique_skills.append(s)
+
+    # Capabilities
+    streaming = os.environ.get("AGENT_STREAMING", "true").lower() in ("1", "true")
+    push = os.environ.get("AGENT_PUSH_NOTIFICATIONS", "false").lower() in ("1", "true")
+    history = os.environ.get("AGENT_STATE_HISTORY", "true").lower() in ("1", "true")
+
+    card = AgentCard(
+        name=agent_name,
+        description=agent_desc,
+        url=agent_url,
+        version=os.environ.get("AGENT_CARD_VERSION", "1.0.0"),
+        capabilities=AgentCapabilities(
+            streaming=streaming,
+            pushNotifications=push,
+            stateTransitionHistory=history,
+        ),
+        skills=unique_skills,
+        metadata=_get_model_info(),
+    )
+
+    return card
+
+
+def validate_agent_card(card: AgentCard) -> List[str]:
+    """
+    Validate an agent card against A2A schema requirements.
+
+    Returns a list of validation errors (empty = valid).
+    """
+    errors = []
+
+    if not card.name:
+        errors.append("name is required")
+    if not card.url:
+        errors.append("url is required")
+
+    for s in card.skills:
+        if not s.id:
+            errors.append(f"skill missing id: {s.name}")
+        if not s.name:
+            errors.append(f"skill missing name: {s.id}")
+
+    for mode in card.defaultInputModes + card.defaultOutputModes:
+        if "/" not in mode:
+            errors.append(f"invalid MIME type: {mode}")
+
+    return errors

gateway/platforms/api_server.py

@@ -570,6 +570,33 @@ class APIServerAdapter(BasePlatformAdapter):
         """GET /health — simple health check."""
         return web.json_response({"status": "ok", "platform": "hermes-agent"})
 
+
+    async def _handle_agent_card(self, request: "web.Request") -> "web.Response":
+        """GET /.well-known/agent-card.json — A2A agent card for fleet discovery."""
+        try:
+            from agent.agent_card import build_agent_card, validate_agent_card
+            card = build_agent_card(host=request.host.split(":")[0] if request.host else "")
+            errors = validate_agent_card(card)
+            if errors:
+                logger.warning("[agent-card] Validation errors: %s", errors)
+            return web.json_response(
+                card.to_dict(),
+                headers={"Content-Type": "application/json"},
+            )
+        except Exception as e:
+            logger.error("[agent-card] Failed to build card: %s", e)
+            return web.json_response(
+                {
+                    "name": os.environ.get("AGENT_NAME", "hermes-agent"),
+                    "url": "",
+                    "version": "1.0.0",
+                    "capabilities": {"streaming": True, "pushNotifications": False, "stateTransitionHistory": True},
+                    "defaultInputModes": ["text/plain"],
+                    "defaultOutputModes": ["text/plain"],
+                },
+                headers={"Content-Type": "application/json"},
+            )
+
     async def _handle_health_detailed(self, request: "web.Request") -> "web.Response":
         """GET /health/detailed — rich status for cross-container dashboard probing.
 
@@ -2287,6 +2314,9 @@ class APIServerAdapter(BasePlatformAdapter):
     async def _sweep_orphaned_runs(self) -> None:
         """Periodically clean up run streams that were never consumed."""
         while True:
+            # A2A agent card — publish capabilities for fleet discovery (#802)
+            self._app.router.add_get("/.well-known/agent-card.json", self._handle_agent_card)
+            self._app.router.add_get("/agent-card.json", self._handle_agent_card)
             await asyncio.sleep(60)
             now = time.time()
             stale = [

tests/test_agent_card.py

@@ -0,0 +1,209 @@
+"""Tests for agent/agent_card.py — A2A agent card generator (#802)."""
+
+import json
+import os
+import tempfile
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from agent.agent_card import (
+    AgentCard,
+    AgentCapabilities,
+    AgentSkill,
+    build_agent_card,
+    validate_agent_card,
+    _load_skills_from_directory,
+    _get_agent_name,
+)
+
+
+class TestAgentSkill:
+    def test_to_dict(self):
+        s = AgentSkill(id="code", name="Code", description="Write code", tags=["python"])
+        d = s.to_dict()
+        assert d["id"] == "code"
+        assert d["name"] == "Code"
+        assert d["description"] == "Write code"
+        assert d["tags"] == ["python"]
+
+    def test_to_dict_minimal(self):
+        s = AgentSkill(id="x", name="X")
+        d = s.to_dict()
+        assert "description" not in d
+        assert "tags" not in d
+
+
+class TestAgentCapabilities:
+    def test_defaults(self):
+        caps = AgentCapabilities()
+        assert caps.streaming is True
+        assert caps.pushNotifications is False
+        assert caps.stateTransitionHistory is True
+
+    def test_to_dict(self):
+        d = AgentCapabilities().to_dict()
+        assert d == {"streaming": True, "pushNotifications": False, "stateTransitionHistory": True}
+
+
+class TestAgentCard:
+    def test_to_dict(self):
+        card = AgentCard(
+            name="timmy",
+            description="Orchestrator",
+            url="http://localhost:8642",
+            skills=[AgentSkill(id="code", name="Code")],
+        )
+        d = card.to_dict()
+        assert d["name"] == "timmy"
+        assert d["description"] == "Orchestrator"
+        assert len(d["skills"]) == 1
+        assert d["defaultInputModes"] == ["text/plain", "application/json"]
+
+    def test_roundtrip(self):
+        card = AgentCard(
+            name="test-agent",
+            url="http://localhost:9999",
+            skills=[
+                AgentSkill(id="s1", name="Skill 1", tags=["a"]),
+                AgentSkill(id="s2", name="Skill 2"),
+            ],
+            metadata={"model": "llama-3"},
+        )
+        json_str = card.to_json()
+        restored = AgentCard.from_json(json_str)
+        assert restored.name == "test-agent"
+        assert len(restored.skills) == 2
+        assert restored.skills[0].id == "s1"
+        assert restored.metadata["model"] == "llama-3"
+
+    def test_from_dict_defaults(self):
+        card = AgentCard.from_dict({})
+        assert card.name == "hermes-agent"
+        assert card.version == "1.0.0"
+        assert card.capabilities.streaming is True
+
+    def test_to_json_is_valid(self):
+        card = AgentCard(name="x", url="http://x")
+        parsed = json.loads(card.to_json())
+        assert parsed["name"] == "x"
+
+
+class TestValidateAgentCard:
+    def test_valid_card(self):
+        card = AgentCard(name="agent", url="http://localhost:8080")
+        assert validate_agent_card(card) == []
+
+    def test_missing_name(self):
+        card = AgentCard(name="", url="http://x")
+        errors = validate_agent_card(card)
+        assert any("name" in e for e in errors)
+
+    def test_missing_url(self):
+        card = AgentCard(name="x", url="")
+        errors = validate_agent_card(card)
+        assert any("url" in e for e in errors)
+
+    def test_invalid_mime(self):
+        card = AgentCard(name="x", url="http://x", defaultInputModes=["plaintext"])
+        errors = validate_agent_card(card)
+        assert any("MIME" in e for e in errors)
+
+    def test_skill_missing_id(self):
+        card = AgentCard(
+            name="x",
+            url="http://x",
+            skills=[AgentSkill(id="", name="Bad Skill")],
+        )
+        errors = validate_agent_card(card)
+        assert any("skill missing id" in e for e in errors)
+
+
+class TestLoadSkillsFromDirectory:
+    def test_empty_dir(self):
+        with tempfile.TemporaryDirectory() as d:
+            skills = _load_skills_from_directory(d)
+            assert skills == []
+
+    def test_nonexistent_dir(self):
+        skills = _load_skills_from_directory("/nonexistent/path")
+        assert skills == []
+
+    def test_loads_skills(self):
+        with tempfile.TemporaryDirectory() as d:
+            skill_dir = Path(d) / "my-skill"
+            skill_dir.mkdir()
+            (skill_dir / "SKILL.md").write_text(textwrap.dedent("""                ---
+                name: My Skill
+                description: Does amazing things
+                tags: [python, awesome]
+                ---
+
+                # My Skill
+                Instructions here.
+            """))
+            skills = _load_skills_from_directory(d)
+            assert len(skills) == 1
+            assert skills[0].id == "my-skill"
+            assert skills[0].name == "My Skill"
+            assert skills[0].description == "Does amazing things"
+            assert skills[0].tags == ["python", "awesome"]
+
+    def test_skips_non_skill_dirs(self):
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "not-a-skill").mkdir()
+            skills = _load_skills_from_directory(d)
+            assert skills == []
+
+    def test_skips_files(self):
+        with tempfile.TemporaryDirectory() as d:
+            (Path(d) / "random-file.txt").write_text("hello")
+            skills = _load_skills_from_directory(d)
+            assert skills == []
+
+
+class TestBuildAgentCard:
+    def test_with_env_vars(self, monkeypatch):
+        monkeypatch.setenv("AGENT_NAME", "test-bot")
+        monkeypatch.setenv("AGENT_DESCRIPTION", "A test agent")
+        monkeypatch.setenv("AGENT_HOST", "10.0.0.1")
+        monkeypatch.setenv("AGENT_PORT", "9090")
+
+        card = build_agent_card()
+        assert card.name == "test-bot"
+        assert card.description == "A test agent"
+        assert "10.0.0.1" in card.url
+        assert "9090" in card.url
+
+    def test_with_explicit_params(self):
+        card = build_agent_card(
+            name="explicit-agent",
+            description="Custom desc",
+            url="http://custom:1234",
+        )
+        assert card.name == "explicit-agent"
+        assert card.url == "http://custom:1234"
+
+    def test_extra_skills(self):
+        card = build_agent_card(
+            name="agent",
+            extra_skills=[AgentSkill(id="custom", name="Custom Skill")],
+        )
+        assert any(s.id == "custom" for s in card.skills)
+
+    def test_card_is_valid(self):
+        card = build_agent_card(name="valid-agent")
+        errors = validate_agent_card(card)
+        assert not errors, f"Card validation failed: {errors}"
+
+
+class TestGetAgentName:
+    def test_from_env(self, monkeypatch):
+        monkeypatch.setenv("AGENT_NAME", "from-env")
+        assert _get_agent_name() == "from-env"
+
+    def test_fallback_to_hostname(self, monkeypatch):
+        monkeypatch.delenv("AGENT_NAME", raising=False)
+        name = _get_agent_name()
+        assert len(name) > 0