FabledScribe/docs/2026-03-23-fable-mcp-plan.md

Fable MCP Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add API key auth to Fabled Assistant and build a standalone Python MCP server that gives Claude full CRUD access to notes, tasks, projects, milestones, semantic search, and chat.

Architecture: Phase 1 adds bearer token auth to the existing Quart app (new api_keys table, updated _check_auth, settings UI tab). Phase 2 is a standalone fable-mcp/ Python package using the mcp[cli] SDK that calls the Fable HTTP API via httpx. The two phases are sequential — Phase 2 can be built/tested independently using a write-scoped API key once Phase 1 is done.

Deployment decision: fable-mcp/ stays permanently inside the scribe repo. It will always be versioned alongside the backend it targets. Future Task A will serve the package from the running Fable Docker image so users can install it directly from their instance.

Tech Stack: Python 3.12, Quart (Phase 1); mcp[cli], httpx, python-dotenv (Phase 2); Vue 3 + TypeScript (Settings UI); pytest for both.

Testing convention: Phase 1 tests run via docker compose run --rm app pytest tests/<file> -v. Phase 2 tests run locally with cd fable-mcp && pytest -v. All Phase 1 tests are unit tests of pure functions — mock DB calls with unittest.mock.AsyncMock; never connect to a real database in tests.

Important discovery: Conversation.conversation_type column already exists in the model (models/conversation.py:23) and list_conversations already filters by it (services/chat.py:49). No migration needed for conversation types — only service/route wiring.

---

Phase 1: Fable API Key Feature

File Map

Action	File	Purpose
Create	alembic/versions/0027_add_api_keys.py	DB migration for api_keys table
Create	src/scribe/models/api_key.py	ApiKey SQLAlchemy model
Modify	src/scribe/models/__init__.py	Export ApiKey
Create	src/scribe/services/api_keys.py	create/list/revoke/lookup service functions
Modify	src/scribe/auth.py	Add bearer token check before session fallback
Create	src/scribe/routes/api_keys.py	GET/POST/DELETE /api/api-keys blueprint
Modify	src/scribe/app.py	Register api_keys_bp and search_bp
Modify	src/scribe/services/chat.py:17-30	Add conversation_type param to create_conversation
Modify	src/scribe/services/chat.py:123-135	Exclude "mcp" type from cleanup_old_conversations
Modify	src/scribe/routes/chat.py:73-79	Pass conversation_type from POST body
Create	src/scribe/routes/search.py	GET /api/search semantic search endpoint
Modify	frontend/src/views/SettingsView.vue	Add "API Keys" tab
Create	tests/test_api_keys.py	Unit tests for service + auth
Create	tests/test_search_route.py	Unit test for search endpoint

---

Task 1: ApiKey model + migration

Files:

• Create: src/scribe/models/api_key.py

• Create: alembic/versions/0027_add_api_keys.py

• Modify: src/scribe/models/__init__.py

• Step 1: Write the model

Create src/scribe/models/api_key.py:

from datetime import datetime, timezone

from sqlalchemy import DateTime, ForeignKey, Index, Integer, Text
from sqlalchemy.orm import Mapped, mapped_column

from scribe.models import Base
from scribe.models.base import CreatedAtMixin


class ApiKey(Base, CreatedAtMixin):
    __tablename__ = "api_keys"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
    )
    name: Mapped[str] = mapped_column(Text, nullable=False)
    key_hash: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
    key_prefix: Mapped[str] = mapped_column(Text, nullable=False)
    scope: Mapped[str] = mapped_column(Text, nullable=False)  # "read" or "write"
    last_used_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    revoked_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )

    __table_args__ = (
        Index("ix_api_keys_user_id", "user_id"),
        Index("ix_api_keys_key_hash", "key_hash"),
    )

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "name": self.name,
            "key_prefix": self.key_prefix,
            "scope": self.scope,
            "last_used_at": self.last_used_at.isoformat() if self.last_used_at else None,
            "created_at": self.created_at.isoformat(),
            "revoked_at": self.revoked_at.isoformat() if self.revoked_at else None,
        }

• Step 2: Export from models init

In src/scribe/models/__init__.py, add after the last import line:

from scribe.models.api_key import ApiKey  # noqa: E402, F401

• Step 3: Write the migration

Create alembic/versions/0027_add_api_keys.py:

"""add api_keys table

Revision ID: 0027
Revises: 0026
Create Date: 2026-03-23
"""
from alembic import op
import sqlalchemy as sa

revision = "0027"
down_revision = "0026"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "api_keys",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        sa.Column("name", sa.Text(), nullable=False),
        sa.Column("key_hash", sa.Text(), nullable=False, unique=True),
        sa.Column("key_prefix", sa.Text(), nullable=False),
        sa.Column("scope", sa.Text(), nullable=False),
        sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
    )
    op.create_index("ix_api_keys_user_id", "api_keys", ["user_id"])
    op.create_index("ix_api_keys_key_hash", "api_keys", ["key_hash"])


def downgrade() -> None:
    op.drop_table("api_keys")

• Step 4: Verify migration runs

docker compose run --rm app alembic upgrade head

Expected: Running upgrade 0026 -> 0027, add api_keys table

• Step 5: Commit

git add src/scribe/models/api_key.py \
        src/scribe/models/__init__.py \
        alembic/versions/0027_add_api_keys.py
git commit -m "feat: add ApiKey model and migration 0027"

---

Task 2: ApiKey service

Files:

• Create: src/scribe/services/api_keys.py

• Create: tests/test_api_keys.py (service tests)

• Step 1: Write the failing tests

Create tests/test_api_keys.py:

"""Unit tests for api_keys service — pure logic, DB calls mocked."""
import hashlib
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from scribe.services.api_keys import (
    _hash_key,
    generate_key,
    create_api_key,
    list_api_keys,
    revoke_api_key,
    lookup_key,
)


def test_generate_key_format():
    key = generate_key()
    assert key.startswith("fmcp_")
    assert len(key) > 12  # prefix + at least some random chars


def test_generate_key_uniqueness():
    keys = {generate_key() for _ in range(100)}
    assert len(keys) == 100  # no duplicates


def test_hash_key_is_sha256():
    key = "fmcp_testkey"
    h = _hash_key(key)
    expected = hashlib.sha256(key.encode()).hexdigest()
    assert h == expected


def test_generate_key_prefix():
    key = "fmcp_abcdefghijklmnop"
    # prefix is first 12 chars of the full key
    from scribe.services.api_keys import _key_prefix
    assert _key_prefix(key) == "fmcp_abcdefg"  # first 12 chars


@pytest.mark.asyncio
async def test_create_api_key_returns_full_key():
    mock_key_obj = MagicMock()
    mock_key_obj.id = 1
    mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}

    with patch("scribe.services.api_keys.async_session") as mock_session_ctx:
        mock_session = AsyncMock()
        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
        mock_session.__aexit__ = AsyncMock(return_value=False)
        mock_session.add = MagicMock()
        mock_session.commit = AsyncMock()
        mock_session_ctx.return_value = mock_session

        full_key, key_dict = await create_api_key(user_id=1, name="test", scope="read")

    assert full_key.startswith("fmcp_")
    assert isinstance(key_dict, dict)


@pytest.mark.asyncio
async def test_lookup_key_returns_none_for_unknown():
    with patch("scribe.services.api_keys.async_session") as mock_session_ctx:
        mock_session = AsyncMock()
        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
        mock_session.__aexit__ = AsyncMock(return_value=False)
        mock_result = MagicMock()
        mock_result.scalars.return_value.first.return_value = None
        mock_session.execute = AsyncMock(return_value=mock_result)
        mock_session_ctx.return_value = mock_session

        result = await lookup_key("fmcp_doesnotexist")

    assert result is None

• Step 2: Run tests to verify they fail

docker compose run --rm app pytest tests/test_api_keys.py -v

Expected: ImportError or ModuleNotFoundError (service doesn't exist yet)

• Step 3: Write the service

Create src/scribe/services/api_keys.py:

import hashlib
import secrets
from datetime import datetime, timezone

from sqlalchemy import select

from scribe.models import async_session
from scribe.models.api_key import ApiKey


def generate_key() -> str:
    """Generate a new full API key. Never stored — caller must hash it."""
    return "fmcp_" + secrets.token_urlsafe(32)


def _hash_key(key: str) -> str:
    return hashlib.sha256(key.encode()).hexdigest()


def _key_prefix(key: str) -> str:
    """Return first 12 chars of the key for display (e.g. 'fmcp_abcdefg')."""
    return key[:12]


async def create_api_key(
    user_id: int, name: str, scope: str
) -> tuple[str, dict]:
    """Create a new API key. Returns (full_key, key_dict). full_key is never stored."""
    if scope not in ("read", "write"):
        raise ValueError("scope must be 'read' or 'write'")
    full_key = generate_key()
    key = ApiKey(
        user_id=user_id,
        name=name,
        key_hash=_hash_key(full_key),
        key_prefix=_key_prefix(full_key),
        scope=scope,
    )
    async with async_session() as session:
        session.add(key)
        await session.commit()
        await session.refresh(key)
        return full_key, key.to_dict()


async def list_api_keys(user_id: int) -> list[dict]:
    """List all non-revoked API keys for the user."""
    async with async_session() as session:
        result = await session.execute(
            select(ApiKey)
            .where(ApiKey.user_id == user_id, ApiKey.revoked_at.is_(None))
            .order_by(ApiKey.created_at.desc())
        )
        return [k.to_dict() for k in result.scalars().all()]


async def revoke_api_key(user_id: int, key_id: int) -> bool:
    """Soft-delete a key by setting revoked_at. Returns True if found."""
    async with async_session() as session:
        result = await session.execute(
            select(ApiKey).where(ApiKey.id == key_id, ApiKey.user_id == user_id)
        )
        key = result.scalars().first()
        if key is None:
            return False
        key.revoked_at = datetime.now(timezone.utc)
        await session.commit()
        return True


async def lookup_key(raw_key: str) -> ApiKey | None:
    """Look up a non-revoked ApiKey by raw token value. Updates last_used_at."""
    key_hash = _hash_key(raw_key)
    async with async_session() as session:
        result = await session.execute(
            select(ApiKey).where(
                ApiKey.key_hash == key_hash,
                ApiKey.revoked_at.is_(None),
            )
        )
        key = result.scalars().first()
        if key is None:
            return None
        key.last_used_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(key)
        return key

• Step 4: Run tests to verify they pass

docker compose run --rm app pytest tests/test_api_keys.py -v

Expected: all tests pass

• Step 5: Commit

git add src/scribe/services/api_keys.py tests/test_api_keys.py
git commit -m "feat: add ApiKey service with create/list/revoke/lookup"

---

Task 3: Auth middleware — bearer token support

Files:

• Modify: src/scribe/auth.py

• Modify: tests/test_api_keys.py (add auth middleware tests)

• Step 1: Add auth middleware tests

Append to tests/test_api_keys.py:

def test_hash_key_deterministic():
    """Same input always produces same hash (needed for lookup)."""
    key = "fmcp_some_test_key_value"
    assert _hash_key(key) == _hash_key(key)


def test_scope_validation():
    with pytest.raises(ValueError, match="scope must be"):
        import asyncio
        asyncio.run(create_api_key(1, "bad", "superadmin"))


# --- Auth middleware tests ---

@pytest.mark.asyncio
async def test_bearer_token_path_sets_g_user(monkeypatch):
    """Valid bearer token authenticates and sets g.user and g.api_key."""
    from unittest.mock import AsyncMock, MagicMock
    from scribe.auth import _check_auth

    # Mock ApiKey object
    fake_key = MagicMock()
    fake_key.user_id = 7
    fake_key.scope = "write"

    # Mock User object
    fake_user = MagicMock()
    fake_user.role = "user"

    monkeypatch.setattr("scribe.auth.lookup_key", AsyncMock(return_value=fake_key))
    monkeypatch.setattr("scribe.auth.get_user_by_id", AsyncMock(return_value=fake_user))

    called_with_user = {}

    async def view_func():
        from quart import g
        called_with_user["user"] = g.user
        called_with_user["api_key"] = g.api_key
        from quart import jsonify
        return jsonify({"ok": True})

    from quart import Quart
    app = Quart(__name__)

    @app.route("/test", methods=["GET"])
    @_check_auth
    async def protected():
        return await view_func()

    async with app.test_request_context("/test", method="GET",
                                        headers={"Authorization": "Bearer fmcp_valid"}):
        # Just verify _check_auth calls lookup_key with the right token
        import scribe.auth as auth_module
        auth_module.lookup_key.assert_called_with  # callable


@pytest.mark.asyncio
async def test_read_only_key_blocked_on_post():
    """Read-only API key returns 403 on non-GET requests."""
    from unittest.mock import AsyncMock, MagicMock
    from scribe.auth import _check_auth
    from quart import Quart

    fake_key = MagicMock()
    fake_key.user_id = 7
    fake_key.scope = "read"

    fake_user = MagicMock()
    fake_user.role = "user"

    app = Quart(__name__)

    with (
        app.test_request_context("/test", method="POST",
                                 headers={"Authorization": "Bearer fmcp_readonly"}),
    ):
        from unittest.mock import patch
        with patch("scribe.auth.lookup_key", AsyncMock(return_value=fake_key)), \
             patch("scribe.auth.get_user_by_id", AsyncMock(return_value=fake_user)):

            async def dummy():
                return "ok"

            wrapped = _check_auth(dummy)
            from quart import current_app
            async with app.app_context():
                resp = await wrapped()
            # Should return 403 for read-only key on POST
            assert resp[1] == 403

• Step 2: Run to verify pass (these are pure logic tests)

docker compose run --rm app pytest tests/test_api_keys.py -v

• Step 3: Update auth.py

Replace src/scribe/auth.py with:

import functools

from quart import g, jsonify, request, session

from scribe.services.auth import get_user_by_id
from scribe.services.api_keys import lookup_key


def _check_auth(f, required_role: str | None = None):
    @functools.wraps(f)
    async def decorated(*args, **kwargs):
        # --- Bearer token path ---
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            raw_key = auth_header[len("Bearer "):]
            api_key = await lookup_key(raw_key)
            if api_key is None:
                return jsonify({"error": "Invalid or revoked API key"}), 401
            user = await get_user_by_id(api_key.user_id)
            if not user:
                return jsonify({"error": "User not found"}), 401
            # Scope enforcement: read-only keys cannot mutate
            if api_key.scope == "read" and request.method not in ("GET", "HEAD", "OPTIONS"):
                return jsonify({"error": "Read-only key cannot perform write operations"}), 403
            # Role check (admin_required routes)
            if required_role and user.role != required_role:
                return jsonify({"error": "Admin access required"}), 403
            g.user = user
            g.api_key = api_key
            return await f(*args, **kwargs)

        # --- Session path (unchanged) ---
        user_id = session.get("user_id")
        if not user_id:
            return jsonify({"error": "Authentication required"}), 401
        user = await get_user_by_id(user_id)
        if not user:
            session.clear()
            return jsonify({"error": "Authentication required"}), 401
        if session.get("session_version") != user.session_version:
            session.clear()
            return jsonify({"error": "Session expired. Please log in again."}), 401
        if required_role and user.role != required_role:
            return jsonify({"error": "Admin access required"}), 403
        g.user = user
        return await f(*args, **kwargs)
    return decorated


def login_required(f):
    return _check_auth(f)


def admin_required(f):
    return _check_auth(f, required_role="admin")


def get_current_user_id() -> int:
    return g.user.id

• Step 4: Run the full test suite to check for regressions

docker compose run --rm app pytest tests/ -v

Expected: all existing tests still pass

• Step 5: Commit

git add src/scribe/auth.py tests/test_api_keys.py
git commit -m "feat: add bearer token auth to _check_auth, falls back to session"

---

Task 4: API key routes + app registration

Files:

• Create: src/scribe/routes/api_keys.py

• Modify: src/scribe/app.py

• Step 1: Write the routes

Create src/scribe/routes/api_keys.py:

from quart import Blueprint, jsonify, request

from scribe.auth import login_required, get_current_user_id
from scribe.services.api_keys import create_api_key, list_api_keys, revoke_api_key

api_keys_bp = Blueprint("api_keys", __name__, url_prefix="/api/api-keys")


@api_keys_bp.route("", methods=["GET"])
@login_required
async def list_keys_route():
    uid = get_current_user_id()
    keys = await list_api_keys(uid)
    return jsonify({"api_keys": keys})


@api_keys_bp.route("", methods=["POST"])
@login_required
async def create_key_route():
    uid = get_current_user_id()
    data = await request.get_json(force=True, silent=True) or {}
    name = (data.get("name") or "").strip()
    scope = (data.get("scope") or "").strip()

    if not name:
        return jsonify({"error": "name is required"}), 400
    if scope not in ("read", "write"):
        return jsonify({"error": "scope must be 'read' or 'write'"}), 400

    full_key, key_dict = await create_api_key(uid, name=name, scope=scope)
    # Return the full key ONCE — it is never retrievable again
    return jsonify({"key": full_key, "api_key": key_dict}), 201


@api_keys_bp.route("/<int:key_id>", methods=["DELETE"])
@login_required
async def revoke_key_route(key_id: int):
    uid = get_current_user_id()
    deleted = await revoke_api_key(uid, key_id)
    if not deleted:
        return jsonify({"error": "API key not found"}), 404
    return "", 204

• Step 2: Register in app.py

In src/scribe/app.py, add the import alongside the other route imports:

from scribe.routes.api_keys import api_keys_bp

And add the registration line after app.register_blueprint(users_bp):

app.register_blueprint(api_keys_bp)

• Step 3: Smoke test the endpoints

docker compose up -d
# Create a key
curl -s -X POST http://localhost:8080/api/api-keys \
  -H "Content-Type: application/json" \
  -H "Cookie: <your session cookie>" \
  -d '{"name":"test","scope":"read"}'
# Expected: {"key": "fmcp_...", "api_key": {...}}

# Test bearer auth
curl -s http://localhost:8080/api/auth/me \
  -H "Authorization: Bearer fmcp_<key from above>"
# Expected: {"id": ..., "username": ...}

• Step 4: Commit

git add src/scribe/routes/api_keys.py src/scribe/app.py
git commit -m "feat: add API key CRUD routes and register blueprint"

---

Task 5: Conversation type wiring

Files:

• Modify: src/scribe/services/chat.py (lines 17-30 and 123-135)
• Modify: src/scribe/routes/chat.py (lines 73-79)

Note: Conversation.conversation_type already exists in the model. list_conversations already filters by conv_type. This task only wires up creation and retention exclusion.

• Step 1: Update create_conversation service

In src/scribe/services/chat.py, change the function signature at line 17:

async def create_conversation(
    user_id: int, title: str = "", model: str = "", conversation_type: str = "chat"
) -> Conversation:
    async with async_session() as session:
        conv = Conversation(
            user_id=user_id,
            title=title,
            model=model,
            conversation_type=conversation_type,
        )
        session.add(conv)
        await session.commit()
        result = await session.execute(
            select(Conversation)
            .options(selectinload(Conversation.messages))
            .where(Conversation.id == conv.id)
        )
        return result.scalars().first()

• Step 2: Update cleanup_old_conversations to exclude "mcp"

In src/scribe/services/chat.py, update the WHERE clause at line 130:

result = await session.execute(
    sa_delete(Conversation)
    .where(
        Conversation.user_id == user_id,
        Conversation.updated_at < cutoff,
        Conversation.conversation_type != "mcp",  # preserve MCP audit trail
    )
    .returning(Conversation.id)
)

• Step 3: Update the POST route to accept conversation_type

In src/scribe/routes/chat.py, update create_conversation_route (around line 73):

@chat_bp.route("/conversations", methods=["POST"])
@login_required
async def create_conversation_route():
    uid = get_current_user_id()
    data = await request.get_json(force=True, silent=True) or {}
    title = data.get("title", "")
    model = data.get("model", Config.OLLAMA_MODEL)
    conversation_type = data.get("conversation_type", "chat")
    # Only allow known types to prevent accidental misuse
    if conversation_type not in ("chat", "mcp"):
        conversation_type = "chat"
    conv = await create_conversation(uid, title=title, model=model, conversation_type=conversation_type)
    return jsonify(conv.to_dict()), 201

• Step 4: Run tests

docker compose run --rm app pytest tests/ -v

Expected: all tests pass

• Step 5: Commit

git add src/scribe/services/chat.py src/scribe/routes/chat.py
git commit -m "feat: wire conversation_type through create_conversation, exclude mcp from retention sweep"

---

Task 6: Semantic search endpoint

Files:

• Create: src/scribe/routes/search.py

• Modify: src/scribe/app.py

• Create: tests/test_search_route.py

• Step 1: Write the failing test

Create tests/test_search_route.py:

"""Unit tests for the search route parameter mapping."""
import pytest
from unittest.mock import patch, AsyncMock


def test_content_type_mapping():
    """Verify content_type string maps to correct is_task value."""
    from scribe.routes.search import _content_type_to_is_task
    assert _content_type_to_is_task("note") is False
    assert _content_type_to_is_task("task") is True
    assert _content_type_to_is_task("all") is None
    assert _content_type_to_is_task("unknown") is None  # default to all

• Step 2: Run to verify it fails

docker compose run --rm app pytest tests/test_search_route.py -v

Expected: ImportError (module doesn't exist yet)

• Step 3: Write the route

Create src/scribe/routes/search.py:

from quart import Blueprint, jsonify, request

from scribe.auth import login_required, get_current_user_id
from scribe.services.embeddings import semantic_search_notes

search_bp = Blueprint("search", __name__, url_prefix="/api/search")


def _content_type_to_is_task(content_type: str) -> bool | None:
    """Map content_type query param to semantic_search_notes is_task arg."""
    if content_type == "note":
        return False
    if content_type == "task":
        return True
    return None  # "all" or unknown → no filter


@search_bp.route("", methods=["GET"])
@login_required
async def search_route():
    uid = get_current_user_id()
    q = (request.args.get("q") or "").strip()
    if not q:
        return jsonify({"error": "q is required"}), 400

    content_type = request.args.get("content_type", "all")
    limit = min(request.args.get("limit", 10, type=int), 50)
    is_task = _content_type_to_is_task(content_type)

    results = await semantic_search_notes(
        uid, q, limit=limit, is_task=is_task, threshold=0.3
    )
    return jsonify({
        "results": [
            {
                "id": note.id,
                "title": note.title,
                "body": note.body or "",
                "is_task": note.is_task,
                "tags": note.tags or [],
                "similarity": score,
            }
            for score, note in results  # semantic_search_notes returns list[tuple[float, Note]]
        ],
        "total": len(results),
    })

Note: check services/embeddings.py to confirm the return type of semantic_search_notes — it should return a list of (Note, float) tuples. If the signature differs, adjust the destructuring above accordingly.

• Step 4: Register in app.py

Add to imports in src/scribe/app.py:

from scribe.routes.search import search_bp

Add registration:

app.register_blueprint(search_bp)

• Step 5: Run tests

docker compose run --rm app pytest tests/test_search_route.py tests/ -v

Expected: all tests pass

• Step 6: Commit

git add src/scribe/routes/search.py \
        src/scribe/app.py \
        tests/test_search_route.py
git commit -m "feat: add GET /api/search semantic search endpoint"

---

Task 7: Settings UI — API Keys tab

Files:

• Modify: frontend/src/views/SettingsView.vue

• Step 1: Add "apikeys" to VALID_TABS

In SettingsView.vue, find the VALID_TABS array (search for VALID_TABS) and add "apikeys" to it.

• Step 2: Add sidebar tab button

In the settings sidebar tab list, add an "API Keys" tab button alongside the existing ones, following the same pattern as existing tabs (look for the <button> elements with @click="activeTab = '...'" and add a matching one for "apikeys").

• Step 3: Add tab panel content

Add the API Keys panel inside the <div class="settings-content"> block, following the existing v-if="activeTab === '...'" pattern:

<!-- API Keys Tab -->
<div v-if="activeTab === 'apikeys'" class="settings-section">
  <h2>API Keys</h2>
  <p class="settings-description">
    API keys let external tools (like the Fable MCP server) access your data
    without a browser session. Write-scoped keys can create and update content;
    read-only keys can only query.
  </p>

  <!-- Create form -->
  <div class="api-key-create-form">
    <input
      v-model="newKeyName"
      placeholder="Key name (e.g. Claude MCP)"
      class="settings-input"
    />
    <div class="api-key-scope-select">
      <label>
        <input type="radio" v-model="newKeyScope" value="read" /> Read-only
      </label>
      <label>
        <input type="radio" v-model="newKeyScope" value="write" /> Read + Write
      </label>
    </div>
    <button @click="createKey" :disabled="!newKeyName || creatingKey" class="btn-primary">
      {{ creatingKey ? 'Generating...' : 'Generate Key' }}
    </button>
  </div>

  <!-- One-time key display modal -->
  <div v-if="newKeyValue" class="api-key-reveal">
    <p><strong>Copy this key now — it will not be shown again.</strong></p>
    <div class="api-key-value-row">
      <code class="api-key-value">{{ newKeyValue }}</code>
      <button @click="copyKey" class="btn-secondary">
        {{ keyCopied ? 'Copied!' : 'Copy' }}
      </button>
    </div>
    <button @click="newKeyValue = ''" class="btn-secondary" style="margin-top: 0.5rem;">
      Done
    </button>
  </div>

  <!-- Keys table -->
  <div v-if="apiKeys.length > 0" class="api-keys-list">
    <table class="api-keys-table">
      <thead>
        <tr>
          <th>Name</th>
          <th>Scope</th>
          <th>Prefix</th>
          <th>Last Used</th>
          <th></th>
        </tr>
      </thead>
      <tbody>
        <tr v-for="key in apiKeys" :key="key.id">
          <td>{{ key.name }}</td>
          <td><span :class="['scope-badge', key.scope]">{{ key.scope }}</span></td>
          <td><code>{{ key.key_prefix }}…</code></td>
          <td>{{ key.last_used_at ? new Date(key.last_used_at).toLocaleDateString() : 'Never' }}</td>
          <td>
            <button
              v-if="revokeConfirm !== key.id"
              @click="revokeConfirm = key.id"
              class="btn-danger-outline btn-sm"
            >Revoke</button>
            <span v-else>
              Sure?
              <button @click="revokeKey(key.id)" class="btn-danger btn-sm">Yes</button>
              <button @click="revokeConfirm = null" class="btn-secondary btn-sm">No</button>
            </span>
          </td>
        </tr>
      </tbody>
    </table>
  </div>
  <p v-else class="settings-empty">No API keys yet.</p>
</div>

• Step 4: Add reactive state and methods to the script section

In the <script setup> block, add alongside the existing reactive state:

// API Keys
const apiKeys = ref<Array<{id: number, name: string, scope: string, key_prefix: string, last_used_at: string | null}>>([])
const newKeyName = ref('')
const newKeyScope = ref<'read' | 'write'>('write')
const newKeyValue = ref('')
const keyCopied = ref(false)
const creatingKey = ref(false)
const revokeConfirm = ref<number | null>(null)

async function fetchApiKeys() {
  const res = await fetch('/api/api-keys', { credentials: 'include' })
  if (res.ok) {
    const data = await res.json()
    apiKeys.value = data.api_keys
  }
}

async function createKey() {
  if (!newKeyName.value) return
  creatingKey.value = true
  try {
    const res = await fetch('/api/api-keys', {
      method: 'POST',
      credentials: 'include',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ name: newKeyName.value, scope: newKeyScope.value }),
    })
    if (res.ok) {
      const data = await res.json()
      newKeyValue.value = data.key
      newKeyName.value = ''
      await fetchApiKeys()
    }
  } finally {
    creatingKey.value = false
  }
}

async function revokeKey(id: number) {
  await fetch(`/api/api-keys/${id}`, { method: 'DELETE', credentials: 'include' })
  revokeConfirm.value = null
  await fetchApiKeys()
}

async function copyKey() {
  await navigator.clipboard.writeText(newKeyValue.value)
  keyCopied.value = true
  setTimeout(() => { keyCopied.value = false }, 2000)
}

• Step 5: Fetch keys when tab activates

Find the existing watch(activeTab, ...) or onMounted logic and add a call to fetchApiKeys() when activeTab becomes "apikeys". If there's no watch on activeTab, add:

watch(activeTab, (tab) => {
  if (tab === 'apikeys') fetchApiKeys()
})

If a watch already exists, add the apikeys case to it.

• Step 6: TypeScript check

docker compose run --rm app sh -c "cd /app/frontend && npx tsc --noEmit"

Expected: no errors

• Step 7: Build and smoke test in browser

docker compose up --build -d

Navigate to Settings → API Keys. Create a key, copy it, verify it appears in the table. Test the bearer token with curl.

• Step 8: Commit

git add frontend/src/views/SettingsView.vue
git commit -m "feat: add API Keys tab to Settings with create/revoke UI"

---

Phase 2: Fable MCP Server

File Map

Action	File	Purpose
Create	fable-mcp/pyproject.toml	Package definition, entry point, deps
Create	fable-mcp/.env.example	Env var template
Create	fable-mcp/fable_mcp/__init__.py	Package marker
Create	fable-mcp/fable_mcp/client.py	FableClient async httpx wrapper + FableAPIError
Create	fable-mcp/fable_mcp/server.py	FastMCP instance + entry point main()
Create	fable-mcp/fable_mcp/tools/__init__.py	Package marker
Create	fable-mcp/fable_mcp/tools/notes.py	Notes CRUD tools
Create	fable-mcp/fable_mcp/tools/tasks.py	Tasks CRUD + status + log tools
Create	fable-mcp/fable_mcp/tools/projects.py	Projects CRUD tools
Create	fable-mcp/fable_mcp/tools/milestones.py	Milestones CRUD tools
Create	fable-mcp/fable_mcp/tools/search.py	Semantic search tool
Create	fable-mcp/fable_mcp/tools/chat.py	Chat send_message tool (SSE consumer)
Create	fable-mcp/tests/conftest.py	pytest fixtures (mock FableClient)
Create	fable-mcp/tests/test_client.py	FableClient unit tests
Create	fable-mcp/tests/test_tools_notes.py	Notes tool unit tests
Create	fable-mcp/tests/test_tools_chat.py	Chat tool unit tests (SSE parsing)

---

Task 8: Package scaffold

Files:

• Create: fable-mcp/pyproject.toml

• Create: fable-mcp/.env.example

• Create: fable-mcp/fable_mcp/__init__.py

• Create: fable-mcp/fable_mcp/tools/__init__.py

• Create: fable-mcp/tests/__init__.py

• Create: fable-mcp/tests/conftest.py

• Step 1: Create pyproject.toml

Create fable-mcp/pyproject.toml:

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "fable-mcp"
version = "0.1.0"
description = "MCP server for Fabled Assistant"
requires-python = ">=3.12"
dependencies = [
  "mcp[cli]>=1.0",
  "httpx>=0.27",
  "python-dotenv>=1.0",
]

[project.scripts]
fable-mcp = "fable_mcp.server:main"

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "respx>=0.21"]

• Step 2: Create .env.example

Create fable-mcp/.env.example:

FABLE_URL=http://localhost:8080
FABLE_API_KEY=fmcp_your_key_here

• Step 3: Create package markers

mkdir -p fable-mcp/fable_mcp/tools fable-mcp/tests
touch fable-mcp/fable_mcp/__init__.py
touch fable-mcp/fable_mcp/tools/__init__.py
touch fable-mcp/tests/__init__.py

• Step 4: Create tests/conftest.py

Create fable-mcp/tests/conftest.py:

"""Shared pytest fixtures for fable-mcp tests."""
import pytest
from unittest.mock import AsyncMock, MagicMock


@pytest.fixture
def mock_client():
    """A mock FableClient that returns empty responses by default."""
    client = AsyncMock()
    client.__aenter__ = AsyncMock(return_value=client)
    client.__aexit__ = AsyncMock(return_value=False)
    return client

• Step 5: Install in dev mode

cd fable-mcp && pip install -e ".[dev]"

• Step 6: Commit scaffold

git add fable-mcp/
git commit -m "feat: scaffold fable-mcp Python package"

---

Task 9: FableClient

Files:

• Create: fable-mcp/fable_mcp/client.py

• Create: fable-mcp/tests/test_client.py

• Step 1: Write the failing tests

Create fable-mcp/tests/test_client.py:

"""Unit tests for FableClient configuration and error handling."""
import os
import pytest
from unittest.mock import patch, MagicMock

import httpx
import respx

from fable_mcp.client import FableClient, FableAPIError


def test_missing_env_vars_raises():
    """Client raises clearly if FABLE_URL or FABLE_API_KEY are missing."""
    with patch.dict(os.environ, {}, clear=True):
        # Remove both vars
        os.environ.pop("FABLE_URL", None)
        os.environ.pop("FABLE_API_KEY", None)
        with pytest.raises(ValueError, match="FABLE_URL"):
            FableClient()


def test_auth_header_set():
    """Client sets Authorization header from FABLE_API_KEY."""
    with patch.dict(os.environ, {"FABLE_URL": "http://localhost:8080", "FABLE_API_KEY": "fmcp_test"}):
        client = FableClient()
        assert client._headers["Authorization"] == "Bearer fmcp_test"


@pytest.mark.asyncio
@respx.mock
async def test_get_raises_on_non_200():
    """FableAPIError is raised for non-2xx responses."""
    respx.get("http://localhost:8080/api/notes").mock(
        return_value=httpx.Response(403, json={"error": "forbidden"})
    )
    with patch.dict(os.environ, {"FABLE_URL": "http://localhost:8080", "FABLE_API_KEY": "fmcp_x"}):
        client = FableClient()
        async with client:
            with pytest.raises(FableAPIError) as exc_info:
                await client.get("/api/notes")
    assert exc_info.value.status_code == 403

• Step 2: Run to verify they fail

cd fable-mcp && pytest tests/test_client.py -v

Expected: ImportError (module doesn't exist yet)

• Step 3: Write the client

Create fable-mcp/fable_mcp/client.py:

"""Async HTTP client wrapping the Fable REST API."""
import os
from typing import Any

import httpx
from dotenv import load_dotenv

load_dotenv()


class FableAPIError(Exception):
    def __init__(self, status_code: int, message: str):
        self.status_code = status_code
        self.message = message
        super().__init__(f"Fable API error {status_code}: {message}")


class FableClient:
    """Async httpx wrapper for the Fable API. Use as async context manager."""

    def __init__(self):
        url = os.environ.get("FABLE_URL", "").rstrip("/")
        api_key = os.environ.get("FABLE_API_KEY", "")
        if not url:
            raise ValueError("FABLE_URL environment variable is required")
        if not api_key:
            raise ValueError("FABLE_API_KEY environment variable is required")
        self._base_url = url
        self._headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }
        self._http: httpx.AsyncClient | None = None

    async def __aenter__(self):
        self._http = httpx.AsyncClient(
            base_url=self._base_url,
            headers=self._headers,
            timeout=30.0,
        )
        return self

    async def __aexit__(self, *args):
        if self._http:
            await self._http.aclose()

    def _check_response(self, response: httpx.Response) -> dict:
        if response.status_code >= 400:
            try:
                detail = response.json().get("error", response.text)
            except Exception:
                detail = response.text
            raise FableAPIError(response.status_code, detail)
        if response.status_code == 204 or not response.content:
            return {}
        return response.json()

    async def get(self, path: str, **params) -> dict:
        resp = await self._http.get(path, params={k: v for k, v in params.items() if v is not None})
        return self._check_response(resp)

    async def post(self, path: str, body: dict | None = None) -> dict:
        resp = await self._http.post(path, json=body or {})
        return self._check_response(resp)

    async def patch(self, path: str, body: dict) -> dict:
        resp = await self._http.patch(path, json=body)
        return self._check_response(resp)

    async def put(self, path: str, body: dict) -> dict:
        resp = await self._http.put(path, json=body)
        return self._check_response(resp)

    async def delete(self, path: str) -> dict:
        resp = await self._http.delete(path)
        return self._check_response(resp)

    async def stream_get(self, path: str):
        """Yields raw SSE lines from a streaming GET endpoint."""
        async with self._http.stream("GET", path) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    yield line[len("data: "):]


# Module-level singleton — initialized in server.py startup
_client: FableClient | None = None


def get_client() -> FableClient:
    """Return the module-level singleton. Must be initialized before use."""
    if _client is None:
        raise RuntimeError("FableClient not initialized. Call init_client() first.")
    return _client


def init_client() -> FableClient:
    """Initialize the module-level singleton from environment variables."""
    global _client
    _client = FableClient()
    return _client

• Step 4: Run tests to verify they pass

cd fable-mcp && pytest tests/test_client.py -v

Expected: all tests pass

• Step 5: Commit

git add fable-mcp/fable_mcp/client.py fable-mcp/tests/test_client.py
git commit -m "feat: add FableClient async httpx wrapper with error handling"

---

Task 10: Notes tools

Files:

• Create: fable-mcp/fable_mcp/tools/notes.py

• Create: fable-mcp/tests/test_tools_notes.py

• Step 1: Write the failing tests

Create fable-mcp/tests/test_tools_notes.py:

"""Unit tests for notes tools — mock FableClient."""
import pytest
from unittest.mock import AsyncMock, patch


@pytest.mark.asyncio
async def test_list_notes_passes_params():
    """list_notes passes query params to client.get."""
    mock_get = AsyncMock(return_value={"notes": [], "total": 0})
    with patch("fable_mcp.tools.notes.get_client") as mock_gc:
        mock_gc.return_value.get = mock_get
        from fable_mcp.tools.notes import list_notes
        result = await list_notes(query="auth", limit=5)
    mock_get.assert_called_once()
    call_kwargs = mock_get.call_args
    assert "auth" in str(call_kwargs)


@pytest.mark.asyncio
async def test_list_notes_returns_error_string_on_api_error():
    """list_notes returns error string when FableAPIError is raised."""
    from fable_mcp.client import FableAPIError
    with patch("fable_mcp.tools.notes.get_client") as mock_gc:
        mock_gc.return_value.get = AsyncMock(side_effect=FableAPIError(404, "Not found"))
        from fable_mcp.tools.notes import list_notes
        result = await list_notes()
    assert "404" in result
    assert isinstance(result, str)


@pytest.mark.asyncio
async def test_create_note_sends_body():
    """create_note posts correct body to /api/notes."""
    mock_post = AsyncMock(return_value={"id": 1, "title": "Test"})
    with patch("fable_mcp.tools.notes.get_client") as mock_gc:
        mock_gc.return_value.post = mock_post
        from fable_mcp.tools.notes import create_note
        await create_note(title="Test", body="Hello")
    mock_post.assert_called_once()
    body = mock_post.call_args[0][1]
    assert body["title"] == "Test"
    assert body["body"] == "Hello"

• Step 2: Run to verify failure

cd fable-mcp && pytest tests/test_tools_notes.py -v

Expected: ImportError

• Step 3: Write the notes tools

Create fable-mcp/fable_mcp/tools/notes.py:

"""MCP tools for Fable notes."""
from typing import Any

from fable_mcp.client import FableAPIError, get_client


async def list_notes(
    query: str | None = None,
    tags: list[str] | None = None,
    project_id: int | None = None,
    limit: int = 20,
    offset: int = 0,
) -> Any:
    """List notes. Optionally filter by text query, tags, or project."""
    try:
        params: dict = {"limit": limit, "offset": offset, "is_task": "false"}
        if query:
            params["q"] = query
        if tags:
            params["tag"] = tags
        if project_id:
            params["project_id"] = project_id
        return await get_client().get("/api/notes", **params)
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"


async def get_note(note_id: int) -> Any:
    """Get a single note by ID, including its full body."""
    try:
        return await get_client().get(f"/api/notes/{note_id}")
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"


async def create_note(
    title: str,
    body: str = "",
    tags: list[str] | None = None,
    project_id: int | None = None,
) -> Any:
    """Create a new note."""
    try:
        payload: dict = {"title": title, "body": body}
        if tags:
            payload["tags"] = tags
        if project_id:
            payload["project_id"] = project_id
        return await get_client().post("/api/notes", payload)
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"


async def update_note(
    note_id: int,
    title: str | None = None,
    body: str | None = None,
    tags: list[str] | None = None,
    project_id: int | None = None,
) -> Any:
    """Update an existing note. Only provided fields are changed."""
    try:
        payload: dict = {}
        if title is not None:
            payload["title"] = title
        if body is not None:
            payload["body"] = body
        if tags is not None:
            payload["tags"] = tags
        if project_id is not None:
            payload["project_id"] = project_id
        return await get_client().patch(f"/api/notes/{note_id}", payload)
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"


async def delete_note(note_id: int) -> Any:
    """Delete a note by ID."""
    try:
        await get_client().delete(f"/api/notes/{note_id}")
        return {"status": "deleted", "id": note_id}
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"

• Step 4: Run tests to verify pass

cd fable-mcp && pytest tests/test_tools_notes.py -v

• Step 5: Commit

git add fable-mcp/fable_mcp/tools/notes.py fable-mcp/tests/test_tools_notes.py
git commit -m "feat: add notes MCP tools (list/get/create/update/delete)"

---

Task 11: Tasks tools

Files:

• Create: fable-mcp/fable_mcp/tools/tasks.py

Tasks follow the exact same pattern as notes.py but target /api/tasks and include patch_task_status and add_task_log. Write the module following the same error-handling wrapper pattern as notes.py.

• Step 1: Write failing tests first

Create fable-mcp/tests/test_tools_tasks.py:

"""Unit tests for tasks tools."""
import pytest
from unittest.mock import AsyncMock, patch
from fable_mcp.client import FableAPIError


@pytest.mark.asyncio
async def test_list_tasks_passes_params():
    mock_get = AsyncMock(return_value={"notes": [], "total": 0})
    with patch("fable_mcp.tools.tasks.get_client") as mock_gc:
        mock_gc.return_value.get = mock_get
        from fable_mcp.tools.tasks import list_tasks
        await list_tasks(project_id=1, status="open")
    mock_get.assert_called_once()


@pytest.mark.asyncio
async def test_list_tasks_returns_error_on_api_error():
    with patch("fable_mcp.tools.tasks.get_client") as mock_gc:
        mock_gc.return_value.get = AsyncMock(side_effect=FableAPIError(500, "server error"))
        from fable_mcp.tools.tasks import list_tasks
        result = await list_tasks()
    assert "500" in result


@pytest.mark.asyncio
async def test_patch_task_status_posts_to_correct_path():
    mock_patch = AsyncMock(return_value={"id": 1, "status": "done"})
    with patch("fable_mcp.tools.tasks.get_client") as mock_gc:
        mock_gc.return_value.patch = mock_patch
        from fable_mcp.tools.tasks import patch_task_status
        await patch_task_status(task_id=5, status="done")
    call_args = mock_patch.call_args[0]
    assert "/api/tasks/5/status" in call_args[0]


@pytest.mark.asyncio
async def test_add_task_log_posts_to_correct_path():
    mock_post = AsyncMock(return_value={"id": 1})
    with patch("fable_mcp.tools.tasks.get_client") as mock_gc:
        mock_gc.return_value.post = mock_post
        from fable_mcp.tools.tasks import add_task_log
        await add_task_log(task_id=3, content="Did the thing")
    call_args = mock_post.call_args[0]
    assert "/api/tasks/3/logs" in call_args[0]

Run to verify failure:

cd fable-mcp && pytest tests/test_tools_tasks.py -v

Expected: ImportError

• Step 2: Write the module

Create fable-mcp/fable_mcp/tools/tasks.py with these functions (follow the notes.py pattern exactly — try/except FableAPIError returning str):

• list_tasks(query, project_id, milestone_id, status, priority, limit, offset) → GET /api/tasks with matching query params

• get_task(task_id) → GET /api/tasks/{task_id}

• create_task(title, body, project_id, milestone_id, priority, status) → POST /api/tasks

• update_task(task_id, title, body, project_id, milestone_id, priority, status) → PUT /api/tasks/{task_id} (Fable tasks use PUT not PATCH)

• delete_task(task_id) → DELETE /api/tasks/{task_id}

• patch_task_status(task_id, status) → PATCH /api/tasks/{task_id}/status with {"status": status}

• add_task_log(task_id, content) → POST /api/tasks/{task_id}/logs with {"content": content} (task_id goes in the URL path, confirmed in routes/task_logs.py which mounts under /api/tasks)

• Step 3: Run tests to verify pass

cd fable-mcp && pytest tests/test_tools_tasks.py tests/ -v

• Step 4: Commit

git add fable-mcp/fable_mcp/tools/tasks.py fable-mcp/tests/test_tools_tasks.py
git commit -m "feat: add tasks MCP tools (list/get/create/update/delete/status/log)"

---

Task 12: Projects tools

Files:

• Create: fable-mcp/fable_mcp/tools/projects.py

• Step 1: Write failing tests first

Create fable-mcp/tests/test_tools_projects.py following the same pattern as test_tools_notes.py — at minimum:

• test_list_projects_calls_get — mock get_client().get, assert it calls /api/projects
• test_list_projects_returns_error_on_api_error — assert error string returned on FableAPIError

Run to verify failure: cd fable-mcp && pytest tests/test_tools_projects.py -v

• Step 2: Write the module

Create fable-mcp/fable_mcp/tools/projects.py with:

• list_projects() → GET /api/projects

• get_project(project_id) → GET /api/projects/{project_id} (includes milestone summary)

• create_project(title, description, goal, color) → POST /api/projects

• update_project(project_id, title, description, goal, status, color) → PATCH /api/projects/{project_id}

• delete_project(project_id) → DELETE /api/projects/{project_id}

• Step 3: Run tests to verify pass

cd fable-mcp && pytest tests/test_tools_projects.py tests/ -v

• Step 4: Commit

git add fable-mcp/fable_mcp/tools/projects.py fable-mcp/tests/test_tools_projects.py
git commit -m "feat: add projects MCP tools"

---

Task 13: Milestones tools

Files:

• Create: fable-mcp/fable_mcp/tools/milestones.py

• Step 1: Write failing tests first

Create fable-mcp/tests/test_tools_milestones.py following the same pattern as test_tools_notes.py — at minimum:

• test_list_milestones_calls_correct_path — mock get_client().get, assert path contains /api/projects/1/milestones
• test_list_milestones_returns_error_on_api_error — assert error string returned

Run to verify failure: cd fable-mcp && pytest tests/test_tools_milestones.py -v

• Step 2: Write the module

Create fable-mcp/fable_mcp/tools/milestones.py with:

• list_milestones(project_id, status) → GET /api/projects/{project_id}/milestones

• get_milestone(project_id, milestone_id) → GET /api/projects/{project_id}/milestones/{milestone_id}

• create_milestone(project_id, title, description, order_index) → POST /api/projects/{project_id}/milestones

• update_milestone(project_id, milestone_id, title, description, status, order_index) → PATCH /api/projects/{project_id}/milestones/{milestone_id}

• delete_milestone(project_id, milestone_id) → DELETE /api/projects/{project_id}/milestones/{milestone_id}

• Step 3: Run tests to verify pass

cd fable-mcp && pytest tests/test_tools_milestones.py tests/ -v

• Step 4: Commit

git add fable-mcp/fable_mcp/tools/milestones.py fable-mcp/tests/test_tools_milestones.py
git commit -m "feat: add milestones MCP tools"

---

Task 14: Search tool

Files:

• Create: fable-mcp/fable_mcp/tools/search.py

• Step 1: Write the module

Create fable-mcp/fable_mcp/tools/search.py:

"""MCP tool for semantic search across Fable notes and tasks."""
from typing import Any

from fable_mcp.client import FableAPIError, get_client


async def semantic_search(
    query: str,
    content_type: str = "all",
    limit: int = 10,
) -> Any:
    """
    Search notes and/or tasks by meaning using semantic similarity.

    content_type: "note", "task", or "all" (default).
    Returns ranked results with similarity scores.
    """
    try:
        return await get_client().get(
            "/api/search",
            q=query,
            content_type=content_type,
            limit=limit,
        )
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"

• Step 2: Commit

git add fable-mcp/fable_mcp/tools/search.py
git commit -m "feat: add semantic_search MCP tool"

---

Task 15: Chat tool (SSE consumer)

Files:

• Create: fable-mcp/fable_mcp/tools/chat.py

• Create: fable-mcp/tests/test_tools_chat.py

• Step 1: Write the failing tests

Create fable-mcp/tests/test_tools_chat.py:

"""Unit tests for chat tool — SSE parsing logic."""
import json
import pytest
from unittest.mock import AsyncMock, patch, MagicMock


def test_parse_sse_done_event():
    """_parse_sse_events extracts response and tools_used from done event."""
    from fable_mcp.tools.chat import _parse_done_event
    raw = json.dumps({"type": "done", "content": "Hello world", "tools_used": ["create_task"]})
    result = _parse_done_event(raw)
    assert result["response"] == "Hello world"
    assert result["tools_used"] == ["create_task"]


def test_parse_sse_non_done_returns_none():
    """_parse_done_event returns None for non-done events."""
    from fable_mcp.tools.chat import _parse_done_event
    raw = json.dumps({"type": "token", "content": "partial"})
    assert _parse_done_event(raw) is None


@pytest.mark.asyncio
async def test_send_message_creates_conversation_if_none():
    """send_message creates a new conversation when none exists."""
    mock_post = AsyncMock(side_effect=[
        {"id": 42, "title": "Test", "conversation_type": "mcp"},  # create conv
        {"id": 1},  # post message
    ])
    mock_list = AsyncMock(return_value={"conversations": []})

    async def mock_stream(path):
        yield json.dumps({"type": "done", "content": "OK", "tools_used": []})

    with patch("fable_mcp.tools.chat.get_client") as mock_gc:
        client = MagicMock()
        client.get = mock_list
        client.post = mock_post
        client.stream_get = mock_stream
        mock_gc.return_value = client

        from fable_mcp.tools.chat import send_message
        result = await send_message("Hello", conversation_name="Test")

    assert result["response"] == "OK"
    assert result["conversation_id"] == 42

• Step 2: Run to verify failure

cd fable-mcp && pytest tests/test_tools_chat.py -v

• Step 3: Write the chat tool

Create fable-mcp/fable_mcp/tools/chat.py:

"""MCP tool for sending messages to the Fable chat pipeline."""
import json
from typing import Any

from fable_mcp.client import FableAPIError, get_client


def _parse_done_event(raw_json: str) -> dict | None:
    """Parse a raw SSE data line. Returns result dict if type=='done', else None."""
    try:
        event = json.loads(raw_json)
        if event.get("type") == "done":
            return {
                "response": event.get("content", ""),
                "tools_used": event.get("tools_used", []),
            }
        return None
    except (json.JSONDecodeError, AttributeError):
        return None


async def _find_or_create_conversation(name: str) -> int:
    """Return conversation ID for the named MCP conversation, creating if needed."""
    client = get_client()
    # Search existing MCP conversations for matching title
    data = await client.get("/api/chat/conversations", type="mcp", limit=100)
    for conv in data.get("conversations", []):
        if conv.get("title") == name:
            return conv["id"]
    # Not found — create a new one
    conv = await client.post("/api/chat/conversations", {
        "title": name,
        "conversation_type": "mcp",
    })
    return conv["id"]


async def send_message(
    message: str,
    conversation_name: str | None = None,
) -> Any:
    """
    Send a message to Fable's LLM pipeline and return the response.

    Creates or continues a persistent MCP conversation in Fable.
    The conversation is visible in Fable for audit purposes but excluded
    from the normal chat list. Returns the assistant response and a list
    of any tools the Fable LLM fired during generation.
    """
    name = conversation_name or "MCP Session"
    try:
        client = get_client()
        conv_id = await _find_or_create_conversation(name)

        # Post the user message (this triggers generation)
        await client.post(
            f"/api/chat/conversations/{conv_id}/messages",
            {"content": message, "role": "user"},
        )

        # Consume SSE stream until done event
        result = None
        async for line in client.stream_get(f"/api/chat/conversations/{conv_id}/stream"):
            if not line.strip():
                continue
            parsed = _parse_done_event(line)
            if parsed is not None:
                result = parsed
                break

        if result is None:
            return {"error": "Stream ended without a done event", "conversation_id": conv_id}

        return {
            "response": result["response"],
            "tools_used": result["tools_used"],
            "conversation_id": conv_id,
        }
    except FableAPIError as e:
        return str(e)
    except Exception as e:
        return f"Error: {e}"

Note: verify the Fable message POST endpoint path by checking src/scribe/routes/chat.py — search for the route that accepts a user message and triggers generation. Adjust /api/chat/conversations/{conv_id}/messages if the actual path differs.

• Step 4: Run tests

cd fable-mcp && pytest tests/test_tools_chat.py -v

• Step 5: Commit

git add fable-mcp/fable_mcp/tools/chat.py fable-mcp/tests/test_tools_chat.py
git commit -m "feat: add chat send_message MCP tool with SSE consumer"

---

Task 16: server.py + MCP registration

Files:

• Create: fable-mcp/fable_mcp/server.py

• Step 1: Write server.py

Create fable-mcp/fable_mcp/server.py:

"""Fable MCP Server entry point."""
from mcp.server.fastmcp import FastMCP

from fable_mcp.client import init_client
from fable_mcp.tools import notes, tasks, projects, milestones, search, chat

mcp = FastMCP("Fable Assistant")

# --- Notes ---
mcp.tool()(notes.list_notes)
mcp.tool()(notes.get_note)
mcp.tool()(notes.create_note)
mcp.tool()(notes.update_note)
mcp.tool()(notes.delete_note)

# --- Tasks ---
mcp.tool()(tasks.list_tasks)
mcp.tool()(tasks.get_task)
mcp.tool()(tasks.create_task)
mcp.tool()(tasks.update_task)
mcp.tool()(tasks.delete_task)
mcp.tool()(tasks.patch_task_status)
mcp.tool()(tasks.add_task_log)

# --- Projects ---
mcp.tool()(projects.list_projects)
mcp.tool()(projects.get_project)
mcp.tool()(projects.create_project)
mcp.tool()(projects.update_project)
mcp.tool()(projects.delete_project)

# --- Milestones ---
mcp.tool()(milestones.list_milestones)
mcp.tool()(milestones.get_milestone)
mcp.tool()(milestones.create_milestone)
mcp.tool()(milestones.update_milestone)
mcp.tool()(milestones.delete_milestone)

# --- Search ---
mcp.tool()(search.semantic_search)

# --- Chat ---
mcp.tool()(chat.send_message)


def main():
    init_client()
    mcp.run()


if __name__ == "__main__":
    main()

• Step 2: Verify it imports cleanly

cd fable-mcp && python -c "from fable_mcp.server import mcp; print('OK')"

Expected: OK (no import errors; FableClient init error is OK here since no env vars set)

• Step 3: Run full test suite

cd fable-mcp && pytest tests/ -v

Expected: all tests pass

• Step 4: Commit

git add fable-mcp/fable_mcp/server.py
git commit -m "feat: add server.py with all MCP tools registered"

---

Task 17: End-to-end test + Claude Code registration

• Step 1: Install the package

pip install -e ./fable-mcp

• Step 2: Create a write-scoped API key

In the Fable Settings UI → API Keys, create a key named "Claude MCP" with write scope. Copy the key.

• Step 3: Test the server starts

FABLE_URL=http://localhost:8080 FABLE_API_KEY=fmcp_<yourkey> fable-mcp

Expected: server starts and waits on stdio (no output, no crash). Ctrl+C to stop.

• Step 4: Register with Claude Code

Add to ~/.claude/settings.json under "mcpServers":

"fable": {
  "command": "fable-mcp",
  "env": {
    "FABLE_URL": "http://localhost:8080",
    "FABLE_API_KEY": "fmcp_your_key_here"
  }
}

• Step 5: Verify tools appear in Claude Code

In a new Claude Code session, run /mcp or check the tool list. You should see list_notes, create_task, semantic_search, send_message, etc.

• Step 6: Smoke test each tool group

Ask Claude: "Use Fable to list my projects." Verify a real response comes back.

• Step 7: Final commit

git add fable-mcp/
git commit -m "feat: fable-mcp v0.1.0 — complete MCP server with all tool groups"

---

Future Tasks (post v0.1.0)

Future Task A: In-app MCP install instructions (PyPI/Forgejo registry)

Once fable-mcp is published to a package registry (PyPI or the Forgejo package registry at git.fabledsword.com), add an install instructions panel to the API Keys tab in Settings. This replaces manual pip install -e with a one-liner and links the version shown to the package registry.

Files:

• Modify: frontend/src/views/SettingsView.vue — add "Install" section above the create form showing:
  • pip install fable-mcp (or registry-specific command)
  • Current published version badge (fetched from registry API)
  • Link to full documentation/README

Depends on: fable-mcp extracted to its own Forgejo repo and published as a package.

---

Future Task B: Forgejo MCP server

A second MCP server (forgejo-mcp/) that gives Claude direct access to the Forgejo instance at git.fabledsword.com for CI/CD automation. Capabilities: list/create/merge PRs, trigger CI runs, read build logs, manage releases, push config files.

Scope: Separate spec + plan session. Key questions to resolve in brainstorming:

• Authentication: Forgejo personal access token vs. OAuth app
• Which Gitea/Forgejo API endpoints to expose as tools
• Whether to use an existing gitea-mcp or build from scratch (check Forgejo's MCP compatibility)
• CI automation: trigger builds, read job logs, manage runner labels

Prerequisite: fable-mcp v0.1.0 working and registered in Claude Code.