Source code for pydocs_mcp.application.mcp_inputs

"""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, limit up to 1000 — 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
input-shape validators on these models (e.g., ``LookupInput.limit`` /
``SearchInput.limit`` defaults and ceilings), which are deployment-time
bounds, not feature toggles. ``configure_from_app_config`` is the single
wire that pushes the YAML-loaded ``AppConfig`` into the module-level
slots those validators 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
_TARGET_RE = re.compile(
    r"^(?:[a-zA-Z_][a-zA-Z0-9_]*(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)?$"
)  # empty or dotted-identifier chain; rejects foo..bar, foo., leading digit
_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 (e.g. markdown/decision
    document paths like ``docs.adr.0001-greeting-format.md``).

    Example: ``is_symbol_target("pkg.mod.X")`` is ``True``;
    ``is_symbol_target("docs.adr.0001-x.md")`` is ``False``.
    """
    return bool(text) and _TARGET_RE.match(text) is not None


# 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


@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 + ceiling).
       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
       ``ReferenceCaptureStage`` to gate capture on/off and pick which
       reference kinds to emit. 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)


[docs] class SearchInput(BaseModel): """Input for the ``search_codebase`` MCP tool.""" query: str = Field(min_length=1, max_length=30000) kind: KindLiteral = "any" package: str = "" scope: ScopeLiteral = "all" # Multi-repo corpus selector (sibling of ``package`` / ``scope``): restrict # the query to one loaded project by name. "" = union across all loaded # projects. No effect on a single-project server. project: str = "" # ``limit`` bounds the chunk-result count. Both the default and the # upper ceiling are driven by YAML (``search.output.default_limit`` / # ``max_limit``), pushed into module-level slots by # ``configure_from_app_config`` at server / CLI startup — parity with # ``LookupInput.limit`` (post-#5c). ``default_factory`` re-reads the # slot on every instantiation, and the ``@field_validator`` reads the # ceiling inside its body, so the model picks up YAML changes without # a re-import. limit: int = Field(default_factory=lambda: _SEARCH_LIMIT_DEFAULT, ge=1) @field_validator("limit") @classmethod def _check_limit_max(cls, v: int) -> int: # Read ``_SEARCH_LIMIT_MAX`` at call time so YAML reloads (or test # overrides) take effect on every ``SearchInput(...)`` rather than # being frozen at class-definition time. if v > _SEARCH_LIMIT_MAX: raise ValueError( f"limit must be <= {_SEARCH_LIMIT_MAX} (configured via search.output.max_limit)" ) return v @field_validator("package") @classmethod def _check_package(cls, v: str) -> str: if v and not _PACKAGE_RE.match(v): raise ValueError("package must match ^[a-zA-Z0-9][a-zA-Z0-9._-]*$ or be '__project__'") 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
[docs] class LookupInput(BaseModel): """Internal routing input for the deprecated ``lookup`` CLI verb — the MCP surface exposes ``get_symbol`` / ``get_references`` instead.""" target: str = "" show: Literal[ "default", "tree", "callers", "callees", "inherits", "impact", "context", "governed_by", ] = "default" # Multi-repo corpus selector: resolve the target inside one loaded project by # name. "" = resolve across all loaded projects (most-recent first). No effect # on a single-project server. project: str = "" # ``limit`` bounds reference-graph output (callers/callees/inherits). # The default and the upper ceiling are BOTH driven by YAML # (``reference_graph.output.default_limit`` / ``max_limit``), pushed # into module-level slots by ``configure_from_app_config`` at server / # CLI startup. ``default_factory`` re-reads the slot on every # instantiation, and the ``@field_validator`` reads the ceiling inside # its body, so the model picks up YAML changes without a re-import. limit: int = Field(default_factory=lambda: _LIMIT_DEFAULT, ge=1) @field_validator("target") @classmethod def _check_target(cls, v: str) -> str: if v and not _TARGET_RE.match(v): raise ValueError( "target must be a dotted identifier like 'pkg.mod.Class.method' or empty" ) 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 @field_validator("limit") @classmethod def _check_limit_max(cls, v: int) -> int: # Read ``_LIMIT_MAX`` at call time so YAML reloads (or test # overrides) take effect on every ``LookupInput(...)`` rather than # being frozen at class-definition time. if v > _LIMIT_MAX: raise ValueError( f"limit must be <= {_LIMIT_MAX} (configured via reference_graph.output.max_limit)" ) return v
# ── 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.
[docs] class OverviewInput(BaseModel): """get_overview — orientation card scope (spec §D1/§D17).""" package: str = "" project: str = "" @field_validator("package") @classmethod def _check_package(cls, v: str) -> str: if v and not _PACKAGE_RE.match(v): raise ValueError("package must match ^[a-zA-Z0-9][a-zA-Z0-9._-]*$ or be '__project__'") 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
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 SymbolInput(BaseModel): """get_symbol — known dotted path (spec §D1). depth='source' is the §D7 recovery contract.""" target: str = Field(min_length=1) depth: DepthLiteral = "summary" project: str = "" @field_validator("target") @classmethod def _check_target(cls, v: str) -> str: if not _TARGET_RE.match(v): raise ValueError("target must be a dotted identifier like 'pkg.mod.Class.method'") 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
[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: if not item or not _TARGET_RE.match(item): raise ValueError( "each target must be a dotted identifier like 'pkg.mod.Class.method'" ) 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
[docs] class ReferencesInput(BaseModel): """get_references — graph traversal incl. ranked transitive impact (spec §D1).""" target: str = Field(min_length=1) direction: DirectionLiteral = "callers" project: str = "" limit: int = Field(default_factory=lambda: _LIMIT_DEFAULT, ge=1) @field_validator("target") @classmethod def _check_target(cls, v: str) -> str: if not _TARGET_RE.match(v): raise ValueError("target must be a dotted identifier like 'pkg.mod.Class.method'") 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 @field_validator("limit") @classmethod def _check_limit_max(cls, v: int) -> int: # Read ``_LIMIT_MAX`` at call time so YAML reloads (or test # overrides) take effect on every ``ReferencesInput(...)`` rather # than being frozen at class-definition time — mirrors LookupInput. if v > _LIMIT_MAX: raise ValueError( f"limit must be <= {_LIMIT_MAX} (configured via reference_graph.output.max_limit)" ) return v
[docs] class WhyInput(BaseModel): """get_why — decision search / per-target governing decisions / dashboard (spec §D11).""" query: str = "" targets: list[str] | None = Field(None, min_length=1, max_length=20) project: str = "" @field_validator("targets") @classmethod def _check_targets(cls, v: list[str] | None) -> list[str] | None: # Unlike ContextInput, why-targets are documented as PATH|QNAME and # DecisionService._classify_target has a '/'-path branch — so this # validates against _WHY_TARGET_RE (admits '/'), keeping only the # pointer-grammar-hostile ':' / ']' rejected. if v is None: return v for item in v: if not item or not _WHY_TARGET_RE.match(item): raise ValueError( "each target must be a dotted name like 'pkg.mod.Class' or a path like 'a/b.py'" ) 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.
[docs] class GrepInput(BaseModel): """grep — exact-string / regex text search over source files (§3.7).""" model_config = ConfigDict(populate_by_name=True) pattern: str = Field(min_length=1) path: str = "" glob: str = "" output_mode: OutputModeLiteral = "files_with_matches" case_insensitive: Annotated[bool, Field(validation_alias="-i")] = False line_numbers: Annotated[bool, Field(validation_alias="-n")] = True after_context: Annotated[int | None, Field(validation_alias="-A", ge=0)] = None before_context: Annotated[int | None, Field(validation_alias="-B", ge=0)] = None context: Annotated[int | None, Field(validation_alias="-C", ge=0)] = None # None ⇒ YAML default (files.grep_head_limit) resolved by the service; # client-supplied values are ceilinged at files.max_head_limit below. head_limit: int | None = Field(None, ge=1) multiline: bool = False scope: ScopeLiteral = "project" project: str = "" @field_validator("pattern") @classmethod def _check_pattern_compiles(cls, v: str) -> str: # Reject uncompilable regexes at the boundary, carrying the # offending pattern (not a vague "invalid input") — the service # would otherwise raise deep inside the filesystem walk. try: re.compile(v) except re.error as exc: raise ValueError(f"pattern is not a valid Python regex: {v!r} ({exc})") from exc return v @field_validator("head_limit") @classmethod def _check_head_limit_max(cls, v: int | None) -> int | None: return _check_files_head_limit(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
[docs] class GlobInput(BaseModel): """glob — find files by name pattern, mtime-descending (§3.8).""" pattern: str = Field(min_length=1) path: str = "" # None ⇒ YAML default (files.glob_head_limit) resolved by the service. head_limit: int | None = Field(None, ge=1) project: str = "" @field_validator("head_limit") @classmethod def _check_head_limit_max(cls, v: int | None) -> int | None: return _check_files_head_limit(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
[docs] class ReadFileInput(BaseModel): """read_file — line-numbered file content within the corpus boundary (§3.9).""" file_path: str = Field(min_length=1) offset: int | None = Field(None, ge=1) # 1-indexed start line # None ⇒ YAML default (files.read_limit) resolved by the service. limit: int | None = Field(None, ge=1) project: str = "" @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