feat(mcp): add get_domain_summary() for low-token domain session orientation

get_state_summary() returns ~10k tokens — too expensive for routine domain
repo sessions that only need their own workstreams and decisions.

New get_domain_summary(domain_slug):
- 5 targeted API calls: topics (filter), workstreams (topic+status), decisions
  (topic+pending), progress (topic, limit 5), repos (domain, slug+SBOM only)
- Returns: topic, active workstreams, blocking decisions, 5 recent events,
  repo SBOM status — all scoped to one domain
- Estimated ~80-90% token reduction vs get_state_summary()

get_state_summary() preserved unchanged for cross-domain / custodian sessions.
Updated its docstring to note the large response and point to get_domain_summary.

Template updated: Step 1 now calls get_domain_summary("{DOMAIN}") instead of
get_state_summary() + get_next_steps(). TOOLS.md updated with usage guidance.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-01 22:05:31 +01:00
parent a3338c3a23
commit 70c8e3cd51
3 changed files with 46 additions and 3 deletions

View File

@@ -24,6 +24,8 @@ mcp = FastMCP(
instructions=(
"Custodian State Hub: tracks topics, workstreams, tasks, decisions, and progress events. "
"Start every session with get_state_summary() for orientation. "
"When working inside a single registered domain repo, prefer get_domain_summary(domain_slug) "
"— it returns the same actionable data scoped to that domain at ~10% of the token cost. "
"All writes emit a progress_event automatically."
),
)
@@ -120,10 +122,51 @@ def get_state_summary() -> str:
Returns a full snapshot: topic/workstream/task/decision totals, blocking
decisions, blocked tasks, open workstreams, and the 20 most recent events.
NOTE: This response is large (~10k tokens). When working inside a single
registered domain repo, use get_domain_summary(domain_slug) instead —
same actionable data scoped to one domain at ~10% of the token cost.
"""
return json.dumps(_get("/state/summary"), indent=2)
@mcp.tool()
def get_domain_summary(domain_slug: str) -> str:
"""Lightweight session orientation for a single domain.
Use this instead of get_state_summary() when working in a registered
domain repo — returns only what is relevant to the specified domain,
typically 80-90% fewer tokens than the full summary.
Args:
domain_slug: the domain slug, e.g. "railiance", "markitect"
Returns: topic, active workstreams, open blocking decisions for this
topic, 5 most recent progress events, and repo SBOM status for this domain.
"""
topics = _get("/topics")
topic = next((t for t in topics if t.get("domain_slug") == domain_slug), None)
if not topic:
return json.dumps({"error": f"No topic found for domain '{domain_slug}'"})
topic_id = topic["id"]
workstreams = _get("/workstreams", {"topic_id": topic_id, "status": "active"})
blocking = _get("/decisions", {"decision_type": "pending", "topic_id": topic_id})
recent = _get("/progress", {"topic_id": topic_id, "limit": 5})
repos = _get("/repos", {"domain": domain_slug})
return json.dumps({
"domain": domain_slug,
"topic_id": topic_id,
"topic_title": topic["title"],
"workstreams": workstreams,
"blocking_decisions": blocking,
"recent_progress": recent,
"repos": [{"slug": r["slug"], "last_sbom_at": r.get("last_sbom_at")} for r in repos],
}, indent=2)
@mcp.tool()
def get_topic(slug: str) -> str:
"""Return a topic (with workstreams) by slug, plus its recent progress events."""