chemenu/tools/chemenu/mcp/server.py

"""The MCP read server: `search`, `types`, `lint` and `status` over `kb/`.

Chemenu's second consumer. The CLI and this are two adapters over one core -
`chemenu.api.Corpus` - so a question answered here and the same question asked
at a terminal go through the same code, and a golden test holds the two
outputs against each other rather than trusting that they agree.

**There is no write path, structurally.** Nothing under `chemenu.commands` is
imported here or in `chemenu.api`, so `new`, `touch`, `xref`, `cite`,
`publish`, `migrate` and `version bump` are not reachable - the functions do
not exist in this process's reach, rather than being filtered out of a list. A
test asserts it by importing this module in a clean interpreter and looking at
`sys.modules`.

**Authentication and rate limiting are not here.** Both are Traefik middleware
in front of the process, per the operator's decision of 2026-09-01: a request
that is not cleanly authenticated does not reach Python at all. What *is* here
is the resource protection that middleware cannot give - the search timeout and
the frontmatter limits - because those exist against an authenticated consumer
damaging itself, which is a different problem from an unauthenticated one.

**The Iteration Budget Gate is deliberately absent.** It exists to stop an
agent *session* from iterating unnoticed over the state of the wiki, which is
why retrieval is exempt from it in the first place. A user who searches too
often is a resource problem - different instrument, different purpose - and
using the gate as a rate limiter would dilute it into one.

**Every response carries the commit it was computed from.** `chemenu.api`
stamps `commit` and `as_of`; a stale checkout otherwise answers confidently and
wrongly. Keeping the checkout current is a `git fetch && git reset --hard`
poll outside this process - see `instructions/mcp-read-server.md` - which needs
no inbound endpoint and no signature checking.
"""
from __future__ import annotations

import os
from pathlib import Path
from typing import Any, Optional

from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

from chemenu import config
from chemenu.api import Corpus
from chemenu.errors import ChemenuError

SERVER_NAME = "chemenu"

# Transports this server will start on. `stdio` is for developing and testing
# it without a network; `streamable-http` is what a deployed instance speaks,
# and the only one the Traefik middleware can sit in front of, because Traefik
# is an HTTP reverse proxy. `sse` is reachable through the SDK but not offered:
# it is the superseded remote transport, and building on it now only moves the
# migration later.
TRANSPORTS = ("stdio", "streamable-http")


class TraceWouldWriteIntoCorpus(RuntimeError):
    """Raised at startup when telemetry would land inside the served tree."""


def check_trace_destination(root: Path) -> None:
    """Refuse to start if a trace would be written into the corpus.

    Telemetry defaults to *on* and writes under `reports/telemetry/` in the
    repo. Today nothing on this path emits - the writer is wired into
    `cli.main()` and the two gates, none of which run here - so this is a guard
    against the future rather than a fix for the present. It is worth having
    anyway: the sync that keeps this checkout current is `git reset --hard`, so
    a trace written into the tree is both a per-request write into a directory
    something else is entitled to wipe, and a silent way for the server to
    dirty the tree its own cache keys on.

    Turn tracing off (`WIKI_TRACE=0`) or point it somewhere else
    (`WIKI_TRACE_DIR`). Refusing rather than correcting it: a server that
    quietly relocates the operator's telemetry is a surprise waiting in a log
    nobody reads.
    """
    if os.environ.get("WIKI_TRACE", "1") == "0":
        return
    destination = os.environ.get("WIKI_TRACE_DIR")
    if destination is None:
        raise TraceWouldWriteIntoCorpus(
            "Telemetry is on and would write into the served checkout "
            f"({root / 'reports' / 'telemetry'}). The sync that keeps this checkout "
            "current is `git reset --hard`, which is entitled to wipe that directory. "
            "Set WIKI_TRACE=0, or point WIKI_TRACE_DIR outside the corpus."
        )
    resolved = Path(destination).expanduser().resolve()
    try:
        resolved.relative_to(Path(root).resolve())
    except ValueError:
        return
    raise TraceWouldWriteIntoCorpus(
        f"WIKI_TRACE_DIR ({resolved}) is inside the served checkout ({root}). "
        "Point it outside, or set WIKI_TRACE=0."
    )


def build_server(
    root: Optional[Path | str] = None, check_trace: bool = True
) -> MCPServer:
    """Assemble the server over one corpus.

    `root` follows `config.resolve_root()` - argument, then `$CHEMENU_ROOT`,
    then the checkout the package lives in - so a deployment points at its
    corpus with one environment variable and no code.
    """
    corpus = Corpus(root)
    if check_trace:
        check_trace_destination(corpus.root)

    server = MCPServer(
        name=SERVER_NAME,
        instructions=(
            "Read access to a Chemenu wiki: compiled, sourced knowledge under kb/. "
            "Every answer carries the commit it was computed from ('commit') and "
            "when it was produced ('as_of'); a null commit means the served tree "
            "has uncommitted changes and the answer corresponds to no revision. "
            "This server is read-only - there is no tool that writes."
        ),
    )

    @server.tool(
        name="search",
        description=(
            "Find pages in kb/ by text, by frontmatter, or by both. Returns "
            "title, path, kind, summary and confidence per hit, so a result can "
            "be judged without fetching the page. Prefer this over listing "
            "files: the answer is a few hundred tokens instead of a whole index."
        ),
    )
    def search(
        query: str | None = None,
        predicates: list[str] | None = None,
        regex: bool = False,
        limit: int = 20,
        sort: str | None = None,
    ) -> dict[str, Any]:
        """Search the wiki.

        `predicates` are frontmatter filters in the CLI's own `--field` syntax,
        ANDed: `confidence<0.6`, `entity_type=system`, `tags~k8s`, `source_url:*`
        (present), `!source_url` (absent). With no `query` this is a pure
        structured query over frontmatter.

        `regex` applies the pattern with ripgrep's linear engine. It is off by
        default, so an accidental `.*` is a literal.
        """
        return _guard(
            lambda: corpus.search(
                text=query,
                predicates=predicates or (),
                regex=regex,
                limit=limit,
                sort=sort,
            )
        )

    @server.tool(
        name="types",
        description=(
            "List the page types this wiki declares - what kinds of page exist, "
            "where each lives, and what its schema is. Read this before "
            "interpreting a page's `kind`."
        ),
    )
    def types() -> dict[str, Any]:
        return _guard(corpus.types)

    @server.tool(
        name="describe_type",
        description=(
            "One page type's full contract: its frontmatter fields with "
            "required/optional and any enums, its subtype field, and its "
            "authoring guidance."
        ),
    )
    def describe_type(name: str) -> dict[str, Any]:
        """`name` is a short type name as listed by `types`, e.g. 'entity'."""
        return _guard(lambda: corpus.describe_type(name))

    @server.tool(
        name="lint",
        description=(
            "The wiki's structural health: broken wikilinks, orphan pages, "
            "index drift, schema gaps, provenance gaps. Findings only - the "
            "JSON form writes no report file."
        ),
    )
    def lint() -> dict[str, Any]:
        return _guard(corpus.lint)

    @server.tool(
        name="status",
        description=(
            "A snapshot: how many pages the wiki holds, how they split across "
            "collections, and how many findings of each kind lint reports. "
            "Cheaper to read than the full lint output."
        ),
    )
    def status() -> dict[str, Any]:
        return _guard(corpus.status)

    return server


def _guard(call):
    """Turn a `ChemenuError` into a plain message for the protocol layer.

    A bad predicate is the caller's argument, not a server fault, and it should
    arrive as a tool error the model can act on, carrying the message that says
    what to do differently. The SDK draws exactly this line: a `ToolError` is a
    deliberate refusal and its text reaches the caller, while anything else is a
    crash whose text stays on the server. Only `ChemenuError` is caught -
    everything else is a genuine fault and belongs in the log, unswallowed.
    """
    try:
        return call()
    except ChemenuError as exc:
        raise ToolError(str(exc)) from exc


def serve(
    transport: str = "stdio",
    root: Optional[Path | str] = None,
    **kwargs: Any,
) -> None:
    """Start the server. `transport` is one of `TRANSPORTS`."""
    if transport not in TRANSPORTS:
        raise ValueError(
            f"unknown transport {transport!r}. Available: {', '.join(TRANSPORTS)}"
        )
    build_server(root).run(transport=transport, **kwargs)