feat(rulebook): MCP tools — 16 tools for rulebook/topic/rule/subscription
This commit is contained in:
@@ -5,7 +5,7 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called
|
||||
from `mcp.server.build_mcp_server`.
|
||||
"""
|
||||
from fabledassistant.mcp.tools import (
|
||||
entities, events, milestones, notes, projects, recent, search, tags, tasks,
|
||||
entities, events, milestones, notes, projects, recent, rulebooks, search, tags, tasks,
|
||||
)
|
||||
|
||||
|
||||
@@ -20,3 +20,4 @@ def register_all(mcp) -> None:
|
||||
tags.register(mcp)
|
||||
recent.register(mcp)
|
||||
entities.register(mcp)
|
||||
rulebooks.register(mcp)
|
||||
|
||||
@@ -0,0 +1,304 @@
|
||||
"""MCP tools for the Scribe Rulebook system.
|
||||
|
||||
Sixteen tools: rulebook/topic/rule CRUD + subscription management. Thin
|
||||
wrappers over services/rulebooks.py — ownership is enforced in the service.
|
||||
|
||||
Destructive ops (delete_*) require confirmed=True; otherwise return a
|
||||
preview-style warning. Mirrors the pattern in delete_event and the design
|
||||
spec.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from fabledassistant.mcp._context import current_user_id
|
||||
from fabledassistant.services import rulebooks as rulebooks_svc
|
||||
|
||||
|
||||
# ── Rulebook CRUD ───────────────────────────────────────────────────────
|
||||
|
||||
async def list_rulebooks() -> dict:
|
||||
"""List all rulebooks owned by the current user.
|
||||
|
||||
Returns id, title, description for each.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rows = await rulebooks_svc.list_rulebooks(uid)
|
||||
return {"rulebooks": [rb.to_dict() for rb in rows]}
|
||||
|
||||
|
||||
async def get_rulebook(rulebook_id: int) -> dict:
|
||||
"""Fetch a rulebook by id with its full topic list."""
|
||||
uid = current_user_id()
|
||||
rb = await rulebooks_svc.get_rulebook(rulebook_id, uid)
|
||||
if rb is None:
|
||||
raise ValueError(f"rulebook {rulebook_id} not found")
|
||||
topics = await rulebooks_svc.list_topics(rulebook_id, uid)
|
||||
data = rb.to_dict()
|
||||
data["topics"] = [t.to_dict() for t in topics]
|
||||
return data
|
||||
|
||||
|
||||
async def create_rulebook(title: str, description: str = "") -> dict:
|
||||
"""Create a new rulebook.
|
||||
|
||||
Args:
|
||||
title: Rulebook name (e.g. "FabledSword family").
|
||||
description: Optional short description of what this rulebook covers.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rb = await rulebooks_svc.create_rulebook(
|
||||
user_id=uid, title=title, description=description,
|
||||
)
|
||||
return rb.to_dict()
|
||||
|
||||
|
||||
async def update_rulebook(
|
||||
rulebook_id: int, title: str = "", description: str = "",
|
||||
) -> dict:
|
||||
"""Update an existing rulebook. Only non-empty fields are changed."""
|
||||
uid = current_user_id()
|
||||
fields: dict = {}
|
||||
if title:
|
||||
fields["title"] = title
|
||||
if description:
|
||||
fields["description"] = description
|
||||
rb = await rulebooks_svc.update_rulebook(rulebook_id, uid, **fields)
|
||||
if rb is None:
|
||||
raise ValueError(f"rulebook {rulebook_id} not found")
|
||||
return rb.to_dict()
|
||||
|
||||
|
||||
async def delete_rulebook(rulebook_id: int, confirmed: bool = False) -> dict:
|
||||
"""Permanently delete a rulebook (cascades to all its topics and rules).
|
||||
|
||||
Pass confirmed=True to actually delete. Without confirmation, returns a
|
||||
preview describing what will be cascaded.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rb = await rulebooks_svc.get_rulebook(rulebook_id, uid)
|
||||
if rb is None:
|
||||
raise ValueError(f"rulebook {rulebook_id} not found")
|
||||
if not confirmed:
|
||||
topics = await rulebooks_svc.list_topics(rulebook_id, uid)
|
||||
rule_count = 0
|
||||
for t in topics:
|
||||
rule_count += len(await rulebooks_svc.list_rules(uid, topic_id=t.id))
|
||||
return {
|
||||
"warning": (
|
||||
f"Rulebook {rulebook_id} ('{rb.title}') contains "
|
||||
f"{len(topics)} topics and {rule_count} rules; all will be "
|
||||
f"deleted. Pass confirmed=True to proceed."
|
||||
),
|
||||
"confirmed_required": True,
|
||||
}
|
||||
await rulebooks_svc.delete_rulebook(rulebook_id, uid)
|
||||
return {"deleted": rulebook_id}
|
||||
|
||||
|
||||
# ── Topic CRUD ─────────────────────────────────────────────────────────
|
||||
|
||||
async def list_topics(rulebook_id: int) -> dict:
|
||||
"""List topics inside a rulebook."""
|
||||
uid = current_user_id()
|
||||
rows = await rulebooks_svc.list_topics(rulebook_id, uid)
|
||||
return {"topics": [t.to_dict() for t in rows]}
|
||||
|
||||
|
||||
async def create_topic(
|
||||
rulebook_id: int, title: str,
|
||||
description: str = "", order_index: int = 0,
|
||||
) -> dict:
|
||||
"""Create a topic within a rulebook.
|
||||
|
||||
Args:
|
||||
rulebook_id: Rulebook to add the topic to.
|
||||
title: Topic name (e.g. "git-workflow").
|
||||
description: Optional description.
|
||||
order_index: Display order (0-based; default 0).
|
||||
"""
|
||||
uid = current_user_id()
|
||||
topic = await rulebooks_svc.create_topic(
|
||||
rulebook_id=rulebook_id, user_id=uid,
|
||||
title=title, description=description, order_index=order_index,
|
||||
)
|
||||
return topic.to_dict()
|
||||
|
||||
|
||||
async def update_topic(
|
||||
topic_id: int, title: str = "",
|
||||
description: str = "", order_index: int = -1,
|
||||
) -> dict:
|
||||
"""Update a topic. Sentinels: title="" / description="" leave unchanged;
|
||||
order_index=-1 leaves unchanged.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
fields: dict = {}
|
||||
if title:
|
||||
fields["title"] = title
|
||||
if description:
|
||||
fields["description"] = description
|
||||
if order_index >= 0:
|
||||
fields["order_index"] = order_index
|
||||
topic = await rulebooks_svc.update_topic(topic_id, uid, **fields)
|
||||
if topic is None:
|
||||
raise ValueError(f"topic {topic_id} not found")
|
||||
return topic.to_dict()
|
||||
|
||||
|
||||
async def delete_topic(topic_id: int, confirmed: bool = False) -> dict:
|
||||
"""Delete a topic and all its rules. Requires confirmed=True."""
|
||||
uid = current_user_id()
|
||||
topic = await rulebooks_svc.get_topic(topic_id, uid)
|
||||
if topic is None:
|
||||
raise ValueError(f"topic {topic_id} not found")
|
||||
if not confirmed:
|
||||
rules = await rulebooks_svc.list_rules(uid, topic_id=topic_id)
|
||||
return {
|
||||
"warning": (
|
||||
f"Topic {topic_id} ('{topic.title}') contains {len(rules)} "
|
||||
f"rules; all will be deleted. Pass confirmed=True to proceed."
|
||||
),
|
||||
"confirmed_required": True,
|
||||
}
|
||||
await rulebooks_svc.delete_topic(topic_id, uid)
|
||||
return {"deleted": topic_id}
|
||||
|
||||
|
||||
# ── Rule CRUD ──────────────────────────────────────────────────────────
|
||||
|
||||
async def list_rules(
|
||||
rulebook_id: int = 0, topic_id: int = 0, project_id: int = 0,
|
||||
) -> dict:
|
||||
"""List rules — filter by rulebook, topic, and/or project.
|
||||
|
||||
Args:
|
||||
rulebook_id: 0 = no filter; positive = restrict to that rulebook.
|
||||
topic_id: 0 = no filter; positive = restrict to that topic.
|
||||
project_id: 0 = no filter; positive = restrict to rules applicable
|
||||
to that project (via its rulebook subscriptions).
|
||||
|
||||
All filters are AND-combined; ownership-scoped.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rows = await rulebooks_svc.list_rules(
|
||||
user_id=uid,
|
||||
rulebook_id=rulebook_id or None,
|
||||
topic_id=topic_id or None,
|
||||
project_id=project_id or None,
|
||||
)
|
||||
return {
|
||||
"rules": [
|
||||
{
|
||||
"id": r.id, "title": r.title, "statement": r.statement,
|
||||
"topic_id": r.topic_id,
|
||||
}
|
||||
for r in rows
|
||||
],
|
||||
"total": len(rows),
|
||||
}
|
||||
|
||||
|
||||
async def get_rule(rule_id: int) -> dict:
|
||||
"""Fetch a rule by id — full statement + why + how_to_apply."""
|
||||
uid = current_user_id()
|
||||
rule = await rulebooks_svc.get_rule(rule_id, uid)
|
||||
if rule is None:
|
||||
raise ValueError(f"rule {rule_id} not found")
|
||||
return rule.to_dict()
|
||||
|
||||
|
||||
async def create_rule(
|
||||
topic_id: int, title: str, statement: str,
|
||||
why: str = "", how_to_apply: str = "", order_index: int = 0,
|
||||
) -> dict:
|
||||
"""Create a new rule under a topic.
|
||||
|
||||
Args:
|
||||
topic_id: The topic to attach the rule to.
|
||||
title: A short imperative title (e.g. "dev is home").
|
||||
statement: The actionable instruction (required). 1-2 sentences.
|
||||
why: Optional rationale — the reason the rule exists.
|
||||
how_to_apply: Optional operationalization — when / where it kicks in.
|
||||
order_index: Display order within the topic (default 0).
|
||||
"""
|
||||
uid = current_user_id()
|
||||
rule = await rulebooks_svc.create_rule(
|
||||
topic_id=topic_id, user_id=uid,
|
||||
title=title, statement=statement,
|
||||
why=why, how_to_apply=how_to_apply, order_index=order_index,
|
||||
)
|
||||
return rule.to_dict()
|
||||
|
||||
|
||||
async def update_rule(
|
||||
rule_id: int, title: str = "", statement: str = "",
|
||||
why: str = "", how_to_apply: str = "", order_index: int = -1,
|
||||
) -> dict:
|
||||
"""Update a rule. Empty strings / order_index=-1 leave fields unchanged."""
|
||||
uid = current_user_id()
|
||||
fields: dict = {}
|
||||
if title:
|
||||
fields["title"] = title
|
||||
if statement:
|
||||
fields["statement"] = statement
|
||||
if why:
|
||||
fields["why"] = why
|
||||
if how_to_apply:
|
||||
fields["how_to_apply"] = how_to_apply
|
||||
if order_index >= 0:
|
||||
fields["order_index"] = order_index
|
||||
rule = await rulebooks_svc.update_rule(rule_id, uid, **fields)
|
||||
if rule is None:
|
||||
raise ValueError(f"rule {rule_id} not found")
|
||||
return rule.to_dict()
|
||||
|
||||
|
||||
async def delete_rule(rule_id: int, confirmed: bool = False) -> dict:
|
||||
"""Permanently delete a rule. Requires confirmed=True."""
|
||||
uid = current_user_id()
|
||||
rule = await rulebooks_svc.get_rule(rule_id, uid)
|
||||
if rule is None:
|
||||
raise ValueError(f"rule {rule_id} not found")
|
||||
if not confirmed:
|
||||
return {
|
||||
"warning": (
|
||||
f"Rule {rule_id} ('{rule.title}') will be permanently deleted. "
|
||||
f"Pass confirmed=True to proceed."
|
||||
),
|
||||
"confirmed_required": True,
|
||||
}
|
||||
await rulebooks_svc.delete_rule(rule_id, uid)
|
||||
return {"deleted": rule_id}
|
||||
|
||||
|
||||
# ── Subscriptions ──────────────────────────────────────────────────────
|
||||
|
||||
async def subscribe_project_to_rulebook(
|
||||
project_id: int, rulebook_id: int,
|
||||
) -> dict:
|
||||
"""Subscribe a project to a rulebook. Its rules will apply to that project."""
|
||||
uid = current_user_id()
|
||||
await rulebooks_svc.subscribe_project(
|
||||
project_id=project_id, rulebook_id=rulebook_id, user_id=uid,
|
||||
)
|
||||
return {"project_id": project_id, "rulebook_id": rulebook_id, "subscribed": True}
|
||||
|
||||
|
||||
async def unsubscribe_project_from_rulebook(
|
||||
project_id: int, rulebook_id: int,
|
||||
) -> dict:
|
||||
"""Remove a project's subscription to a rulebook."""
|
||||
uid = current_user_id()
|
||||
await rulebooks_svc.unsubscribe_project(
|
||||
project_id=project_id, rulebook_id=rulebook_id, user_id=uid,
|
||||
)
|
||||
return {"project_id": project_id, "rulebook_id": rulebook_id, "subscribed": False}
|
||||
|
||||
|
||||
def register(mcp) -> None:
|
||||
for fn in (
|
||||
list_rulebooks, get_rulebook, create_rulebook, update_rulebook, delete_rulebook,
|
||||
list_topics, create_topic, update_topic, delete_topic,
|
||||
list_rules, get_rule, create_rule, update_rule, delete_rule,
|
||||
subscribe_project_to_rulebook, unsubscribe_project_from_rulebook,
|
||||
):
|
||||
mcp.tool(name=fn.__name__)(fn)
|
||||
Reference in New Issue
Block a user