distinct_mcp/__init__.py

"""MCP servers for the tools people actually ask a small local model for.

Four servers, all in-process, none of them installed or approved by the act of
importing this package.

* :mod:`distinct_mcp.lists` publishes the per-run planning list as
  ``plan.todo@1``.
* :mod:`distinct_mcp.files` publishes reading, writing and listing inside one
  operator-nominated directory as ``files.read_file@1``,
  ``files.write_file@1`` and ``files.list_files@1``.
* :mod:`distinct_mcp.text` publishes ``text.count_text@1``,
  ``text.sort_list@1`` and ``text.check_json@1``.
* :mod:`distinct_mcp.numbers` publishes ``numbers.summarise_numbers@1``.

**In-process, and not by accident.** Every server here is an
:class:`~distinct_tools.mcp.InProcessMcpServer`. The reason is in that module's
docstring: a stdio MCP server performs its own I/O, so ``ToolContext`` cannot
govern it and ``required_hosts`` becomes a record of consent rather than a
control, leaving the OS-native isolation layer as the only thing containing it.
An in-process server has no such gap. Nothing in this package spawns a process,
opens a socket or leaves anything running when the agent exits. The file server
is the only one that touches a disk, and it is bounded to one directory a person
chose.

**Nothing here is a parallel install path.** Each builder returns a server that
goes through :func:`distinct_tools.mcp.install_mcp_server`, which is what
applies the tool cap, the description bounds and the digest pinning, and which
turns each tool into an ordinary :class:`~distinct_tools.core.ToolSpec` in an
ordinary registry behind the ordinary broker::

    registry = Registry()
    installed = install_mcp_server(registry, file_mcp_server(chosen), FILE_MANIFEST)
    persist(pin_manifest(FILE_MANIFEST, installed))   # so a later change is caught

**Installing is not approving, and the default approves nothing.** The
operator's :class:`~distinct_tools.approval.OperatorPolicy` is a separate,
second decision, and with nothing approved every one of these tools fails
closed: ``ToolBroker`` refuses to construct at all rather than quietly dropping
the tool from the run.

**How a worker gets these without an operator writing code.**
:func:`install_catalogue` installs exactly the refs an operator approved and
skips every server none of whose tools were named, so a worker that approved
``text.count_text@1`` never builds the file server at all.
``distinct_tools.default_registry`` calls it, which is the whole path from an
approved ref to a callable tool. The file server needs one more thing, a
directory, and it reads it from ``DISTINCT_FILE_ROOT``: named there rather than
guessed, because there is no folder this package could pick that somebody
consented to. Approved with no directory named, it is skipped and the reason
comes back in :attr:`CatalogueInstall.notes` rather than as a crash at
start-up.

**One trap worth naming, because it is easy to walk into.** The bundle name
``local`` in ``DISTINCT_APPROVED_TOOLS`` expands to "every installed tool that
declares no hosts", against whatever spec list ``load_policy`` was given. None
of these servers declares a host, so passing the MCP specs to ``load_policy``
makes ``local`` cover ``files.write_file@1`` as well. Load the policy against
``distinct_tools.INSTALLABLE_SPECS`` and name each MCP ref in full if that is
not what you meant. There is a test for both halves of this.

**What is not claimed.** No measurement in this repository says a sub-7B model
does better with these servers than without them, and one measurement recorded
in :mod:`distinct_tools.skills` points the other way for planning specifically.
Each module states its own argument and its own gaps; :mod:`distinct_mcp.files`
states, at length, what its containment cannot enforce.
"""

from __future__ import annotations

import os
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, replace

from distinct_tools.core import Registry, ToolRef
from distinct_tools.mcp import McpError, McpServerManifest, install_mcp_server

from .files import FILE_MANIFEST, FileVault, file_mcp_server
from .lists import LIST_MANIFEST, list_mcp_server
from .numbers import NUMBERS_MANIFEST, numbers_mcp_server
from .text import TEXT_MANIFEST, text_mcp_server

#: Where the file server's one directory is named. There is deliberately no
#: default. A default would be a directory nobody chose, and the whole argument
#: for this server is that a person picked the folder it may touch.
FILE_ROOT_ENV_VAR = "DISTINCT_FILE_ROOT"


@dataclass(frozen=True)
class CatalogueEntry:
    """One server an operator may choose to install, and what it costs them."""

    manifest: McpServerManifest
    summary: str
    tool_names: tuple[str, ...]
    #: True when the builder needs a directory from the operator. There is no
    #: default for it, so a console cannot offer this server as one click.
    requires_directory: bool = False


CATALOGUE: tuple[CatalogueEntry, ...] = (
    CatalogueEntry(
        manifest=LIST_MANIFEST,
        summary="A numbered list of what is left to do, held for one run and erased with it.",
        tool_names=("todo",),
    ),
    CatalogueEntry(
        manifest=FILE_MANIFEST,
        summary=(
            "Read, write and list text files inside one directory you nominate, and nowhere else."
        ),
        tool_names=("read_file", "write_file", "list_files"),
        requires_directory=True,
    ),
    CatalogueEntry(
        manifest=TEXT_MANIFEST,
        summary="Count text exactly, order a list, and check whether text is valid JSON.",
        tool_names=("count_text", "sort_list", "check_json"),
    ),
    CatalogueEntry(
        manifest=NUMBERS_MANIFEST,
        summary="Total, mean, middle, smallest and largest of a list of numbers, exactly.",
        tool_names=("summarise_numbers",),
    ),
)


def describe_catalogue() -> str:
    """Render the catalogue for an operator's console.

    Says what each server would add and what it still needs, because a console
    that lists capabilities without saying they are unapproved invites the
    reader to assume they are on.
    """

    lines = ["MCP servers available to install. None is installed or approved until you say so."]
    for entry in CATALOGUE:
        refs = ", ".join(
            f"{entry.manifest.tool_id_for(name)}@{entry.manifest.version}"
            for name in entry.tool_names
        )
        lines.append(f"  {entry.manifest.slug}: {entry.summary}")
        lines.append(f"    installs {refs}")
        if entry.requires_directory:
            lines.append(
                "    needs a directory you nominate; there is no default and it will not "
                "start without one"
            )
    lines.append(
        "  Installing registers a tool. Approving it is a separate decision, and approving "
        "network access is a third one that none of these servers asks for."
    )
    return "\n".join(lines)


# --------------------------------------------------------------------------
# Installing the catalogue: what turns an approved ref into a callable tool
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class CatalogueInstall:
    """What :func:`install_catalogue` did, and what it could not do.

    ``notes`` is the half that matters. A server the operator approved but
    which could not be built is skipped, so every call to it fails closed, and
    the reason has to travel back to somebody who can act on it. A silent skip
    would leave an operator believing a tool was on when it was not.

    ``skipped`` carries the refs that were recognised and not installed, so a
    caller can tell the difference between "this package knows that name and
    could not offer it" and "nothing anywhere knows that name", which are
    different problems with different fixes.
    """

    refs: tuple[ToolRef, ...] = ()
    skipped: tuple[ToolRef, ...] = ()
    notes: tuple[str, ...] = ()


def catalogue_refs() -> tuple[str, ...]:
    """Every ``id@version`` this package could install, sorted.

    Derived from the manifests rather than by building the servers, because
    the file server refuses to build without a directory and a list of what is
    approvable must not depend on whether one has been nominated yet.
    """

    return tuple(
        sorted(
            f"{entry.manifest.tool_id_for(name)}@{entry.manifest.version}"
            for entry in CATALOGUE
            for name in entry.tool_names
        )
    )


def _file_root(environ: Mapping[str, str]) -> str:
    value = str(environ.get(FILE_ROOT_ENV_VAR, "") or "").strip()
    if not value:
        raise McpError(
            f"no directory has been nominated, so the file server was not installed and "
            f"every call to it fails closed. Set {FILE_ROOT_ENV_VAR} to the folder this "
            "worker may read and write, and approve the tools by name"
        )
    return value


#: Slug to builder. Keyed by slug, and a slug with no entry here is reported
#: rather than crashing the registry build: adding a server to
#: :data:`CATALOGUE` and forgetting this table is a mistake in this repository,
#: and it should cost the operator that one server rather than every tool they
#: approved.
_BUILDERS = {
    LIST_MANIFEST.slug: lambda environ: list_mcp_server(),
    TEXT_MANIFEST.slug: lambda environ: text_mcp_server(),
    NUMBERS_MANIFEST.slug: lambda environ: numbers_mcp_server(),
    FILE_MANIFEST.slug: lambda environ: file_mcp_server(_file_root(environ)),
}


def _ref_order(ref: ToolRef) -> tuple[str, str]:
    """``ToolRef`` is frozen but not ordered, so sorting needs a key."""

    return (ref.tool_id, ref.version)


def install_catalogue(
    registry: Registry,
    *,
    only: Iterable[ToolRef],
    egress: Iterable[ToolRef] = (),
    environ: Mapping[str, str] | None = None,
) -> CatalogueInstall:
    """Install exactly the catalogue tools named in ``only``, and nothing else.

    ``only`` is the operator's approved ref set. A server whose tools are all
    absent from it is never built, which is why an operator who has not
    approved the file tools never has a directory resolved or a vault
    constructed on their behalf.

    ``egress`` is the operator's separate network approval. No server in this
    package declares a host, so today the check never fires; it is here so
    that adding one later cannot slip a networked tool in through this path
    with only the first of the two approvals. A manifest that declares hosts
    is refused unless every one of its wanted refs is in ``egress`` as well.

    Failures are per server and are returned rather than raised. A worker that
    refuses to start because one of four servers is misconfigured is worse
    than a worker that runs the other three and says which one is missing.
    """

    if not isinstance(registry, Registry):
        raise TypeError("registry must be a Registry")
    values = os.environ if environ is None else environ
    approved = frozenset(only)
    cleared = frozenset(egress)

    refs: list[ToolRef] = []
    skipped: list[ToolRef] = []
    notes: list[str] = []
    for entry in CATALOGUE:
        wanted = tuple(
            name
            for name in entry.tool_names
            if ToolRef(entry.manifest.tool_id_for(name), entry.manifest.version) in approved
        )
        if not wanted:
            continue
        wanted_refs = [
            ToolRef(entry.manifest.tool_id_for(name), entry.manifest.version) for name in wanted
        ]
        if entry.manifest.declares_network and not frozenset(wanted_refs) <= cleared:
            hosts = ", ".join(sorted(entry.manifest.allowed_hosts))
            notes.append(
                f"{entry.manifest.slug}: needs outbound access to {hosts} and this worker has "
                "not approved network access for it, so it was not installed"
            )
            skipped.extend(wanted_refs)
            continue
        builder = _BUILDERS.get(entry.manifest.slug)
        if builder is None:  # pragma: no cover - only reachable via a bug here
            notes.append(
                f"{entry.manifest.slug}: this package lists the server but has no way to "
                "build it, so it was not installed"
            )
            skipped.extend(wanted_refs)
            continue
        try:
            server = builder(values)
            # The manifest is narrowed to what was approved before anything is
            # read from the server, so a tool the operator did not name is
            # never registered even though the server offers it.
            installed = install_mcp_server(
                registry, server, replace(entry.manifest, tool_names=wanted)
            )
        except McpError as exc:
            notes.append(f"{entry.manifest.slug}: {exc}")
            skipped.extend(wanted_refs)
            continue
        refs.extend(item.ref for item in installed)
    return CatalogueInstall(
        tuple(sorted(refs, key=_ref_order)),
        tuple(sorted(skipped, key=_ref_order)),
        tuple(notes),
    )


__all__ = [
    "CATALOGUE",
    "FILE_MANIFEST",
    "FILE_ROOT_ENV_VAR",
    "LIST_MANIFEST",
    "NUMBERS_MANIFEST",
    "TEXT_MANIFEST",
    "CatalogueEntry",
    "CatalogueInstall",
    "FileVault",
    "catalogue_refs",
    "describe_catalogue",
    "file_mcp_server",
    "install_catalogue",
    "list_mcp_server",
    "numbers_mcp_server",
    "text_mcp_server",
]