Add agent registry: persona + prompt + memory + tool scope + trust (kb#134, A2A-2)
Per DESIGN-a2a-agents.md v2.1 §2, §5, §5b: an agent is a persona, not a queue. Sibling of kb#133's model-registry.yaml, same load/get/CLI shape. - agent-registry.yaml: populates adolf (trusted, scoped core tools), claude-coder (trusted, implementer), torgash (sandboxed, marketplace- only, target state - not yet built), researcher (sandboxed/untrusted- input, target state), kimi-endpoint/local-small-endpoint (trivial model-agents from #133), and alvis/elizaveta as first-class human agents with trust=human, own inboxes, own private memory bank targets (adolf-alvis/adolf-elizaveta/adolf-shared - split itself is kb#153). Also carries trust_classes with numeric rank, a runtimes: section for Claude Code (kb#133 explicitly excluded it from model-registry.yaml), and memory_bank_policy documenting §5b's hard rules. - agent_registry.py: effective_card() is what makes "backbone swap is one field" true rather than aspirational - tier/cost_class/lifecycle are never stored statically on an agent, they're resolved at read time from `backbone` via model-registry.yaml or runtimes:. Also trust_rank() and can_reach_vault() (kb#147's vault-access DECIDED rule as a checkable function, not just a comment). Several Cards (torgash, researcher, elizaveta's KB identity, the split memory banks) describe target state for things not yet built/migrated - flagged with status/note fields rather than implied live, matching the task's scope (registry only, no runtime migration). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014t8Qg9gi7H7HtT8MncoXAB
This commit is contained in:
207
openai/agent_registry.py
Executable file
207
openai/agent_registry.py
Executable file
@@ -0,0 +1,207 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user