#!/usr/bin/env python3 """agent_registry — reads agent-registry.yaml (kb#134, A2A-2). Sibling of model_registry.py (kb#133): same load/get/CLI shape, same philosophy ("registry is data, not logic"). This module's one piece of real logic is effective_card() — the mechanism that makes "switching a backbone is one field" true rather than aspirational (see the header comment in agent-registry.yaml): an agent's static fields (persona, trust class, tool scope, memory banks, preferred tier) never duplicate what the CURRENT backbone provides (tier, cost_class, availability a(t), context window). Those are resolved at read time by dereferencing `backbone` into model-registry.yaml (via model_registry.get_model) or, for runtimes model-registry.yaml deliberately excludes (kb#133: Claude Code), into this file's own `runtimes:` section. Usage (library): from agent_registry import load_registry, get_agent, effective_card, trust_rank reg = load_registry() adolf = get_agent(reg, "adolf") card = effective_card(reg, "adolf") # -> persona/trust/tools/memory + resolved tier/cost/a(t) trust_rank(reg, "torgash") < trust_rank(reg, "adolf") # sandboxed < trusted Usage (CLI, for manual verification): ./agent_registry.py list ./agent_registry.py get --id adolf ./agent_registry.py effective-card --id adolf ./agent_registry.py trust-rank --id torgash ./agent_registry.py can-reach-vault --id torgash # kb#147 sanity check """ import argparse import json import os import sys import yaml import model_registry as mr HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_REGISTRY_PATH = os.path.join(HERE, "agent-registry.yaml") class AgentRegistryError(Exception): pass def load_registry(path=None): """Load and lightly validate agent-registry.yaml.""" path = path or DEFAULT_REGISTRY_PATH with open(path) as f: reg = yaml.safe_load(f) if not reg or "agents" not in reg: raise AgentRegistryError(f"{path}: missing top-level 'agents' list") ids = [a["id"] for a in reg["agents"]] if len(ids) != len(set(ids)): raise AgentRegistryError(f"{path}: duplicate agent ids in {ids}") runtime_ids = [r["id"] for r in reg.get("runtimes", [])] if len(runtime_ids) != len(set(runtime_ids)): raise AgentRegistryError(f"{path}: duplicate runtime ids in {runtime_ids}") return reg def get_agent(registry, agent_id): for a in registry["agents"]: if a["id"] == agent_id: return a raise AgentRegistryError(f"unknown agent id: {agent_id!r}") def get_runtime(registry, runtime_id): for r in registry.get("runtimes", []): if r["id"] == runtime_id: return r raise AgentRegistryError(f"unknown runtime id: {runtime_id!r}") # --------------------------------------------------------------------------- # trust_rank — §5 "human > trusted > sandboxed > untrusted" as a comparable # integer, for kb#140 (routing) / kb#147 (grant enforcement) to use directly # instead of re-deriving an ordering from the string. # --------------------------------------------------------------------------- def trust_rank(registry, agent_id): a = get_agent(registry, agent_id) classes = registry.get("trust_classes", {}) cls = a["trust_class"] if cls not in classes: raise AgentRegistryError(f"{agent_id}: unknown trust_class {cls!r} (have: {sorted(classes)})") return classes[cls]["rank"] def can_reach_vault(registry, agent_id): """kb#147 sanity check: vault access = trusted-or-human only (DECIDED). Cross-checks the agent's declared tool_allowlist.vault_access flag against its trust rank, so a registry typo (vault_access: true on a sandboxed agent) is a raised error, not a silent enforcement gap.""" a = get_agent(registry, agent_id) classes = registry.get("trust_classes", {}) trusted_rank = classes.get("trusted", {}).get("rank") declared = ((a.get("tool_allowlist") or {}).get("vault_access")) if a.get("tool_allowlist") else False rank = trust_rank(registry, agent_id) if declared and rank < trusted_rank: raise AgentRegistryError( f"{agent_id}: tool_allowlist.vault_access=true but trust_class=" f"{a['trust_class']!r} (rank {rank}) < trusted (rank {trusted_rank}) — " "registry inconsistency, fix before this is load-bearing for kb#147" ) return bool(declared) and rank >= trusted_rank # --------------------------------------------------------------------------- # effective_card — the full A2A Agent Card: static agent fields merged with # the CURRENT backbone's resolved tier/cost_class/a(t)/context_tokens. # --------------------------------------------------------------------------- def _resolve_backbone(registry, backbone_id, model_registry=None): """backbone can point into either this file's runtimes: (flat-subscription runtimes model-registry.yaml deliberately excludes, kb#133) or into model-registry.yaml's models: (everything else). Try runtimes first — it's the smaller, local list.""" try: return "runtime", get_runtime(registry, backbone_id) except AgentRegistryError: pass model_registry = model_registry if model_registry is not None else mr.load_registry() return "model", mr.get_model(model_registry, backbone_id) def effective_card(registry, agent_id, model_registry=None): """Return the full A2A Agent Card for `agent_id`: its own persona/trust/ tools/memory fields plus tier/cost_class/context_tokens/lifecycle resolved from whatever `backbone` currently names. Humans and trivial endpoints with backbone=None get resolved_backbone=None — there's nothing to dereference.""" a = get_agent(registry, agent_id) card = dict(a) # shallow copy; don't mutate the loaded registry backbone_id = a.get("backbone") if backbone_id is None: card["resolved_backbone"] = None return card source, backbone = _resolve_backbone(registry, backbone_id, model_registry) card["resolved_backbone"] = { "source": source, # "runtime" | "model" — which file it came from "id": backbone_id, "tier": backbone["tier"], "cost_class": backbone["cost_class"], "lifecycle": backbone["lifecycle"], "context_tokens": backbone.get("context_tokens"), "metered": backbone.get("metered", False), } return card # --------------------------------------------------------------------------- # CLI — manual verification only, not part of the library contract. # --------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--registry", default=None, help="path to agent-registry.yaml (default: sibling file)") ap.add_argument("--model-registry", default=None, help="path to model-registry.yaml (default: sibling file)") sub = ap.add_subparsers(dest="cmd", required=True) p = sub.add_parser("get") p.add_argument("--id", required=True) p = sub.add_parser("effective-card") p.add_argument("--id", required=True) p = sub.add_parser("trust-rank") p.add_argument("--id", required=True) p = sub.add_parser("can-reach-vault") p.add_argument("--id", required=True) sub.add_parser("list") args = ap.parse_args() reg = load_registry(args.registry) model_reg = mr.load_registry(args.model_registry) if args.model_registry else None try: if args.cmd == "get": print(json.dumps(get_agent(reg, args.id), indent=2)) elif args.cmd == "effective-card": print(json.dumps(effective_card(reg, args.id, model_reg), indent=2)) elif args.cmd == "trust-rank": print(trust_rank(reg, args.id)) elif args.cmd == "can-reach-vault": ok = can_reach_vault(reg, args.id) print(json.dumps({"id": args.id, "can_reach_vault": ok})) sys.exit(0 if ok else 1) elif args.cmd == "list": for a in reg["agents"]: backbone = a.get("backbone") or "-" print(f"{a['id']:22} trust={a['trust_class']:9} " f"tier={str(a.get('preferred_tier')):6} backbone={backbone:16} " f"role={a['persona']['role']}") except AgentRegistryError as exc: print(f"error: {exc}", file=sys.stderr) sys.exit(2) if __name__ == "__main__": main()