"""Pydantic input models for MCP tools (sub-PR #6 §4.3).
Enforces format via regex + protocol-safety caps. Limits are permissive:
query up to 30k chars, and a result limit bounded by YAML rather than
rejected — covers runaway clients without rejecting legit edge cases.
Per CLAUDE.md §"MCP API surface vs YAML configuration": pipeline tunables
live in YAML, NOT on the MCP tool surface. The one allowed exception is
the deployment-time bounds on these models (``LookupInput.limit`` /
``SearchInput.limit`` defaults, the lookup ceiling, the search ceiling
``clamp_search_limit`` applies), which are bounds, not feature toggles.
``configure_from_app_config`` is the single wire that pushes the
YAML-loaded ``AppConfig`` into the module-level slots they read at
runtime.
"""
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Annotated, Literal, Protocol, runtime_checkable
from pydantic import BaseModel, ConfigDict, Field, field_validator
if TYPE_CHECKING:
# Type-only import — keeps the runtime ``application -> retrieval`` edge
# lazy (the actual ``cfg.reference_graph`` / ``cfg.search`` attribute
# reads in ``configure_from_app_config`` happen on whatever the caller
# passes, validated structurally via ``_ConfigShape``). Avoids the
# circular-import risk called out in the pre-refactor docstring.
from pydocs_mcp.retrieval.config import (
FilesConfig,
ReferenceGraphConfig,
SearchConfig,
)
from pydocs_mcp.retrieval.config.models import SymbolSourceConfig
# ── Shared enum vocabularies (contract §3) ────────────────────────────────
#
# Single source for every constrained-string parameter on the nine-tool
# surface. Three consumers reference these aliases so the value sets can
# never drift: the pydantic input models below, the MCP handler signatures
# in ``server.py`` (typing the params with the aliases makes FastMCP
# advertise the enums in each tool's inputSchema — contract §6 note 3),
# and the argparse ``choices`` in ``__main__.py`` (via ``typing.get_args``).
KindLiteral = Literal["docs", "api", "any", "decision"]
ScopeLiteral = Literal["project", "deps", "all"]
DepthLiteral = Literal["summary", "tree", "source"]
DirectionLiteral = Literal["callers", "callees", "inherits", "impact", "governed_by"]
OutputModeLiteral = Literal["content", "files_with_matches", "count"]
# Format validators — reject malformed input at the boundary.
_PACKAGE_RE = re.compile(
r"^(?:[a-zA-Z0-9](?:[a-zA-Z0-9._-]*[a-zA-Z0-9])?|__project__)$"
) # alphanumeric start AND end (rejects trailing dot/dash); dots/dashes/underscores allowed in middle
# Dotted-target grammar (contract §3), widened per ADR 0023 decision (e) so the
# validator accepts every qualified name the index emits: dotted Python
# identifiers, module ids that keep their file suffix and may carry digits and
# hyphens inside a segment (``src.lib.rs``, ``my-pkg.mod``,
# ``docs.adr.0001-greeting-format.md``), and the ``module#slug`` heading anchors
# the markdown / notebook / text-section chunkers store. Before the widening a
# search row advertised names the symbol tools then refused.
_TARGET_SEGMENT = r"[A-Za-z0-9_][A-Za-z0-9_-]*"
# One segment shape for the dotted chain AND the anchor: the anchor slugs
# (``install-steps``, ``cell-3``, ``L1-40``) obey the same character set.
_TARGET_RE = re.compile(
rf"^(?:{_TARGET_SEGMENT}(?:\.{_TARGET_SEGMENT})*(?:#{_TARGET_SEGMENT})?)?$"
) # empty, or a dotted chain with one optional '#' anchor; rejects foo..bar,
# foo., a leading '-' or '.', spaces, path separators, control characters, and
# the ':' / ']' that would corrupt the [[next:…]] pointer-token grammar.
_SYMBOL_TARGET_SHAPE = (
"dot-separated segments of letters, digits, '_' or '-' (each starting with a "
"letter, digit or '_'), optionally followed by a '#' heading anchor — e.g. "
"'pkg.mod.Class.method', 'docs.adr.0001-notes.md' or 'docs.guide.md#install'"
)
_WHY_TARGET_RE = re.compile(
r"^[A-Za-z0-9_.\-/]+$"
) # get_why targets are documented as PATH|QNAME (DecisionService._classify_target
# branches on '/'), so '/' is admitted here — unlike _TARGET_RE, which guards the
# symbol/context tools. ':' and ']' stay forbidden: they are the only characters
# that corrupt the [[next:…]] pointer-token grammar (application/formatting.py).
def is_symbol_target(text: str) -> bool:
"""True iff ``text`` is a target the symbol-shaped tools would accept.
Single source of the dotted-target grammar (``_TARGET_RE``), exported for
pointer-render gating: ``application/formatting.py`` suppresses a
``get_symbol`` / ``get_context`` / ``get_references`` follow-up pointer
whose target this predicate rejects, so a response never advertises a
call the tools' own input validators refuse (a path-shaped name like
``docs/adr/0001-greeting-format.md``, a module id carrying a space).
Example: ``is_symbol_target("pkg.mod.X")`` and
``is_symbol_target("docs.adr.0001-x.md#context")`` are ``True``;
``is_symbol_target("docs/adr/0001-x.md")`` is ``False``.
"""
return bool(text) and _TARGET_RE.match(text) is not None
def _validated_symbol_target(value: str) -> str:
"""``value`` unchanged when it is a non-empty symbol target; raise otherwise.
One rejection message for all four symbol-shaped inputs, carrying the
offending value and the expected shape so a client can fix the call from
the error alone.
"""
if is_symbol_target(value):
return value
raise ValueError(f"invalid target: got {value!r}, expected {_SYMBOL_TARGET_SHAPE}")
# Module-level slots — installed by ``configure_from_app_config`` at
# server / CLI startup. The initial values match the shipped
# ``default_config.yaml`` (``reference_graph.output.default_limit=50``,
# ``max_limit=1000`` for LookupInput; ``search.output.default_limit=10``,
# ``max_limit=1000`` for SearchInput) so the models behave correctly even
# if ``configure_from_app_config`` is never called (e.g., direct unit
# tests that just instantiate ``LookupInput`` / ``SearchInput``).
#
# Why module-level globals instead of class-level defaults? Pydantic
# resolves field defaults / validator constraints at class definition,
# which happens at import time — long before ``AppConfig.load`` runs. A
# ``default_factory=lambda: _LIMIT_DEFAULT`` and a validator that reads
# ``_LIMIT_MAX`` inside its body re-read the slots on every model
# instantiation, so a single ``configure_from_app_config`` call at startup
# is enough to make every subsequent ``LookupInput(...)`` /
# ``SearchInput(...)`` validate against the YAML-supplied bounds.
_LIMIT_DEFAULT: int = 50
_LIMIT_MAX: int = 1000
# SearchInput's historical default is 10 (not 50) — distinct knob from
# LookupInput. Both pairs are tunable via separate YAML sub-models
# (``reference_graph.output.*`` vs ``search.output.*``); deployments can
# adjust one without the other.
_SEARCH_LIMIT_DEFAULT: int = 10
_SEARCH_LIMIT_MAX: int = 1000
# get_symbol(depth="source") verbatim line cap — installed from
# ``cfg.symbol_source.max_lines`` by ``configure_from_app_config`` at
# startup. Initial literal matches the shipped ``default_config.yaml``
# (``symbol_source.max_lines: 400``) so direct construction of
# ``SymbolSourceService`` behaves correctly if ``configure_from_app_config``
# is never called (same rationale as the limit slots above — importing the
# canonical config constant here would invert the application -> retrieval
# layering). The per-project ``build_sqlite_symbol_source_service`` factory
# reads this slot for its fallback when no ``config`` is passed.
_SYMBOL_SOURCE_MAX_LINES: int = 400
# Ceiling for client-supplied ``head_limit`` / ``limit`` caps on the
# filesystem tools (grep/glob/read_file) — installed from
# ``cfg.files.max_head_limit`` by ``configure_from_app_config``. Initial
# literal matches the shipped ``default_config.yaml``
# (``files.max_head_limit: 10000``). The per-tool YAML *defaults*
# (``files.grep_head_limit`` etc.) intentionally have NO slot here: the
# models keep ``head_limit=None`` and ``FileToolsService._effective_limit``
# resolves ``None`` against its ``files_config`` — only the ceiling is an
# input-shape concern.
_FILES_HEAD_LIMIT_MAX: int = 10000
def clamp_search_limit(requested: int) -> int:
"""``requested`` bounded by the configured ``search.output.max_limit``.
The single source of the search ceiling for every consumer of a client
``limit``: the slot is read at call time, so a YAML reload takes effect
without a re-import. Pure — the caller decides whether a clamp is worth
reporting (``application/search_limit.py`` does).
Example: ``clamp_search_limit(5000)`` is ``1000`` on the shipped config.
"""
return min(requested, _SEARCH_LIMIT_MAX)
@runtime_checkable
class _ConfigShape(Protocol):
"""Structural shape of the YAML-loaded ``AppConfig`` that
:func:`configure_from_app_config` reads (I11).
Replaces the previous ``cfg: Any`` parameter with a typed Protocol
that documents exactly which cfg sub-trees the function consumes —
``reference_graph`` (for ``capture`` / ``resolver`` / ``output``
sub-models), ``search`` (for the ``search.output`` bounds),
``symbol_source`` (for the get_symbol(depth="source") line cap), and
``files`` (for the grep/glob/read_file ``max_head_limit`` ceiling).
The Protocol is ``@runtime_checkable`` so unit tests can structurally
verify any duck-typed config carrier satisfies the shape via
``isinstance(cfg, _ConfigShape)``. The real ``AppConfig`` (defined in
:mod:`pydocs_mcp.retrieval.config`) satisfies this Protocol nominally
— no nominal subclassing required.
Module-private (``_ConfigShape``) on purpose: external callers should
pass a real ``AppConfig`` instance from
:class:`pydocs_mcp.retrieval.config.AppConfig`. The Protocol exists
only to document the structural contract and avoid a hard runtime
dependency on ``AppConfig`` at the type level (which would create a
circular import between ``application`` and ``retrieval``).
"""
reference_graph: ReferenceGraphConfig
search: SearchConfig
symbol_source: SymbolSourceConfig
files: FilesConfig
def configure_from_app_config(cfg: _ConfigShape) -> None:
"""Install YAML-loaded settings into the module-level slots this
package reads at runtime.
Called ONCE at server / CLI startup (see ``server.py::run`` and
``__main__.py::_cmd_*``). The parameter is typed against
:class:`_ConfigShape` — a structural Protocol covering the two cfg
sub-trees this function reads (``reference_graph`` and ``search``).
Stamp coupling is gone: callers no longer pass an untyped ``Any``;
static type-checkers can verify the contract.
Five slots are updated:
1. ``_LIMIT_DEFAULT`` / ``_LIMIT_MAX`` here in ``mcp_inputs`` — read
by ``LookupInput.limit`` (default + ceiling).
2. ``_SEARCH_LIMIT_DEFAULT`` / ``_SEARCH_LIMIT_MAX`` here in
``mcp_inputs`` — read by ``SearchInput.limit`` (default) and by
``clamp_search_limit`` (ceiling, applied rather than rejected).
Separate slot pair so deployments can tune search and lookup
limits independently.
3. ``_SYMBOL_SOURCE_MAX_LINES`` here in ``mcp_inputs`` — the
get_symbol(depth="source") verbatim line cap, read by the
per-project ``build_sqlite_symbol_source_service`` factory as its
no-config fallback.
4. ``_FILES_HEAD_LIMIT_MAX`` here in ``mcp_inputs`` — the ceiling on
client-supplied ``head_limit`` caps for ``GrepInput`` /
``GlobInput`` (tool-contracts.md §3.7-3.8).
5. ``_CAPTURE_CONFIG`` in ``extraction.pipeline.stages`` — read by a
``ReferenceCaptureStage`` built WITHOUT an app config to gate capture
on/off and pick which reference kinds to emit; a stage built from a
pipeline YAML binds ``reference_graph.capture`` itself, the object
``ContentHashStage`` folds (issue #347). Pushed via
``_set_capture_config`` so the stage module owns its own slot (no
cross-package mutation).
"""
global _LIMIT_DEFAULT, _LIMIT_MAX
global _SEARCH_LIMIT_DEFAULT, _SEARCH_LIMIT_MAX
global _SYMBOL_SOURCE_MAX_LINES
global _FILES_HEAD_LIMIT_MAX
output = cfg.reference_graph.output
_LIMIT_DEFAULT = output.default_limit
_LIMIT_MAX = output.max_limit
search_output = cfg.search.output
_SEARCH_LIMIT_DEFAULT = search_output.default_limit
_SEARCH_LIMIT_MAX = search_output.max_limit
_SYMBOL_SOURCE_MAX_LINES = cfg.symbol_source.max_lines
_FILES_HEAD_LIMIT_MAX = cfg.files.max_head_limit
# Local import — keeps the application -> extraction edge lazy so
# importing ``mcp_inputs`` at app startup doesn't drag in the whole
# extraction pipeline (or fight with the existing import order).
from pydocs_mcp.extraction.pipeline.stages import (
_set_capture_config,
_set_similar_config,
)
_set_capture_config(cfg.reference_graph.capture)
# Synthetic kNN 'similar' edge generation (off by default) — same
# module-level-slot push as the capture config above.
_set_similar_config(cfg.reference_graph.similar_edges)
# AC #15 stdlib-idx: push resolver config so IndexingService picks up
# the include_stdlib toggle on next reindex. Parity with the capture
# config push above — same module-level slot pattern.
from pydocs_mcp.extraction.strategies.stdlib_qnames import (
_set_resolver_config,
)
_set_resolver_config(cfg.reference_graph.resolver)
# ── Task-shaped tool inputs (spec §D1) ──────────────────────────────────
#
# The task-shaped tools reuse the same YAML-wired limit slots and the
# same ``_PACKAGE_RE`` / ``_TARGET_RE`` boundary validators as
# ``SearchInput`` / ``LookupInput`` above. ``project`` carries the
# identical multi-repo corpus-selector semantics on every model that has it.
def _check_files_head_limit(v: int | None) -> int | None:
"""Shared ceiling check for GrepInput.head_limit / GlobInput.head_limit.
Reads ``_FILES_HEAD_LIMIT_MAX`` at call time (not class-definition
time) so a single ``configure_from_app_config`` call at startup makes
every subsequent instantiation validate against the YAML ceiling —
same pattern as the SearchInput/LookupInput limit validators.
"""
if v is not None and v > _FILES_HEAD_LIMIT_MAX:
raise ValueError(
f"head_limit must be <= {_FILES_HEAD_LIMIT_MAX} (configured via files.max_head_limit)"
)
return v
[docs]
class ContextInput(BaseModel):
"""get_context — batched targets under one shared budget (spec §D1)."""
targets: list[str] = Field(min_length=1, max_length=20)
project: str = ""
@field_validator("targets")
@classmethod
def _check_targets(cls, v: list[str]) -> list[str]:
# Each item must be a non-empty dotted identifier, mirroring
# SymbolInput/ReferencesInput.target — an unvalidated item gets
# interpolated into "[[...]]" pointer tokens downstream
# (formatting.py), so ":" / "]]" here would corrupt the grammar.
for item in v:
_validated_symbol_target(item)
return v
@field_validator("project")
@classmethod
def _check_project(cls, v: str) -> str:
if v and not _PACKAGE_RE.match(v):
raise ValueError("project must match ^[a-zA-Z0-9][a-zA-Z0-9._-]*$")
return v
# ── Filesystem tool inputs (tool-contracts.md §3.7-3.9) ──────────────────
#
# The grep flags -i/-n/-A/-B/-C are the contract's literal wire names.
# ``validation_alias`` (NOT ``alias``) is load-bearing: FastMCP's
# ``model_dump_one_level`` prefers ``alias`` for the handler kwarg name
# (mcp/server/fastmcp/utilities/func_metadata.py), which would break
# invocation — ``validation_alias`` was spike-verified to (a) advertise
# the dash names in the inputSchema, (b) accept dash-keyed calls, and
# (c) dump by Python field name (what ``FileToolsService``'s structural
# ``GrepRequest`` Protocol reads). ``populate_by_name=True`` lets
# server-internal callers (CLI verbs, tests) construct by field name
# without dash-keyed dicts.