← back to A2a Lab
cabinet_directory.py
222 lines
"""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 a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
AgentCapabilities,
AgentCard,
AgentInterface,
AgentSkill,
Message,
Part,
Role,
)
from starlette.applications import Starlette
import semantic
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 []
name = c.get("name", "")
desc = c.get("description", "")
dom, trig = desc, ""
if "(triggers:" in desc:
dom, trig = desc.split("(triggers:", 1)
trig = trig.rstrip(") ")
skill_names = " ".join(s.get("name", "") for s in skills)
skill_descs = " ".join(s.get("description", "") for s in skills)
# FIELD-WEIGHTED: an officer's identity (name/domain) and triggers must dominate
# incidental vocabulary buried in one skill's description (repetition = higher TF).
blob = " ".join(
[name] * 3 + [dom] * 3 + [trig] * 2 + [skill_names] * 2 + [skill_descs]
).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()
_SCORER = semantic.Scorer([o["blob"] for o in 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 _matched_terms(query: str, officer: dict) -> list[str]:
"""Which query (or synonym-expanded) terms actually appear in the officer — for the 'why'."""
terms = set(_tokens(query))
for t in list(terms):
terms |= semantic.SYNONYMS.get(t, set())
return sorted(t for t in terms if t in officer["blob"])
def find(query: str, top: int = 3):
"""Lexical-semantic ranking (TF-IDF + synonyms + trigram cosine); keyword fallback."""
try:
scores = _SCORER.score(query)
ranked = sorted(zip(OFFICERS, scores), key=lambda x: x[1], reverse=True)
out = [(o, s, _matched_terms(query, o)) for o, s in ranked if s > 0.0]
if out:
return out[:top]
except Exception:
pass
# fallback: raw keyword-hit-count
q = _tokens(query)
scored = [(o, float(len([w for w in q if w in o["blob"]])), sorted({w for w in q if w in o["blob"]}))
for o in OFFICERS]
scored = [x for x in scored if x[1] > 0]
scored.sort(key=lambda x: x[1], reverse=True)
return scored[:top]
def _shortlist(query: str, ranked, note: str = "") -> str:
"""Present find() results as a ranked SHORTLIST (a routing hint, not an oracle).
Routing here is a lexical-semantic score over 12 short officer docs — good enough to
narrow to a few candidates, NOT to trust blindly as a single answer (held-out top-1
~40-75%; see eval_find.py). So we always show up to 3 and flag weak/ambiguous picks.
"""
if not ranked:
return f"no officer matched {query!r}. Try 'list' to see all officers."
s1 = ranked[0][1]
s2 = ranked[1][1] if len(ranked) > 1 else 0.0
lines = []
if note:
lines.append(note)
lines.append(f"top {len(ranked)} officer(s) for {query!r} — ranked shortlist, pick the best fit:")
for i, (o, score, hits) in enumerate(ranked, 1):
lines.append(f" {i}. {o['vp']} (score {score:.3f}; matched: {', '.join(hits) or '—'})")
if s1 < 0.15:
lines.append(" ⚠ low confidence — verify the pick or rephrase (this is a hint, not a decision).")
elif s1 - s2 < 0.05:
lines.append(" ⚠ top picks are close — could be any of the above; use judgment.")
return "\n".join(lines)
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>"
return _shortlist(q, find(q))
# bare query with no verb → treat as find
ranked = find(cmd)
if ranked:
return _shortlist(cmd, ranked, note=f"(interpreted as find {cmd!r})")
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")