fix: implement A2A agent card for fleet discovery (closes #819)

agent/agent_card.py

@@ -0,0 +1,224 @@
+"""A2A Agent Card — publish capabilities for fleet discovery.
+
+Each fleet agent publishes an A2A-compliant agent card describing its capabilities.
+Standard discovery endpoint: /.well-known/agent-card.json
+
+Issue #819: feat: A2A agent card — publish capabilities for fleet discovery
+"""
+
+import json
+import os
+import socket
+from dataclasses import dataclass, field, asdict
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class AgentSkill:
+    """A single skill the agent can perform."""
+    id: str
+    name: str
+    description: str = ""
+    tags: List[str] = field(default_factory=list)
+    examples: List[str] = field(default_factory=list)
+    input_modes: List[str] = field(default_factory=lambda: ["text/plain"])
+    output_modes: List[str] = field(default_factory=lambda: ["text/plain"])
+
+
+@dataclass
+class AgentCapabilities:
+    """What the agent can do."""
+    streaming: bool = True
+    push_notifications: bool = False
+    state_transition_history: bool = True
+
+
+@dataclass
+class AgentCard:
+    """A2A-compliant agent card."""
+    name: str
+    description: str
+    url: str
+    version: str = "1.0.0"
+    capabilities: AgentCapabilities = field(default_factory=AgentCapabilities)
+    skills: List[AgentSkill] = field(default_factory=list)
+    default_input_modes: List[str] = field(default_factory=lambda: ["text/plain", "application/json"])
+    default_output_modes: List[str] = field(default_factory=lambda: ["text/plain", "application/json"])
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to JSON-serializable dict."""
+        d = asdict(self)
+        # Rename for A2A spec compliance
+        d["defaultInputModes"] = d.pop("default_input_modes")
+        d["defaultOutputModes"] = d.pop("default_output_modes")
+        return d
+
+    def to_json(self) -> str:
+        """Serialize to JSON string."""
+        return json.dumps(self.to_dict(), indent=2)
+
+
+def _load_skills_from_directory(skills_dir: Path) -> List[AgentSkill]:
+    """Scan ~/.hermes/skills/ for SKILL.md frontmatter."""
+    skills = []
+    
+    if not skills_dir.exists():
+        return skills
+    
+    for skill_dir in skills_dir.iterdir():
+        if not skill_dir.is_dir():
+            continue
+        
+        skill_md = skill_dir / "SKILL.md"
+        if not skill_md.exists():
+            continue
+        
+        try:
+            content = skill_md.read_text(encoding="utf-8")
+            
+            # Parse YAML frontmatter
+            if content.startswith("---"):
+                parts = content.split("---", 2)
+                if len(parts) >= 3:
+                    import yaml
+                    try:
+                        metadata = yaml.safe_load(parts[1]) or {}
+                    except Exception:
+                        metadata = {}
+                    
+                    name = metadata.get("name", skill_dir.name)
+                    desc = metadata.get("description", "")
+                    tags = metadata.get("tags", [])
+                    
+                    skills.append(AgentSkill(
+                        id=skill_dir.name,
+                        name=name,
+                        description=desc[:200] if desc else "",
+                        tags=tags if isinstance(tags, list) else [],
+                    ))
+        except Exception:
+            continue
+    
+    return skills
+
+
+def validate_agent_card(card: AgentCard) -> List[str]:
+    """Validate agent card against A2A schema requirements.
+    
+    Returns list of validation errors (empty if valid).
+    """
+    errors = []
+    
+    if not card.name:
+        errors.append("name is required")
+    if not card.url:
+        errors.append("url is required")
+    
+    # Validate MIME types
+    valid_modes = {"text/plain", "application/json", "image/png", "audio/wav"}
+    for mode in card.default_input_modes:
+        if mode not in valid_modes:
+            errors.append(f"invalid input mode: {mode}")
+    for mode in card.default_output_modes:
+        if mode not in valid_modes:
+            errors.append(f"invalid output mode: {mode}")
+    
+    # Validate skills
+    for skill in card.skills:
+        if not skill.id:
+            errors.append(f"skill missing id: {skill.name}")
+    
+    return errors
+
+
+def build_agent_card(
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+    url: Optional[str] = None,
+    version: Optional[str] = None,
+    skills: Optional[List[AgentSkill]] = None,
+    extra_skills: Optional[List[AgentSkill]] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+) -> AgentCard:
+    """Build an A2A agent card from config and environment.
+    
+    Priority: explicit params > env vars > config.yaml > defaults
+    """
+    # Load config
+    config_model = ""
+    config_provider = ""
+    try:
+        from hermes_cli.config import load_config
+        cfg = load_config()
+        model_cfg = cfg.get("model", {})
+        if isinstance(model_cfg, dict):
+            config_model = model_cfg.get("default", "")
+            config_provider = model_cfg.get("provider", "")
+        elif isinstance(model_cfg, str):
+            config_model = model_cfg
+    except Exception:
+        pass
+    
+    # Resolve values with priority
+    agent_name = name or os.environ.get("HERMES_AGENT_NAME", "") or "hermes"
+    agent_desc = description or os.environ.get("HERMES_AGENT_DESCRIPTION", "") or "Sovereign AI agent"
+    agent_url = url or os.environ.get("HERMES_AGENT_URL", "") or f"http://localhost:{os.environ.get('HERMES_API_PORT', '8642')}"
+    agent_version = version or os.environ.get("HERMES_AGENT_VERSION", "") or "1.0.0"
+    
+    # Load skills
+    if skills is not None:
+        agent_skills = skills
+    else:
+        from hermes_constants import get_hermes_home
+        skills_dir = get_hermes_home() / "skills"
+        agent_skills = _load_skills_from_directory(skills_dir)
+    
+    # Add extra skills
+    if extra_skills:
+        existing_ids = {s.id for s in agent_skills}
+        for skill in extra_skills:
+            if skill.id not in existing_ids:
+                agent_skills.append(skill)
+    
+    # Build metadata
+    card_metadata = {
+        "model": config_model or os.environ.get("HERMES_MODEL", ""),
+        "provider": config_provider or os.environ.get("HERMES_PROVIDER", ""),
+        "hostname": socket.gethostname(),
+    }
+    if metadata:
+        card_metadata.update(metadata)
+    
+    # Build capabilities
+    capabilities = AgentCapabilities(
+        streaming=True,
+        push_notifications=False,
+        state_transition_history=True,
+    )
+    
+    return AgentCard(
+        name=agent_name,
+        description=agent_desc,
+        url=agent_url,
+        version=agent_version,
+        capabilities=capabilities,
+        skills=agent_skills,
+        metadata=card_metadata,
+    )
+
+
+def get_agent_card_json() -> str:
+    """Get agent card as JSON string (for HTTP endpoint)."""
+    try:
+        card = build_agent_card()
+        return card.to_json()
+    except Exception as e:
+        # Graceful fallback — return minimal card so discovery doesn't break
+        fallback = AgentCard(
+            name="hermes",
+            description="Sovereign AI agent",
+            url=f"http://localhost:{os.environ.get('HERMES_API_PORT', '8642')}",
+        )
+        return fallback.to_json()

tests/test_agent_card.py

@@ -0,0 +1,132 @@
+"""Tests for A2A agent card — Issue #819."""
+import json
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from agent.agent_card import (
+    AgentSkill, AgentCapabilities, AgentCard,
+    validate_agent_card, build_agent_card, get_agent_card_json,
+    _load_skills_from_directory
+)
+
+
+class TestAgentSkill:
+    def test_creation(self):
+        skill = AgentSkill(id="code", name="Code", tags=["python"])
+        assert skill.id == "code"
+        assert "python" in skill.tags
+
+
+class TestAgentCapabilities:
+    def test_defaults(self):
+        caps = AgentCapabilities()
+        assert caps.streaming == True
+        assert caps.push_notifications == False
+
+
+class TestAgentCard:
+    def test_to_dict(self):
+        card = AgentCard(name="timmy", description="test", url="http://localhost:8642")
+        d = card.to_dict()
+        assert d["name"] == "timmy"
+        assert "defaultInputModes" in d
+    
+    def test_to_json(self):
+        card = AgentCard(name="timmy", description="test", url="http://localhost:8642")
+        j = card.to_json()
+        parsed = json.loads(j)
+        assert parsed["name"] == "timmy"
+
+
+class TestValidation:
+    def test_valid_card(self):
+        card = AgentCard(name="timmy", description="test", url="http://localhost:8642")
+        errors = validate_agent_card(card)
+        assert len(errors) == 0
+    
+    def test_missing_name(self):
+        card = AgentCard(name="", description="test", url="http://localhost:8642")
+        errors = validate_agent_card(card)
+        assert any("name" in e for e in errors)
+    
+    def test_missing_url(self):
+        card = AgentCard(name="timmy", description="test", url="")
+        errors = validate_agent_card(card)
+        assert any("url" in e for e in errors)
+    
+    def test_invalid_input_mode(self):
+        card = AgentCard(
+            name="timmy", description="test", url="http://localhost:8642",
+            default_input_modes=["invalid/mode"]
+        )
+        errors = validate_agent_card(card)
+        assert any("invalid input mode" in e for e in errors)
+    
+    def test_skill_missing_id(self):
+        card = AgentCard(
+            name="timmy", description="test", url="http://localhost:8642",
+            skills=[AgentSkill(id="", name="test")]
+        )
+        errors = validate_agent_card(card)
+        assert any("skill missing id" in e for e in errors)
+
+
+class TestBuildAgentCard:
+    def test_builds_valid_card(self):
+        card = build_agent_card()
+        assert card.name
+        assert card.url
+        errors = validate_agent_card(card)
+        assert len(errors) == 0
+    
+    def test_explicit_params_override(self):
+        card = build_agent_card(name="custom", description="custom desc")
+        assert card.name == "custom"
+        assert card.description == "custom desc"
+    
+    def test_extra_skills(self):
+        extra = [AgentSkill(id="extra", name="Extra")]
+        card = build_agent_card(extra_skills=extra)
+        assert any(s.id == "extra" for s in card.skills)
+
+
+class TestGetAgentCardJson:
+    def test_returns_valid_json(self):
+        j = get_agent_card_json()
+        parsed = json.loads(j)
+        assert "name" in parsed
+    
+    def test_graceful_fallback(self):
+        # Even if something fails, should return valid JSON
+        j = get_agent_card_json()
+        assert j  # Non-empty
+
+
+class TestLoadSkills:
+    def test_empty_dir(self, tmp_path):
+        skills = _load_skills_from_directory(tmp_path / "nonexistent")
+        assert len(skills) == 0
+    
+    def test_parses_skill_md(self, tmp_path):
+        skill_dir = tmp_path / "test-skill"
+        skill_dir.mkdir()
+        skill_md = skill_dir / "SKILL.md"
+        skill_md.write_text("""---
+name: Test Skill
+description: A test skill
+tags:
+  - test
+  - example
+---
+Content here
+""")
+        skills = _load_skills_from_directory(tmp_path)
+        assert len(skills) == 1
+        assert skills[0].name == "Test Skill"
+        assert "test" in skills[0].tags
+
+
+if __name__ == "__main__":
+    import pytest
+    pytest.main([__file__, "-v"])