`cabinet_cards.py`

169 lines
"""Generate A2A agent cards from Steve's cabinet.yaml.

Turns ~/Projects/agent-cabinet/cabinet.yaml (the org chart: President → VP →
Director → Skills) into one A2A `AgentCard` per VP, where each director's `owns`
line becomes an `AgentSkill`. Writes cards/<vp>.agent-card.json (proto→JSON).

NOTE: cabinet.yaml is NOT strictly-valid YAML — its `owns: [...]` flow lists carry
unquoted prose with ':' and '/', which PyYAML rejects. So this uses a tolerant,
shallow line parser for exactly the fields we need rather than a YAML load.

Run: . .venv/bin/activate && python cabinet_cards.py
"""
from __future__ import annotations

import json
import re
from pathlib import Path

from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
from google.protobuf.json_format import MessageToDict

CABINET = Path.home() / "Projects" / "agent-cabinet" / "cabinet.yaml"
OUT = Path(__file__).parent / "cards"
# Convention only — these cards describe capabilities; a live directory server would
# assign real ports. Base kept explicit so it's obvious they aren't served yet.
BASE = "http://127.0.0.1:41300"


def slug(s: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", (s or "").lower()).strip("-")[:40]


def parse_cabinet(text: str) -> list[dict]:
    """Shallow, tolerant parse: vp / domain / triggers[] / directors[{name,kind,owns}]."""
    vps: list[dict] = []
    cur: dict | None = None
    section: str | None = None  # 'triggers' | 'directors' | None
    pending_dir: dict | None = None

    for raw in text.splitlines():
        if not raw.strip() or raw.lstrip().startswith("#"):
            continue
        m_vp = re.match(r"^\s{2}-\s+vp:\s*(.+?)\s*$", raw)
        if m_vp:
            if cur:
                if pending_dir:
                    cur["directors"].append(pending_dir)
                    pending_dir = None
                if not cur["domain"] and cur.get("_domlines"):
                    cur["domain"] = " ".join(cur["_domlines"])
                vps.append(cur)
            cur = {"vp": m_vp.group(1).strip(), "domain": "", "triggers": [], "directors": []}
            section = None
            continue
        if cur is None:
            continue
        m_dom = re.match(r"^\s{4}domain:\s*(.*)$", raw)
        if m_dom:
            val = m_dom.group(1).strip()
            if val in (">", ">-", ">+", "|", "|-", "|+", ""):
                section = "domain_block"      # YAML folded/literal scalar — accumulate
                cur["_domlines"] = []
            else:
                cur["domain"] = val
                section = None
            continue
        if section == "domain_block":
            m_cont = re.match(r"^\s{6,}(\S.*)$", raw)
            if m_cont:
                cur["_domlines"].append(m_cont.group(1).strip())
                continue
            cur["domain"] = " ".join(cur["_domlines"])  # block ended
            section = None
            # fall through — process THIS line as a normal key (triggers:/directors:/…)
        if re.match(r"^\s{4}triggers:\s*$", raw):
            section = "triggers"
            continue
        if re.match(r"^\s{4}directors:\s*$", raw):
            if pending_dir:
                cur["directors"].append(pending_dir)
                pending_dir = None
            section = "directors"
            continue
        if section == "triggers":
            m = re.match(r"^\s{6}-\s+(.+?)\s*$", raw)
            if m:
                cur["triggers"].append(m.group(1).strip())
                continue
        if section == "directors":
            m_d = re.match(r"^\s{6}-\s+(skill|subagent):\s*(.+?)\s*$", raw)
            if m_d:
                if pending_dir:
                    cur["directors"].append(pending_dir)
                pending_dir = {"kind": m_d.group(1), "name": m_d.group(2).strip(), "owns": ""}
                continue
            m_o = re.match(r"^\s{8}owns:\s*\[?(.*?)\]?\s*$", raw)
            if m_o and pending_dir is not None:
                pending_dir["owns"] = m_o.group(1).strip()
                continue
        # any other key at vp-field indent ends the current section
        if re.match(r"^\s{4}\w", raw):
            if pending_dir:
                cur["directors"].append(pending_dir)
                pending_dir = None
            section = None

    if cur:
        if pending_dir:
            cur["directors"].append(pending_dir)
        if not cur["domain"] and cur.get("_domlines"):
            cur["domain"] = " ".join(cur["_domlines"])
        vps.append(cur)
    return vps


def build_card(vp: dict) -> AgentCard:
    trig = " · ".join(vp["triggers"][:4])
    desc = vp["domain"] or vp["vp"]
    if trig:
        desc = f"{desc}  (triggers: {trig})"
    skills = []
    for d in vp["directors"]:
        owns = d["owns"] or d["name"]
        skills.append(
            AgentSkill(
                id=slug(d["name"]),
                name=d["name"],
                description=owns[:300],
                tags=[d["kind"], "cabinet"],
            )
        )
    return AgentCard(
        name=vp["vp"],
        description=desc[:500],
        version="0.1.0",
        supported_interfaces=[
            AgentInterface(
                url=f"{BASE}/{vp['vp']}/",
                protocol_binding="JSONRPC",
                protocol_version="1.0",
            )
        ],
        capabilities=AgentCapabilities(streaming=False, push_notifications=False),
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        skills=skills,
    )


def main() -> int:
    vps = parse_cabinet(CABINET.read_text(encoding="utf-8"))
    OUT.mkdir(exist_ok=True)
    total_skills = 0
    for vp in vps:
        card = build_card(vp)
        total_skills += len(card.skills)
        path = OUT / f"{vp['vp']}.agent-card.json"
        path.write_text(
            json.dumps(MessageToDict(card, preserving_proto_field_name=False), indent=2),
            encoding="utf-8",
        )
        print(f"  {vp['vp']:24} {len(card.skills):2} skills → {path.name}")
    print(f"\nGenerated {len(vps)} cabinet agent cards / {total_skills} skills → {OUT}/")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())