server.py

"""wiki-mcp — MCP server wrapping SQLite FTS5 wiki index.

Exposes 5 tools:
  • search        — fast BM25 search across markdown vault
  • get_note      — fetch full note by path
  • backlinks     — what links to this note
  • list_tags     — all tags w/ counts
  • reindex       — force rebuild

Run:
  WIKI_PATH=/tmp/wiki python3 server.py
or via Claude Code:
  claude mcp add wiki -- python3 /root/wiki-mcp/server.py
"""
from __future__ import annotations
import os
import sys
import threading
from pathlib import Path
from mcp.server.fastmcp import FastMCP

# Force wiki_index to use our chosen vault before importing
WIKI_PATH = os.environ.get("WIKI_PATH") or os.environ.get("WIKI_DIR", "/tmp/wiki")
DB_PATH = os.environ.get("WIKI_DB") or os.environ.get("WIKI_INDEX_DB", "/root/wiki-mcp/wiki.db")
WIKI_AUTHOR = os.environ.get("WIKI_AUTHOR", "")  # Team: set per user for attribution
WIKI_TRANSPORT = os.environ.get("WIKI_TRANSPORT", "stdio")  # "stdio" or "http"
WIKI_PORT = int(os.environ.get("WIKI_PORT", "8787"))
# wiki_index reads these specific names:
os.environ["WIKI_DIR"] = WIKI_PATH
os.environ["WIKI_INDEX_DB"] = DB_PATH

sys.path.insert(0, str(Path(__file__).parent))
import wiki_index  # noqa: E402
import wiki_writer  # noqa: E402

mcp = FastMCP(
    "wiki-mcp",
    instructions=f"""Wiki search server backed by SQLite FTS5 + BM25 ranking.
Vault: {WIKI_PATH}

ALWAYS prefer `search` over reading whole files — returns ranked snippets
in ~10ms with 20-50x fewer tokens than full reads.

Search syntax:
  - free text → AND-prefix match: "auth bypass" → matches docs w/ both
  - "quoted phrase" → exact phrase
  - #tag → tag lookup (skips FTS)
  - prefix: → restrict to a folder

Workflow:
  1. search(query) for ranked snippets
  2. If a hit looks promising, get_note(path) for full text
  3. backlinks(path) to see what references it
""".strip(),
)

_conn = None
_lock = threading.RLock()


def _get_conn():
    global _conn
    if _conn is None:
        with _lock:
            if _conn is None:
                _conn = wiki_index._connect()
                wiki_index.init_db(_conn)
    return _conn


@mcp.tool()
def search(query: str, limit: int = 5) -> dict:
    """Fast ranked search across the wiki.

    Returns top-N matches w/ BM25 score and 24-word highlighted snippet.
    Use this BEFORE reading full files.

    Args:
        query: free text, "quoted phrase", or #tag
        limit: max results (default 5, max 20)
    """
    limit = max(1, min(int(limit), 20))
    with _lock:
        hits = wiki_index.fts_search(_get_conn(), query, limit)
    # Return compact form
    return {
        "query": query,
        "count": len(hits),
        "hits": [
            {
                "path": h.get("path"),
                "title": h.get("title"),
                "snippet": h.get("ctx", "").replace("<mark>", "**").replace("</mark>", "**"),
                "rank": round(h.get("rank", 0.0), 3),
            }
            for h in hits
        ],
    }


@mcp.tool()
def get_note(path: str) -> dict:
    """Fetch full body of a note by relative path (e.g. 'Bug-Bounty/_Index.md')."""
    full = Path(WIKI_PATH) / path
    if not full.exists() or not full.is_file():
        return {"error": f"not found: {path}"}
    try:
        body = full.read_text(encoding="utf-8")
    except Exception as e:
        return {"error": str(e)}
    return {"path": path, "size": len(body), "body": body}


@mcp.tool()
def backlinks(path: str) -> dict:
    """Notes that link to the given note (via [[wikilink]] syntax)."""
    with _lock:
        links = wiki_index.backlinks(_get_conn(), path)
    return {"path": path, "count": len(links), "linked_from": links}


@mcp.tool()
def list_tags() -> dict:
    """All #tags in the vault w/ usage counts."""
    with _lock:
        tags = wiki_index.list_tags(_get_conn())
    return {"count": len(tags), "tags": tags}


@mcp.tool()
def reindex(full: bool = False) -> dict:
    """Rebuild the FTS index. Set full=true to drop+rebuild from scratch."""
    with _lock:
        result = wiki_index.reindex(_get_conn(), full=bool(full))
    return result


@mcp.tool()
def stats() -> dict:
    """Index health: note count, db size, last reindex time."""
    with _lock:
        return wiki_index.stats(_get_conn())


@mcp.tool()
def lint_note(body: str) -> dict:
    """Validate a markdown body against the wiki schema WITHOUT writing.

    Returns errors list (empty = passes). Use this BEFORE write_note when
    drafting — cheaper than write+rollback.

    Checks:
      • required frontmatter (title, created, updated, type, tags)
      • type ∈ {entity, concept, comparison, query, runbook, decision, journal}
      • tags ∈ taxonomy (auto-loaded from SCHEMA.md)
      • date format YYYY-MM-DD
      • ≥1 [[wikilink]] outbound
      • body ≤ 150 lines
      • H1 present
    """
    errors = wiki_writer.lint(body, vault_dir=WIKI_PATH)
    return {"ok": not errors, "errors": errors}


@mcp.tool()
def frontmatter_template(type: str) -> dict:
    """Get a starter frontmatter blob + skeleton body for a given note type.

    Use this BEFORE drafting a note — gives you the shape, suggested folder,
    suggested tags (filtered to taxonomy), and skeleton sections. Avoids
    write_note rejections from missing frontmatter.

    Args:
        type: one of entity|concept|comparison|query|runbook|decision|journal

    Returns:
        purpose, frontmatter dict, suggested_folder, skeleton_body, schema_rules
    """
    return wiki_writer.frontmatter_template(type, vault_dir=WIKI_PATH)


@mcp.tool()
def taxonomy() -> dict:
    """Return the current tag taxonomy (parsed from SCHEMA.md) plus valid types.

    Use this BEFORE picking tags for a new note — only listed tags will pass lint.
    Adding a new tag requires editing SCHEMA.md first.
    """
    tags = sorted(wiki_writer.load_taxonomy(WIKI_PATH))
    return {
        "vault": WIKI_PATH,
        "schema_file": "SCHEMA.md",
        "tag_count": len(tags),
        "tags": tags,
        "valid_types": sorted(wiki_writer.VALID_TYPES),
        "to_add_tag": "edit SCHEMA.md taxonomy section, then re-call this tool to refresh cache",
    }


@mcp.tool()
def write_note(
    folder: str,
    title: str,
    body: str,
    type: str,
    tags: list,
    sources: list = None,
    auto_link: list = None,
    overwrite: bool = False,
) -> dict:
    """Create a new wiki note with enforced schema.

    Filename is auto-slugified from title. Frontmatter is auto-built.
    Lints before write — raises if schema fails.

    Args:
        folder: relative folder under vault (e.g. 'Bug-Bounty', 'entities')
        title: human title (becomes H1 + frontmatter)
        body: markdown body. Will get H1 prepended if missing.
        type: one of entity|concept|comparison|query|runbook|decision|journal
        tags: list of tag strings (must be in SCHEMA.md taxonomy)
        sources: optional list of source URLs/paths
        auto_link: if body has no [[wikilinks]], list of titles to add as Related
        overwrite: bypass dedupe check
    """
    try:
        result = wiki_writer.write_note(
            vault_dir=WIKI_PATH,
            folder=folder,
            title=title,
            body=body,
            type_=type,
            tags=tags or [],
            sources=sources,
            auto_link=auto_link,
            overwrite=overwrite,
            author=WIKI_AUTHOR or None,
        )
        # Auto-reindex incrementally
        with _lock:
            wiki_index.reindex(_get_conn(), full=False)
        return {"ok": True, **result}
    except wiki_writer.WriteError as e:
        return {"ok": False, "error": str(e)}


@mcp.tool()
def update_note(
    path: str,
    body: str = None,
    add_tags: list = None,
    bump_updated: bool = True,
) -> dict:
    """Patch an existing note. Lints before write. Bumps `updated:` field.

    Args:
        path: relative path under vault (e.g. 'Bug-Bounty/jenkins.md')
        body: replacement body (keeps frontmatter; H1 re-added if missing)
        add_tags: tags to merge into existing tags
        bump_updated: refresh `updated:` to today (default True)
    """
    try:
        result = wiki_writer.update_note(
            vault_dir=WIKI_PATH,
            path=path,
            body=body,
            add_tags=add_tags,
            bump_updated=bump_updated,
        )
        with _lock:
            wiki_index.reindex(_get_conn(), full=False)
        return {"ok": True, **result}
    except wiki_writer.WriteError as e:
        return {"ok": False, "error": str(e)}


@mcp.tool()
def append_section(path: str, section_title: str, content: str) -> dict:
    """Append a new `## section` to an existing note. Bumps `updated:`.

    Args:
        path: relative note path
        section_title: H2 heading text (without `## `)
        content: markdown content for the section
    """
    try:
        result = wiki_writer.append_section(
            vault_dir=WIKI_PATH,
            path=path,
            section_title=section_title,
            content=content,
        )
        with _lock:
            wiki_index.reindex(_get_conn(), full=False)
        return {"ok": True, **result}
    except wiki_writer.WriteError as e:
        return {"ok": False, "error": str(e)}


@mcp.tool()
def stubs(limit: int = 20) -> dict:
    """Find [[wikilinks]] pointing to non-existent notes — knowledge gaps.

    High reference counts = important missing knowledge. Create these notes
    to fill gaps. Great for team sprint planning: "what do we need to document?"

    Args:
        limit: max stubs to return (default 20)
    """
    with _lock:
        hits = wiki_index.stubs(_get_conn(), limit=limit)
    return {"count": len(hits), "stubs": hits}


@mcp.tool()
def recent(days: int = 7, limit: int = 20) -> dict:
    """Notes modified in the last N days, newest first.

    Use for standup reviews, change tracking, and team activity feeds.

    Args:
        days: lookback window (default 7)
        limit: max results (default 20)
    """
    with _lock:
        hits = wiki_index.recent(_get_conn(), days=days, limit=limit)
    return {"days": days, "count": len(hits), "notes": hits}


@mcp.tool()
def orphans(limit: int = 30) -> dict:
    """Notes with zero incoming links — potential orphans.

    Orphan notes aren't discoverable via graph traversal. Either link them
    from related notes or consider merging/removing them.

    Args:
        limit: max results (default 30)
    """
    with _lock:
        hits = wiki_index.orphans(_get_conn(), limit=limit)
    return {"count": len(hits), "orphans": hits}


@mcp.tool()
def suggest_split(path: str) -> dict:
    """Analyze an oversized note (>150 lines) and suggest split points.

    Returns H2 section boundaries with proposed filenames for each split.
    Preserves the parent note as a hub linking to children.

    Args:
        path: relative path to note
    """
    return wiki_writer.suggest_split(vault_dir=WIKI_PATH, path=path)


@mcp.tool()
def health() -> dict:
    """Comprehensive vault health report for team dashboards.

    Returns: schema compliance %, oversized notes, type distribution,
    author coverage, tag drift (used but not in taxonomy), timing.
    Use for weekly team reviews and quality monitoring.
    """
    with _lock:
        return wiki_writer.health(vault_dir=WIKI_PATH, conn=_get_conn())


@mcp.tool()
def format_note(path: str, dry_run: bool = True) -> dict:
    """Auto-fix a note's frontmatter to pass schema validation.

    Fixes missing title (from filename), created (from mtime), updated,
    type (mapped from non-standard values), H1, and adds a [[wikilink]]
    if missing. Does NOT fix over-150-line or invalid tags.

    Args:
        path: relative path to note (e.g. 'Knowledge/my-note.md')
        dry_run: if True, return what would change without writing (default True)
    """
    result = wiki_writer.format_note(vault_dir=WIKI_PATH, path=path, dry_run=dry_run)
    if not dry_run and result.get("written"):
        with _lock:
            wiki_index.reindex(_get_conn(), full=False)
    return result


@mcp.tool()
def format_vault(dry_run: bool = True) -> dict:
    """Scan entire vault and auto-fix notes that fail schema validation.

    Returns summary of clean/fixed/unfixable notes. Set dry_run=false
    to apply fixes. Safe: only touches frontmatter + adds missing H1/wikilinks.

    Args:
        dry_run: if True, report only without writing (default True)
    """
    result = wiki_writer.format_vault(vault_dir=WIKI_PATH, dry_run=dry_run)
    if not dry_run:
        with _lock:
            wiki_index.reindex(_get_conn(), full=True)
    return result


if __name__ == "__main__":
    # Initial index build
    with _lock:
        wiki_index.reindex(_get_conn(), full=True)
    if WIKI_TRANSPORT == "http":
        # Team deployment: HTTP transport for shared wiki server
        mcp.run(transport="streamable-http", host="0.0.0.0", port=WIKI_PORT)
    else:
        mcp.run()  # stdio transport by default