diff --git a/src/fabledassistant/mcp/tools/__init__.py b/src/fabledassistant/mcp/tools/__init__.py
index d162f6d..f3ec021 100644
--- a/src/fabledassistant/mcp/tools/__init__.py
+++ b/src/fabledassistant/mcp/tools/__init__.py
@@ -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)
diff --git a/src/fabledassistant/mcp/tools/rulebooks.py b/src/fabledassistant/mcp/tools/rulebooks.py
new file mode 100644
index 0000000..adb9ba3
--- /dev/null
+++ b/src/fabledassistant/mcp/tools/rulebooks.py
@@ -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)
diff --git a/tests/test_mcp_tool_rulebooks.py b/tests/test_mcp_tool_rulebooks.py
new file mode 100644
index 0000000..84e5d55
--- /dev/null
+++ b/tests/test_mcp_tool_rulebooks.py
@@ -0,0 +1,185 @@
+"""Tests for MCP rulebook tools — patches the service layer."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from fabledassistant.mcp._context import _user_id_ctx
+
+
+@pytest.fixture(autouse=True)
+def _bind_user():
+    token = _user_id_ctx.set(7)
+    yield
+    _user_id_ctx.reset(token)
+
+
+def _fake_rulebook(id=1, title="t"):
+    rb = MagicMock()
+    rb.id = id
+    rb.title = title
+    rb.to_dict.return_value = {"id": id, "title": title}
+    return rb
+
+
+def _fake_topic(id=10, title="git"):
+    t = MagicMock()
+    t.id = id
+    t.title = title
+    t.to_dict.return_value = {"id": id, "title": title}
+    return t
+
+
+def _fake_rule(id=100, title="r", statement="s"):
+    r = MagicMock()
+    r.id = id
+    r.title = title
+    r.topic_id = 10
+    r.statement = statement
+    r.to_dict.return_value = {"id": id, "title": title, "statement": statement}
+    return r
+
+
+@pytest.mark.asyncio
+async def test_list_rulebooks_wraps_in_dict():
+    rows = [_fake_rulebook(id=1), _fake_rulebook(id=2)]
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.list_rulebooks",
+        AsyncMock(return_value=rows),
+    ):
+        from fabledassistant.mcp.tools.rulebooks import list_rulebooks
+        out = await list_rulebooks()
+    assert len(out["rulebooks"]) == 2
+
+
+@pytest.mark.asyncio
+async def test_get_rulebook_includes_topics():
+    rb = _fake_rulebook(id=1)
+    topics = [_fake_topic(id=10), _fake_topic(id=11)]
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.get_rulebook",
+        AsyncMock(return_value=rb),
+    ), patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.list_topics",
+        AsyncMock(return_value=topics),
+    ):
+        from fabledassistant.mcp.tools.rulebooks import get_rulebook
+        out = await get_rulebook(rulebook_id=1)
+    assert out["id"] == 1
+    assert len(out["topics"]) == 2
+
+
+@pytest.mark.asyncio
+async def test_get_rulebook_raises_when_not_found():
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.get_rulebook",
+        AsyncMock(return_value=None),
+    ):
+        from fabledassistant.mcp.tools.rulebooks import get_rulebook
+        with pytest.raises(ValueError, match="rulebook 999 not found"):
+            await get_rulebook(rulebook_id=999)
+
+
+@pytest.mark.asyncio
+async def test_create_rule_passes_required_fields():
+    rule = _fake_rule()
+    mock = AsyncMock(return_value=rule)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.create_rule", mock):
+        from fabledassistant.mcp.tools.rulebooks import create_rule
+        await create_rule(
+            topic_id=10, title="dev is home", statement="Work directly on dev",
+        )
+    kwargs = mock.call_args.kwargs
+    assert kwargs["user_id"] == 7
+    assert kwargs["topic_id"] == 10
+    assert kwargs["statement"] == "Work directly on dev"
+
+
+@pytest.mark.asyncio
+async def test_update_rule_only_sends_non_default_fields():
+    rule = _fake_rule()
+    mock = AsyncMock(return_value=rule)
+    with patch("fabledassistant.mcp.tools.rulebooks.rulebooks_svc.update_rule", mock):
+        from fabledassistant.mcp.tools.rulebooks import update_rule
+        await update_rule(rule_id=1, statement="new statement")
+    args, kwargs = mock.call_args
+    assert args == (1, 7)
+    assert kwargs == {"statement": "new statement"}
+
+
+@pytest.mark.asyncio
+async def test_delete_rule_without_confirmed_returns_warning():
+    """delete_rule with confirmed=False returns a preview, not an action."""
+    rule = _fake_rule()
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.get_rule",
+        AsyncMock(return_value=rule),
+    ), patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.delete_rule",
+        AsyncMock(),
+    ) as mock_delete:
+        from fabledassistant.mcp.tools.rulebooks import delete_rule
+        out = await delete_rule(rule_id=1, confirmed=False)
+    assert out.get("confirmed_required") is True
+    assert "confirmed=True" in out.get("warning", "")
+    assert not mock_delete.called  # service NOT called
+
+
+@pytest.mark.asyncio
+async def test_delete_rule_with_confirmed_calls_service():
+    rule = _fake_rule()
+    mock_delete = AsyncMock()
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.get_rule",
+        AsyncMock(return_value=rule),
+    ), patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.delete_rule",
+        mock_delete,
+    ):
+        from fabledassistant.mcp.tools.rulebooks import delete_rule
+        out = await delete_rule(rule_id=1, confirmed=True)
+    assert out == {"deleted": 1}
+    assert mock_delete.called
+
+
+@pytest.mark.asyncio
+async def test_subscribe_project_to_rulebook_calls_service():
+    mock = AsyncMock()
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.subscribe_project", mock,
+    ):
+        from fabledassistant.mcp.tools.rulebooks import subscribe_project_to_rulebook
+        out = await subscribe_project_to_rulebook(project_id=3, rulebook_id=1)
+    assert out["subscribed"] is True
+    assert mock.called
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_project_from_rulebook_calls_service():
+    mock = AsyncMock()
+    with patch(
+        "fabledassistant.mcp.tools.rulebooks.rulebooks_svc.unsubscribe_project", mock,
+    ):
+        from fabledassistant.mcp.tools.rulebooks import unsubscribe_project_from_rulebook
+        out = await unsubscribe_project_from_rulebook(project_id=3, rulebook_id=1)
+    assert out["subscribed"] is False
+    assert mock.called
+
+
+def test_register_attaches_all_sixteen_tools():
+    """register(mcp) should call mcp.tool(name=...) for all 16 tools."""
+    from fabledassistant.mcp.tools.rulebooks import register
+    registered: list[str] = []
+
+    class FakeMCP:
+        def tool(self, name=None):
+            def decorator(fn):
+                registered.append(name)
+                return fn
+            return decorator
+
+    register(FakeMCP())
+    assert len(registered) == 16
+    # spot-check a few names
+    assert "list_rulebooks" in registered
+    assert "create_rule" in registered
+    assert "subscribe_project_to_rulebook" in registered