FabledScribe/src/fabledassistant/mcp/server.py

"""FastMCP instance + Quart mount-point. Tools are registered in mcp/tools/."""
from __future__ import annotations

from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
from quart import Quart

_INSTRUCTIONS = """
Scribe is the user's self-hosted second-brain and project-management data
store. You (Claude) are the assistant.

Hierarchy: Project -> Milestone -> Task/Note.

What each part is for, and when to reach for it:
- Project: the top-level container for a body of work.
- Milestone: groups related tasks within a project toward a goal (status
  active/done). Use one when a chunk of work needs its own arc.
- Task: a unit of actionable work with a lifecycle (status
  todo/in_progress/done/cancelled, optional priority). A task is a note with a
  status — reach for one when there is something to DO. Record progress over
  time with work-logs (add_task_log) rather than rewriting the body.
- Plan: a task with kind=plan — HOW you'll execute a chunk of work. The body
  holds the design + step checklist; work-logs record progress. Start one with
  start_planning when beginning non-trivial work, before writing code.
- Note: durable free-form knowledge — reference material, decisions, dev-logs.
  No lifecycle, not actionable. Reach for one to CAPTURE something worth keeping.
- Typed entities (person/place/list): structured records about people, places,
  and checklists.

Mechanics:
- Notes and Tasks share a model; tasks are notes with is_task=True.
- Use the *_note tools for notes, the *_task tools for tasks. Don't mix them.
- Typed entities (person, place, list) are notes with a non-default note_type
  plus type-specific columns; use the dedicated *_person / *_place / *_list
  tools rather than create_note.
- Tags are plain strings (no `#` prefix). Empty list clears tags; omit to leave
  unchanged on updates.
- For optional integer FKs (project_id, milestone_id, parent_id), use 0 to mean
  "not set".

Scribe maintains a Rulebook system (Rulebook -> Topic -> Rule). Rules carry
an actionable statement plus optional Why and How-to-apply context. At the
start of any session that touches Scribe, call list_always_on_rules() to
load the standing rules — treat them as binding. When you also have a project
in scope, get_project(id) returns applicable_rules (rules from rulebooks the
project subscribes to) and subscribed_rulebooks; consult those too. Full text
(Why / How-to-apply) is available via get_rule(id).

Engineering and workflow rules live in Scribe. When you notice a pattern
worth codifying, call create_rule (cross-project, lands in a rulebook+topic)
or create_project_rule (one project only, no rulebook ceremony). Do NOT add
new engineering rules to CLAUDE.md or to ~/.claude/.../memory/feedback_*.md
— those stores are reserved for facts about the user (preferences, role,
communication style) and codebase onboarding pointers, respectively. Before
creating a rule, call list_always_on_rules and list_rules(project_id=...) to
avoid duplicates. Coordinate with the operator on whether a new rule belongs
in a project, an existing rulebook+topic, or a new rulebook.

When you are working on a specific project, call enter_project(project_id)
ONCE at session start (or whenever the active project changes). It returns the
project, its applicable_rules + project_rules + subscribed_rulebooks, milestone
summary, open tasks, and recent notes — everything you need to know the lay of
the land before mutating. Don't call get_project + get_applicable_rules + a
search separately when enter_project already composes them.

Plans are tasks with kind=plan, and Scribe is the canonical home for them.
When you begin non-trivial work, call start_planning(project_id, title) FIRST —
before any brainstorming, design, or plan-writing skill runs. start_planning
seeds the plan body, returns the project's applicable_rules, and gives you the
task id you'll write into. If a skill or habit tells you to save a plan or spec
to `docs/superpowers/plans/*.md` or `docs/superpowers/specs/*.md`, that path is
superseded here: put the spec/plan content in the kind=plan task's body via
update_task, and record progress with add_task_log. Local .md files are not
the record — the task is.

Deletes are recoverable: every delete_* tool moves the entity (and its
descendants) to the trash and returns a deleted_batch_id. Use list_trash() to
see trashed batches, restore(deleted_batch_id) to undo a deletion, and
purge_trash(deleted_batch_id, confirmed=True) for a permanent delete. Trash
auto-purges after the operator's retention window.
"""


def build_mcp_server() -> FastMCP:
    """Build the FastMCP instance with all tools registered.

    DNS-rebinding protection is disabled: FastMCP's default allow-list is
    just localhost variants, which means any deployment behind a reverse
    proxy (Traefik with a hostname like devassistant.traefik.internal,
    Cloudflare, nginx, etc.) gets 421 Misdirected Request. The threat
    model that protection addresses — a malicious browser page rebinding
    DNS to hit a localhost MCP — doesn't apply here: this is HTTP transport
    behind a reverse proxy with bearer-token auth as the real security
    boundary.
    """
    # stateless_http=True: don't hand the client a persistent Mcp-Session-Id.
    # The stateful default strands Claude Code after a container redeploy —
    # it reconnects with the now-unknown session id, the server returns 404,
    # and the client won't re-initialize on a 404 (Claude Code issue #60949),
    # so the connection stays dead until a manual /mcp retry. Stateless makes
    # every request self-contained (bearer-auth only), so a post-deploy
    # reconnect just works. Trade-off: no server-pushed list_changed stream,
    # which we don't use — tools are re-fetched on reconnect anyway.
    mcp = FastMCP(
        "scribe",
        instructions=_INSTRUCTIONS.strip(),
        stateless_http=True,
        transport_security=TransportSecuritySettings(
            enable_dns_rebinding_protection=False,
        ),
    )
    from fabledassistant.mcp.tools import register_all
    register_all(mcp)
    return mcp


def mount_mcp(app: Quart) -> None:
    """Mount the FastMCP streamable-HTTP ASGI sub-app at /mcp on the Quart app.

    A small ASGI middleware between Quart and the FastMCP sub-app validates the
    Bearer token against the api_keys table. Authenticated requests have their
    user_id attached to the ASGI scope under "scribe_user_id" for tool handlers
    to read.

    FastMCP's streamable_http session manager owns a task group that must be
    running before it can serve requests. In a stand-alone Starlette deployment
    that would happen via the Starlette `lifespan` parameter. Since we're hosted
    inside Quart, we hook the session manager's `run()` async context manager
    into Quart's serving lifecycle (before_serving / after_serving).
    """
    from fabledassistant.mcp.auth import resolve_bearer_to_user_id

    mcp = build_mcp_server()
    mcp_asgi = mcp.streamable_http_app()
    app.mcp_instance = mcp

    @app.before_serving
    async def _start_mcp_session() -> None:
        cm = mcp.session_manager.run()
        await cm.__aenter__()
        app._mcp_session_cm = cm

    @app.after_serving
    async def _stop_mcp_session() -> None:
        cm = getattr(app, "_mcp_session_cm", None)
        if cm is not None:
            await cm.__aexit__(None, None, None)

    async def auth_wrapped(scope, receive, send):
        if scope["type"] != "http":
            return await mcp_asgi(scope, receive, send)
        # ASGI headers are lowercase bytes per spec; lowercase explicitly to be safe.
        headers = {k.decode().lower(): v.decode() for k, v in scope.get("headers", [])}
        user_id = await resolve_bearer_to_user_id(headers.get("authorization"))
        if user_id is None:
            await send({
                "type": "http.response.start",
                "status": 401,
                "headers": [
                    (b"content-type", b"application/json"),
                    (b"www-authenticate", b'Bearer realm="scribe-mcp"'),
                ],
            })
            await send({
                "type": "http.response.body",
                "body": b'{"error":"unauthorized"}',
            })
            return
        scope["scribe_user_id"] = user_id
        from fabledassistant.mcp._context import _user_id_ctx
        token = _user_id_ctx.set(user_id)
        try:
            await mcp_asgi(scope, receive, send)
        finally:
            _user_id_ctx.reset(token)

    original_asgi = app.asgi_app

    async def dispatch(scope, receive, send):
        if scope["type"] == "http":
            path = scope.get("path", "")
            if path == "/mcp" or path.startswith("/mcp/"):
                # Don't rewrite the path: FastMCP's streamable_http_app mounts
                # its handler at /mcp by default. If we strip the prefix to "/",
                # FastMCP's internal routing returns 404 because there's no
                # handler at "/" — only at "/mcp". Pass the scope through
                # untouched and let FastMCP's own routing match.
                return await auth_wrapped(scope, receive, send)
        return await original_asgi(scope, receive, send)

    app.asgi_app = dispatch