"""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"]