← back to A2a Lab
A2A cabinet directory server (list/find/card) + fix block-scalar domain parsing
d13a06a084db54e5e7e6c9a2af5f68a719ffb173 · 2026-08-01 21:59:49 -0700 · Steve
cabinet_directory.py serves+routes the 12 officer cards over A2A (verified over the wire).
cabinet_cards.py now handles YAML folded/literal domain scalars (vp-cncp/vp-consulting).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Files touched
M README.mdM cabinet_cards.pyA cabinet_directory.pyM cards/vp-cncp.agent-card.jsonM cards/vp-consulting.agent-card.json
Diff
commit d13a06a084db54e5e7e6c9a2af5f68a719ffb173
Author: Steve <steve@designerwallcoverings.com>
Date: Sat Aug 1 21:59:49 2026 -0700
A2A cabinet directory server (list/find/card) + fix block-scalar domain parsing
cabinet_directory.py serves+routes the 12 officer cards over A2A (verified over the wire).
cabinet_cards.py now handles YAML folded/literal domain scalars (vp-cncp/vp-consulting).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
README.md | 16 ++++
cabinet_cards.py | 23 ++++-
cabinet_directory.py | 179 ++++++++++++++++++++++++++++++++++++
cards/vp-cncp.agent-card.json | 2 +-
cards/vp-consulting.agent-card.json | 2 +-
5 files changed, 217 insertions(+), 5 deletions(-)
diff --git a/README.md b/README.md
index 6f53a57..0ed7819 100644
--- a/README.md
+++ b/README.md
@@ -120,6 +120,22 @@ Note: `cabinet.yaml` is **not strictly-valid YAML** (its `owns: [...]` flow list
unquoted prose with `:` and `/`, which PyYAML rejects), so the generator uses a tolerant
line parser. That invalidity is a latent bug for anything that YAML-loads the file.
+## Cabinet Directory — serve + route the cards over A2A (verified)
+
+`cabinet_directory.py` loads the generated `cards/` and exposes the whole cabinet as one
+discoverable A2A agent, so the org chart is **queryable and routable over the protocol**:
+
+```bash
+python cabinet_directory.py # 127.0.0.1:41243, 12 officers loaded
+A2A_BASE=http://127.0.0.1:41243 python client.py "list"
+A2A_BASE=http://127.0.0.1:41243 python client.py "find scrape a vendor catalog to shopify"
+A2A_BASE=http://127.0.0.1:41243 python client.py "card vp-security"
+```
+
+Skills: `list` (all officers + skill counts), `find <task>` (rank officers by trigger/skill
+match — e.g. "rotate a leaked api key" → **vp-security**), `card <vp>` (full card). Verified
+end-to-end over JSON-RPC. This is the discovery/routing layer the cards were built for.
+
## Next steps (not done — future work)
- Add an A2A *client-side* helper using the SDK's own `create_client` (proto
diff --git a/cabinet_cards.py b/cabinet_cards.py
index 2c66083..a759275 100644
--- a/cabinet_cards.py
+++ b/cabinet_cards.py
@@ -47,17 +47,32 @@ def parse_cabinet(text: str) -> list[dict]:
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*(.+?)\s*$", raw)
+ m_dom = re.match(r"^\s{4}domain:\s*(.*)$", raw)
if m_dom:
- cur["domain"] = m_dom.group(1).strip()
- section = None
+ 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
@@ -93,6 +108,8 @@ def parse_cabinet(text: str) -> list[dict]:
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
diff --git a/cabinet_directory.py b/cabinet_directory.py
new file mode 100644
index 0000000..2cf44c8
--- /dev/null
+++ b/cabinet_directory.py
@@ -0,0 +1,179 @@
+"""A2A Cabinet Directory — serve + route Steve's 12 cabinet officer cards over A2A.
+
+Loads the generated cards/*.agent-card.json (from cabinet_cards.py) and exposes the
+cabinet as a single discoverable A2A agent. Makes the org chart queryable/routable over
+the protocol instead of just described.
+
+Skills (drive via message text):
+ "list" -> every officer + skill count
+ "card <vp>" -> one officer's description + skills
+ "find <task>" -> rank officers by relevance to a task (trigger/skill/desc match)
+
+Run: . .venv/bin/activate && python cabinet_directory.py (listens on 127.0.0.1:41243)
+"""
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from pathlib import Path
+
+import uvicorn
+from starlette.applications import Starlette
+
+from a2a.types import (
+ AgentCard,
+ AgentInterface,
+ AgentCapabilities,
+ AgentSkill,
+ Message,
+ Part,
+ Role,
+)
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore
+from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
+
+HOST = "127.0.0.1"
+PORT = 41243
+CARDS_DIR = Path(__file__).parent / "cards"
+_STOP = {"the", "a", "an", "and", "or", "to", "for", "of", "with", "who", "what",
+ "handles", "handle", "which", "officer", "do", "i", "need", "can", "help", "me"}
+
+
+def _load_officers() -> list[dict]:
+ """Each officer: {vp, description, skills:[{name,description}], blob (searchable)}."""
+ out = []
+ for f in sorted(CARDS_DIR.glob("*.agent-card.json")):
+ c = json.loads(f.read_text(encoding="utf-8"))
+ skills = c.get("skills", []) or []
+ blob = " ".join(
+ [c.get("name", ""), c.get("description", "")]
+ + [f"{s.get('name','')} {s.get('description','')}" for s in skills]
+ ).lower()
+ out.append(
+ {
+ "vp": c.get("name", f.stem),
+ "description": c.get("description", ""),
+ "skills": [{"name": s.get("name", ""), "description": s.get("description", "")} for s in skills],
+ "blob": blob,
+ }
+ )
+ return out
+
+
+OFFICERS = _load_officers()
+
+
+def _tokens(s: str) -> list[str]:
+ return [w for w in re.findall(r"[a-z0-9]+", (s or "").lower()) if w not in _STOP and len(w) > 1]
+
+
+def find(query: str, top: int = 3) -> list[tuple[dict, int, list[str]]]:
+ q = _tokens(query)
+ scored = []
+ for o in OFFICERS:
+ hits = [w for w in q if w in o["blob"]]
+ if hits:
+ scored.append((o, len(hits), sorted(set(hits))))
+ scored.sort(key=lambda x: x[1], reverse=True)
+ return scored[:top]
+
+
+def handle(text: str) -> str:
+ cmd = (text or "").strip()
+ low = cmd.lower()
+ if low in ("", "help"):
+ return "cabinet-directory: 'list' | 'card <vp>' | 'find <task/keywords>'"
+ if low == "list":
+ lines = [f"{len(OFFICERS)} cabinet officers:"]
+ for o in OFFICERS:
+ lines.append(f" • {o['vp']:22} {len(o['skills']):2} skills — {o['description'][:70]}")
+ return "\n".join(lines)
+ if low.startswith("card"):
+ parts = cmd.split(maxsplit=1)
+ if len(parts) < 2:
+ return "usage: card <vp>"
+ name = parts[1].strip().lower()
+ match = next((o for o in OFFICERS if o["vp"].lower() == name), None) \
+ or next((o for o in OFFICERS if name in o["vp"].lower()), None)
+ if not match:
+ return f"no officer matches {parts[1]!r}. Try 'list'."
+ lines = [f"{match['vp']} — {match['description']}", f" skills ({len(match['skills'])}):"]
+ for s in match["skills"]:
+ lines.append(f" · {s['name']}: {s['description'][:90]}")
+ return "\n".join(lines)
+ if low.startswith("find"):
+ q = cmd[4:].strip()
+ if not q:
+ return "usage: find <task or keywords>"
+ ranked = find(q)
+ if not ranked:
+ return f"no officer matched {q!r}. Try 'list' to see all."
+ lines = [f"best officer(s) for {q!r}:"]
+ for o, score, hits in ranked:
+ lines.append(f" → {o['vp']} (score {score}; matched: {', '.join(hits)})")
+ return "\n".join(lines)
+ # bare query with no verb → treat as find
+ ranked = find(cmd)
+ if ranked:
+ lines = [f"(interpreted as find {cmd!r})"]
+ for o, score, hits in ranked:
+ lines.append(f" → {o['vp']} (score {score}; matched: {', '.join(hits)})")
+ return "\n".join(lines)
+ return f"unknown command {cmd!r} — try 'help'"
+
+
+class CabinetDirectoryExecutor(AgentExecutor):
+ async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
+ try:
+ user_text = context.get_user_input() or ""
+ except Exception:
+ user_text = ""
+ reply = Message(
+ message_id=str(uuid.uuid4()),
+ role=Role.ROLE_AGENT,
+ parts=[Part(text=handle(user_text))],
+ )
+ await event_queue.enqueue_event(reply)
+
+ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+ raise NotImplementedError("directory is read-only.")
+
+
+def build_agent_card() -> AgentCard:
+ return AgentCard(
+ name="Cabinet Directory",
+ description=f"A2A directory over Steve's {len(OFFICERS)}-officer cabinet. Discover and "
+ "route tasks to the right VP officer by trigger/skill match.",
+ version="0.1.0",
+ supported_interfaces=[
+ AgentInterface(url=f"http://{HOST}:{PORT}/", 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=[
+ AgentSkill(id="list", name="List officers", description="List every cabinet officer + skill count.",
+ tags=["cabinet", "directory"], examples=["list"]),
+ AgentSkill(id="find", name="Route a task", description="Rank officers by relevance to a task.",
+ tags=["cabinet", "routing"], examples=["find who handles DNS and SSL", "find scrape a vendor catalog"]),
+ AgentSkill(id="card", name="Get officer card", description="Return one officer's full card + skills.",
+ tags=["cabinet", "directory"], examples=["card vp-engineering"]),
+ ],
+ )
+
+
+def build_app() -> Starlette:
+ card = build_agent_card()
+ handler = DefaultRequestHandler(
+ agent_executor=CabinetDirectoryExecutor(), task_store=InMemoryTaskStore(), agent_card=card
+ )
+ return Starlette(routes=[*create_agent_card_routes(card), *create_jsonrpc_routes(handler, "/")])
+
+
+if __name__ == "__main__":
+ print(f"A2A Cabinet Directory on http://{HOST}:{PORT} ({len(OFFICERS)} officers loaded)")
+ uvicorn.run(build_app(), host=HOST, port=PORT, log_level="warning")
diff --git a/cards/vp-cncp.agent-card.json b/cards/vp-cncp.agent-card.json
index c97ebf2..8a6ed3a 100644
--- a/cards/vp-cncp.agent-card.json
+++ b/cards/vp-cncp.agent-card.json
@@ -1,6 +1,6 @@
{
"name": "vp-cncp",
- "description": ">- (triggers: cncp officer | drive cncp | run the operation | status of everything \u00b7 keep projects moving | keep everything flowing | nothing should stall \u00b7 clear the approvals | clear the queue | clear tasks | advance the tasks \u00b7 unstick the pipeline | who's stalled | what's blocked | who's idle)",
+ "description": "CROSS-CUTTING flow officer (chief-of-staff). Standing operations owner whose one job is to keep the WHOLE operation moving so nothing stalls \u2014 drives the CNCP command center (http://localhost:3333, repo ~/cncp-starter, cncp-*.json) and ORCHESTRATES every other cabinet officer. Clears the approvals queue, advances stalled tasks + the 1700+ parking-lot items, un-sticks quiet projects, and logs wins as work lands. PROPOSE + ORCHESTRATE ONLY \u2014 routes reversible work to the owning officer, surfaces o",
"supportedInterfaces": [
{
"url": "http://127.0.0.1:41300/vp-cncp/",
diff --git a/cards/vp-consulting.agent-card.json b/cards/vp-consulting.agent-card.json
index 8f7fe8a..46c4e4f 100644
--- a/cards/vp-consulting.agent-card.json
+++ b/cards/vp-consulting.agent-card.json
@@ -1,6 +1,6 @@
{
"name": "vp-consulting",
- "description": ">- (triggers: consulting | officer consulting | consulting client | client portal \u00b7 onboard a consulting client | build a website and social for | intake questionnaire \u00b7 new consulting client | spin up a client portal | concept versions | growth command center)",
+ "description": "Officer Consulting \u2014 the Consulting business: a productized \"client-growth consultancy in a box\" that turns an intake questionnaire into a per-client portal (a NEW website shown as several concept versions + a deeply integrated-social growth command center), even for clients with no website. Fuses two proven builds \u2014 fantasea-consulting (deliverable/concept-version generator) + prestige-car-wash (social/growth command center). Owns the `consulting` skill and every ~/Projects/consulting-<slug>/ e",
"supportedInterfaces": [
{
"url": "http://127.0.0.1:41300/vp-consulting/",
← 9ea5e3d generate A2A agent cards from cabinet.yaml (12 VPs / 166 ski
·
back to A2a Lab
·
add SDK-native client (client_sdk.py) + lexical-semantic fin 714e492 →