"""The ``search`` + ``execute`` meta-tool pair for the long-tail verb surface.
The curated domain toolsets (``_toolsets``) cover the common path; the rest of
the operator-callable verb tree is reached through two meta-tools, the Cloudflare
precedent for a large API surface (ADR decision R2): ``search`` maps a natural
query onto matching command keys with their intent and mutability, and
``execute`` runs one command key with typed arguments.
The load-bearing guarantee is that ``execute`` is NOT a side door. It routes
through the exact same gates a direct tool call runs - :func:`gate_refusal`
applies the persona-scope boundary and the permanent live-write block before any
dispatch, producing byte-identical refusals - so a verb an active persona may not
call directly cannot be reached by naming it to ``execute`` either. The dispatch
itself is injected (:func:`meta_execute` takes the server's subprocess runner) so
this module stays SDK-independent and unit-tested.
"""
from __future__ import annotations
from collections.abc import Callable
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from ...application.command_search import CommandDoc, CommandIndex, build_command_index
from ...application.operator_surface import declared_risk
from ._hitl import ConfirmationPolicy, confirmation_for_tool
from ._persona_scope import AgentPersona, handoff_denial_message, is_handoff_denied, is_tool_in_persona_scope
from ._tools import McpToolDescriptor
from ._toolsets import MAX_ACTIVE_TOOLSETS, Toolset, _family_domain_map, build_toolsets, toolset_for_command
_STRICT_FROZEN = ConfigDict(frozen=True, strict=True, validate_assignment=True, extra="forbid")
#: The subprocess runner the server injects into :func:`meta_execute`: it takes a
#: descriptor and the named arguments and returns the CLI envelope plus an error
#: flag, exactly as the direct call path runs it.
ToolRunner = Callable[[McpToolDescriptor, dict[str, object]], tuple[dict[str, object], bool]]
#: Curated outcome-vocabulary aliases folded into a command's search document
#: (ADR ``mcp-progressive-discovery`` P2/S08). This is INDEX text only, never the
#: model-facing tool description (H5 English-boundary): it lets an outcome-phrased
#: query ("file my quarterly VAT", "do my taxes") reach the composite ``quickfile``
#: chain that literal-verb tokens miss. English plus the Spanish outcome nouns the
#: CLI help uses (``presentar``, ``declaración``, ``trimestral``, ``autoliquidación``).
_COMMAND_ALIASES: dict[str, str] = {
"quickfile": (
"file my taxes do my taxes file my return file quarterly taxes "
"submit quarterly VAT IVA tax return declaration "
"presentar la declaración trimestral autoliquidación de impuestos "
"one command full filing chain"
),
}
def _command_doc(descriptor: McpToolDescriptor) -> CommandDoc:
"""Build the weighted searchable document for one command.
The document splits into BM25-weighted tiers (ADR ``mcp-progressive-discovery``
P2/S07): ``key_and_name`` (the command key's dotted tokens - ``calculate``,
``export``, ``iva_wallet`` - plus the tool name) ranks highest, curated outcome
``aliases`` next, then the English ``description``, then the command's own
per-verb CLI ``help``. The help is the CLI's Spanish domain vocabulary, so
folding it into the index - NOT into the English model-facing description (H5) -
lets a Spanish concept query recall the right verb; the aliases surface the
composite verbs on outcome-phrased queries without touching the model surface.
"""
key_tokens = descriptor.command_key.replace(".", " ").replace("_", " ")
key_and_name = f"{descriptor.command_key} {key_tokens} {descriptor.name}"
return CommandDoc(
command_key=descriptor.command_key,
tool_name=descriptor.name,
key_and_name=key_and_name,
description=descriptor.description,
aliases=_COMMAND_ALIASES.get(descriptor.command_key, ""),
help=descriptor.verb_schema.help,
)
[docs]
def build_command_search_index(descriptors: tuple[McpToolDescriptor, ...]) -> CommandIndex:
"""Build the hybrid command-search index over the descriptor set.
Built once per server from the full descriptor set so ``search`` reaches the
whole verb universe, not just the advertised surface.
"""
return build_command_index(_command_doc(descriptor) for descriptor in descriptors)
[docs]
def search_commands(
query: str,
*,
descriptors: tuple[McpToolDescriptor, ...],
index: CommandIndex | None = None,
limit: int = 20,
) -> tuple[MetaSearchResult, ...]:
"""Rank the command surface against ``query`` for the ``search`` meta-tool.
Backed by the hybrid command index (FTS5 lexical + Spanish stemming +
diacritics folding, degrading to token overlap on a minimal install), so a
concept query bridges the operator's vocabulary to the command's own tokens
where a bare substring match would miss it (ADR ``mcp-progressive-discovery``
P2). Each result carries the mutability hints AND the per-verb input schema so
it is actionable in one further ``execute`` round-trip. ``index`` may be a
prebuilt index (the server builds it once); when omitted it is built from
``descriptors``.
Returns:
The matched commands, highest score first, capped at ``limit``.
"""
if not query.strip():
return ()
search_index = index if index is not None else build_command_search_index(descriptors)
by_key = {descriptor.command_key: descriptor for descriptor in descriptors}
results: list[MetaSearchResult] = []
for hit in search_index.search(query, limit=limit):
descriptor = by_key.get(hit.command_key)
if descriptor is None:
continue
results.append(
MetaSearchResult(
command_key=descriptor.command_key,
tool_name=descriptor.name,
description=descriptor.description,
read_only=descriptor.annotations.read_only_hint,
destructive=descriptor.annotations.destructive_hint,
score=hit.score,
input_schema=descriptor.input_schema,
)
)
return tuple(results)
[docs]
def search_commands_response(
query: str,
*,
descriptors: tuple[McpToolDescriptor, ...],
index: CommandIndex | None = None,
limit: int = 20,
) -> MetaSearchResponse:
"""Rank the command surface and report how much of it overflowed the page.
Returns the same capped page as :func:`search_commands` alongside the full
match count over the whole verb corpus, so the client sees whether the page
truncated the result set. ``total_matches`` counts every command key the index
matches (capped internally at the corpus size, never a semantic backend - that
is a later step); ``truncated`` and the recovery ``hint`` follow from it. A
blank query returns the empty response.
Returns:
The :class:`MetaSearchResponse` for ``query``.
"""
if not query.strip():
return MetaSearchResponse(results=(), total_matches=0, truncated=False, hint="")
search_index = index if index is not None else build_command_search_index(descriptors)
results = search_commands(query, descriptors=descriptors, index=search_index, limit=limit)
known_keys = {descriptor.command_key for descriptor in descriptors}
all_hits = search_index.search(query, limit=len(descriptors))
total_matches = sum(1 for hit in all_hits if hit.command_key in known_keys)
truncated = total_matches > len(results)
hint = (
"More commands matched than were returned; narrow the query, call `describe` with a "
"command_key for one command's full schema, or activate a domain toolset through `toolsets` "
"to widen the advertised surface."
if truncated
else ""
)
return MetaSearchResponse(results=results, total_matches=total_matches, truncated=truncated, hint=hint)
[docs]
def describe_command(
command_key: str,
*,
descriptors: tuple[McpToolDescriptor, ...],
) -> MetaDescribeResult | None:
"""Return one command's full descriptor by key, or ``None`` when unexposed.
Resolves everything from the live descriptor set and the real classifiers -
the annotation hints from the descriptor, the confirmation tier from
:func:`~entrypoints.mcp._hitl.confirmation_for_tool`, the declared risk from
:func:`~application.operator_surface.declared_risk` (all-false for a read-only
command with no row), the owning toolset from
:func:`~entrypoints.mcp._toolsets.toolset_for_command`, and the reachable
personas from the same scope + handoff-deny gates the call path enforces. A
key that names no exposed descriptor returns ``None``.
Returns:
The :class:`MetaDescribeResult` for ``command_key``, or ``None``.
"""
descriptor = next((candidate for candidate in descriptors if candidate.command_key == command_key), None)
if descriptor is None:
return None
risk = declared_risk(command_key)
toolset = toolset_for_command(command_key, family_map=_family_domain_map())
reachable = tuple(
sorted(
persona.value
for persona in AgentPersona
if is_tool_in_persona_scope(persona=persona, command_key=command_key)
and not is_handoff_denied(persona=persona, command_key=command_key)
)
)
return MetaDescribeResult(
command_key=descriptor.command_key,
tool_name=descriptor.name,
description=descriptor.description,
input_schema=descriptor.input_schema,
read_only=descriptor.annotations.read_only_hint,
destructive=descriptor.annotations.destructive_hint,
idempotent=descriptor.annotations.idempotent_hint,
open_world=descriptor.annotations.open_world_hint,
confirmation_tier=confirmation_for_tool(command_key=command_key).value,
risk_destructive=risk.destructive if risk is not None else False,
risk_handoff=risk.handoff if risk is not None else False,
risk_live_write=risk.live_write if risk is not None else False,
owning_toolset=toolset.value if toolset is not None else None,
reachable_personas=reachable,
)
[docs]
def gate_refusal(*, persona: AgentPersona | None, descriptor: McpToolDescriptor) -> str | None:
"""Return the refusal a tool call incurs, or ``None`` when it may proceed.
The single gate sequence run by BOTH the direct call path and ``execute``: an
out-of-scope call is refused, then a persona's handoff-denied verb, then the
permanent live-write block. The messages are byte-identical to the direct
path's refusals, so the two entry points cannot diverge — the per-verb
handoff deny (verifier-only export/record-marker, ADR R6(iii)) is enforced
here STRUCTURALLY, not left to the sync path's incidental no-elicitation
fallback.
Returns:
The refusal message, or ``None`` when the call is allowed.
"""
if persona is not None and not is_tool_in_persona_scope(persona=persona, command_key=descriptor.command_key):
return f"refused: {descriptor.command_key!r} is outside the active persona {persona.value!r}'s tool scope"
if persona is not None and is_handoff_denied(persona=persona, command_key=descriptor.command_key):
return handoff_denial_message(persona=persona, command_key=descriptor.command_key)
if (
confirmation_for_tool(command_key=descriptor.command_key)
is ConfirmationPolicy.BLOCK
):
return "refused: AEAT live-write is permanently forbidden"
return None