the-nexus/nexus/a2a/card.py

"""
A2A Agent Card — generation, loading, and serving.

Reads from ~/.hermes/agent_card.yaml (or a passed path) and produces
a valid A2A AgentCard that can be served at /.well-known/agent-card.json.
"""

from __future__ import annotations

import logging
import os
from pathlib import Path
from typing import Optional

import yaml

from nexus.a2a.types import (
    AgentCard,
    AgentCapabilities,
    AgentInterface,
    AgentSkill,
)

logger = logging.getLogger("nexus.a2a.card")

DEFAULT_CARD_PATH = Path.home() / ".hermes" / "agent_card.yaml"


def load_card_config(path: Path = DEFAULT_CARD_PATH) -> dict:
    """Load raw YAML config for agent card."""
    if not path.exists():
        raise FileNotFoundError(
            f"Agent card config not found at {path}. "
            f"Copy config/agent_card.example.yaml to {path} and customize it."
        )
    with open(path) as f:
        return yaml.safe_load(f)


def build_card(config: dict) -> AgentCard:
    """
    Build an AgentCard from a config dict.

    Expected YAML structure (see config/agent_card.example.yaml):

    name: "Bezalel"
    description: "CI/CD and deployment specialist"
    version: "1.0.0"
    url: "https://bezalel.example.com"
    protocol_binding: "HTTP+JSON"
    skills:
      - id: "ci-health"
        name: "CI Health Check"
        description: "Run CI pipeline health checks"
        tags: ["ci", "devops"]
      - id: "deploy"
        name: "Deploy Service"
        description: "Deploy a service to production"
        tags: ["deploy", "ops"]
    default_input_modes: ["text/plain"]
    default_output_modes: ["text/plain"]
    streaming: false
    push_notifications: false
    auth:
      scheme: "bearer"
      token_env: "A2A_AUTH_TOKEN"
    """
    name = config["name"]
    description = config["description"]
    version = config.get("version", "1.0.0")
    url = config.get("url", "http://localhost:8080")
    binding = config.get("protocol_binding", "HTTP+JSON")

    # Build skills
    skills = []
    for s in config.get("skills", []):
        skills.append(
            AgentSkill(
                id=s["id"],
                name=s.get("name", s["id"]),
                description=s.get("description", ""),
                tags=s.get("tags", []),
                examples=s.get("examples", []),
                input_modes=s.get("inputModes", config.get("default_input_modes", ["text/plain"])),
                output_modes=s.get("outputModes", config.get("default_output_modes", ["text/plain"])),
            )
        )

    # Build security schemes from auth config
    auth = config.get("auth", {})
    security_schemes = {}
    security_requirements = []

    if auth.get("scheme") == "bearer":
        security_schemes["bearerAuth"] = {
            "httpAuthSecurityScheme": {
                "scheme": "Bearer",
                "bearerFormat": auth.get("bearer_format", "token"),
            }
        }
        security_requirements = [
            {"schemes": {"bearerAuth": {"list": []}}}
        ]
    elif auth.get("scheme") == "api_key":
        key_name = auth.get("key_name", "X-API-Key")
        security_schemes["apiKeyAuth"] = {
            "apiKeySecurityScheme": {
                "location": "header",
                "name": key_name,
            }
        }
        security_requirements = [
            {"schemes": {"apiKeyAuth": {"list": []}}}
        ]

    return AgentCard(
        name=name,
        description=description,
        version=version,
        supported_interfaces=[
            AgentInterface(
                url=url,
                protocol_binding=binding,
                protocol_version="1.0",
            )
        ],
        capabilities=AgentCapabilities(
            streaming=config.get("streaming", False),
            push_notifications=config.get("push_notifications", False),
        ),
        default_input_modes=config.get("default_input_modes", ["text/plain"]),
        default_output_modes=config.get("default_output_modes", ["text/plain"]),
        skills=skills,
        security_schemes=security_schemes,
        security_requirements=security_requirements,
    )


def load_agent_card(path: Path = DEFAULT_CARD_PATH) -> AgentCard:
    """Full pipeline: load YAML → build AgentCard."""
    config = load_card_config(path)
    return build_card(config)


def get_auth_headers(config: dict) -> dict:
    """
    Build auth headers from the agent card config for outbound requests.

    Returns dict of HTTP headers to include.
    """
    auth = config.get("auth", {})
    headers = {"A2A-Version": "1.0"}

    scheme = auth.get("scheme")
    if scheme == "bearer":
        token_env = auth.get("token_env", "A2A_AUTH_TOKEN")
        token = os.environ.get(token_env, "")
        if token:
            headers["Authorization"] = f"Bearer {token}"
    elif scheme == "api_key":
        key_env = auth.get("key_env", "A2A_API_KEY")
        key_name = auth.get("key_name", "X-API-Key")
        key = os.environ.get(key_env, "")
        if key:
            headers[key_name] = key

    return headers