distinct_mcp/text.py

"""Three mechanical operations on text, chosen against the viability bar.

:mod:`distinct_tools.local` states the filter this package inherits: below 7B,
what survives is a tool that answers in one call, is deterministic, and leaves
no ambiguity about when to call it. Multi-call, stateful benefit is unreachable
when BFCL multi-turn accuracy is 16.88 per cent at 1.7B and 1.38 per cent at
0.6B. Each tool here is argued against that bar rather than added because it
seemed handy.

``count_text``. A language model does not see characters; it sees tokens, and
counting by generating tokens is the failure this replaces. The trigger is
about as unambiguous as a trigger gets, because the user's question contains the
words "how many". One call, one exact answer, no state.

``sort_list``. Ordering a list by generating it again is where a small model
drops an item or invents one. The tool returns the same items in a defined
order and reports how many came back, so a loss is visible rather than silent.
Determinism needs a *total* order, not merely a sensible one: items compare
case-insensitively first and by code point second, so two spellings that differ
only in case can never swap places between calls.

``check_json``. :mod:`distinct_tools.local` names deterministic verification as
a supported category, on the grounds that amplifying a weak model needs an
external soundness signal and that a model reviewing its own work is documented
to make reasoning worse. Whether a string parses as JSON is exactly that kind of
signal: external, exact, and not a judgement. It pairs with the file server,
where the alternative to checking is saving malformed JSON to a user's disk.

**What is not claimed.** Nothing here has been measured on a small model in
this project. The argument is that each tool computes an exact answer to a
subproblem, which is the category the tool-use literature supports most
strongly, rather than supplying more text to reason over, which is the category
it supports least. Whether a 0.6B model *remembers* to call them is a different
question and this package has no measurement of it. Every description says when
not to call the tool, because a description that never says to stop is the most
common cause of a repeated-tool loop.
"""

from __future__ import annotations

import json
from collections.abc import Mapping, Sequence
from typing import Any

from distinct_tools.core import ToolInputError
from distinct_tools.local import MAX_TEXT_CHARACTERS
from distinct_tools.mcp import InProcessMcpServer, McpServerManifest, McpToolDefinition

from .common import choice_argument, flag_argument, reject_unknown, structured, text_argument

MAX_ITEMS = 200
MAX_ITEM_CHARACTERS = 200
MAX_PHRASE_CHARACTERS = 200

_ORDERS = ("ascending", "descending")

#: The manifest an operator would approve to install this server. No hosts, and
#: a short timeout: every call is a pure function of its arguments and none of
#: them can block on anything.
TEXT_MANIFEST = McpServerManifest(slug="text", version="1", timeout_seconds=5.0)


def _count_text(arguments: Mapping[str, Any]) -> dict[str, Any]:
    reject_unknown(arguments, frozenset({"text", "phrase"}))
    text = text_argument(arguments, "text", MAX_TEXT_CHARACTERS)
    phrase = text_argument(arguments, "phrase", MAX_PHRASE_CHARACTERS, required=False)
    result: dict[str, Any] = {
        "characters": len(text),
        "characters_excluding_spaces": sum(1 for item in text if not item.isspace()),
        # A word is a run of non-whitespace. Said in the description too,
        # because a count whose rule the caller cannot see is not checkable.
        "words": len(text.split()),
        "lines": len(text.splitlines()),
    }
    if phrase:
        result["phrase"] = phrase
        # ``str.count`` counts non-overlapping occurrences, so "aa" appears
        # once in "aaa" and not twice. Stated rather than left to be found.
        result["occurrences"] = text.count(phrase)
        result["occurrences_ignoring_case"] = text.lower().count(phrase.lower())
    return structured(result)


def _sort_key(value: object) -> tuple[str, str]:
    text = str(value)
    return (text.casefold(), text)


def _sort_list(arguments: Mapping[str, Any]) -> dict[str, Any]:
    reject_unknown(arguments, frozenset({"items", "order", "unique"}))
    items = arguments.get("items")
    if not isinstance(items, Sequence) or isinstance(items, str | bytes):
        raise ToolInputError("items must be a list")
    if not items:
        raise ToolInputError("items must contain at least one entry")
    if len(items) > MAX_ITEMS:
        raise ToolInputError(f"items must contain at most {MAX_ITEMS} entries")

    numeric = all(isinstance(item, int | float) and not isinstance(item, bool) for item in items)
    textual = all(isinstance(item, str) for item in items)
    if not numeric and not textual:
        # Refused rather than guessed. Sorting a mixture means choosing an
        # order between a number and a word, and any choice would be this
        # tool's opinion rather than the caller's instruction.
        raise ToolInputError(
            "items must be all numbers or all words, not a mixture. Pass numbers as numbers "
            "so they are ordered by value"
        )
    if textual:
        for item in items:
            if len(item) > MAX_ITEM_CHARACTERS:
                raise ToolInputError(f"each item must be at most {MAX_ITEM_CHARACTERS} characters")

    order = choice_argument(arguments, "order", _ORDERS, default="ascending")
    unique = flag_argument(arguments, "unique")
    key = None if numeric else _sort_key
    ordered = sorted(items, key=key, reverse=order == "descending")
    if unique:
        deduplicated: list[Any] = []
        for item in ordered:
            if not deduplicated or item != deduplicated[-1]:
                deduplicated.append(item)
        ordered = deduplicated
    return structured(
        {
            "items": list(ordered),
            "count": len(ordered),
            "removed": len(items) - len(ordered),
            "order": order,
            "sorted_as": "numbers" if numeric else "words",
            "already_in_order": list(items) == list(ordered),
        }
    )


def _reject_constant(name: str) -> Any:
    raise ValueError(f"{name} is not valid JSON")


def _check_json(arguments: Mapping[str, Any]) -> dict[str, Any]:
    reject_unknown(arguments, frozenset({"text"}))
    text = text_argument(arguments, "text", MAX_TEXT_CHARACTERS)
    try:
        # ``parse_constant`` refuses NaN and Infinity, which Python accepts by
        # default and the JSON grammar does not. "Valid" here has to mean what
        # the next reader will accept, not what this parser tolerates.
        value = json.loads(text, parse_constant=_reject_constant)
    except ValueError as exc:
        line = getattr(exc, "lineno", None)
        column = getattr(exc, "colno", None)
        return structured(
            {
                "valid": False,
                "error": str(exc)[:200],
                "line": line if isinstance(line, int) else None,
                "column": column if isinstance(column, int) else None,
            }
        )
    kinds = {
        dict: "object",
        list: "array",
        str: "string",
        bool: "true or false",
        int: "number",
        float: "number",
        type(None): "null",
    }
    size = len(value) if isinstance(value, dict | list) else None
    return structured(
        {
            "valid": True,
            "error": None,
            "top_level": kinds.get(type(value), "unknown"),
            # Reported so the caller can check nothing was lost, without the
            # document being sent back: the caller already has the text, and
            # returning it would spend the run's output budget on a copy.
            "size": size,
        }
    )


_DEFINITIONS: tuple[tuple[McpToolDefinition, Any], ...] = (
    (
        McpToolDefinition(
            name="count_text",
            description=(
                "Count exactly how many characters, words and lines a piece of text has, and how "
                "many times a phrase appears in it. Use it whenever the answer has to be a "
                "number of things: counting by reading is unreliable and this is exact. A word "
                "is a run of characters between spaces, and a phrase is counted only where it "
                "appears without overlapping. It counts and does nothing else: it cannot tell "
                "you what the text means, and to find the passages about a topic use "
                "search_document instead. Do not call it twice for the same text and phrase."
            ),
            input_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["text"],
                "properties": {
                    "text": {
                        "type": "string",
                        "minLength": 1,
                        "maxLength": MAX_TEXT_CHARACTERS,
                        "description": "the text to count",
                    },
                    "phrase": {
                        "type": "string",
                        "maxLength": MAX_PHRASE_CHARACTERS,
                        "description": "optional phrase to count occurrences of",
                    },
                },
            },
        ),
        _count_text,
    ),
    (
        McpToolDefinition(
            name="sort_list",
            description=(
                "Put a list into order, and remove repeats if you ask it to. Give numbers as "
                "numbers and words as strings, and do not mix the two in one list: a list of "
                "words is ordered as words, so '10' comes before '9'. Use it when the user asked "
                "for something in order, or when the list is long enough that ordering it "
                "yourself would drop an entry. It returns how many entries came back so you can "
                "see that none was lost. It never changes the entries themselves and cannot "
                "order them by meaning or importance."
            ),
            input_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["items"],
                "properties": {
                    "items": {
                        "type": "array",
                        "minItems": 1,
                        "maxItems": MAX_ITEMS,
                        "description": "the entries, all numbers or all words",
                        "items": {"type": ["string", "number"]},
                    },
                    "order": {
                        "type": "string",
                        "enum": list(_ORDERS),
                        "description": "smallest first (ascending) or largest first (descending)",
                    },
                    "unique": {
                        "type": "boolean",
                        "description": "true to keep only one of each repeated entry",
                    },
                },
            },
        ),
        _sort_list,
    ),
    (
        McpToolDefinition(
            name="check_json",
            description=(
                "Check whether a piece of text is valid JSON and, when it is not, say where it "
                "first goes wrong. Use it before handing JSON to someone or saving it to a file. "
                "It answers yes or no with the line and column of the problem. It does not "
                "repair the text and does not send the text back to you, because you already "
                "have it. Do not call it on text that was never meant to be JSON."
            ),
            input_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["text"],
                "properties": {
                    "text": {
                        "type": "string",
                        "minLength": 1,
                        "maxLength": MAX_TEXT_CHARACTERS,
                        "description": "the text that should be JSON",
                    }
                },
            },
        ),
        _check_json,
    ),
)


def text_mcp_server() -> InProcessMcpServer:
    """An in-process MCP server publishing the three text tools.

    Install it with
    ``install_mcp_server(registry, text_mcp_server(), TEXT_MANIFEST)``, which
    produces ``text.count_text@1``, ``text.sort_list@1`` and
    ``text.check_json@1``. None of them opens a socket, reads a file, spawns a
    process or consults a model.
    """

    server = InProcessMcpServer()
    for definition, handler in _DEFINITIONS:
        server.add_tool(definition, handler)
    return server


__all__ = ["MAX_ITEMS", "MAX_ITEM_CHARACTERS", "TEXT_MANIFEST", "text_mcp_server"]