MCP utilities¶

Shared machinery for building MCP servers and clients used by Klea apps: metadata schemas, tool registration, tool access levels, an httpx session lifespan, path permission (discovery, the sensitive-file matcher, and the interactive-approval payload/schema), and the reusable bundled tools server.

Schemas¶

Schemas shared by MCP servers and clients.

File: klea_utils/mcp/schemas.py

class klea_utils.mcp.schemas.ToolCallSchema(*, tool: str = '', args: dict[str, ~typing.Any]=<factory>, reason: str = '', paths: list[str] = <factory>, step: int = 0)[source]¶

Bases: BaseModel

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

paths: list[str]¶: Filesystem paths the picker declared the call will touch, used by the client-side permission gate as the expected discovery tier (ADR-0007 update 2026-10-08). Empty when the picker declared none.

step: int¶: 1-based plan step this call belongs to (0 when there is no plan, e.g. RAG). Set by the agent’s picker so a batch of calls can be attributed back to their originating step (ADR-0041).

class klea_utils.mcp.schemas.ToolCallsSchema(*, tool_calls: list[ToolCallSchema] = <factory>)[source]¶

Bases: BaseModel

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Bases: BaseModel

Metadata used to describe an MCP tool to clients and models.

The read_only / destructive / idempotent / open_world fields map 1:1 to the standard MCP ToolAnnotations hints (readOnlyHint / destructiveHint / idempotentHint / openWorldHint); see https://fastmcp.wiki/en/servers/tools#mcp-annotations for what each hint means and how a client is expected to act on it. klea_utils.mcp.registry.register_tools() folds them onto the registered tool.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Errors¶

Custom error classes for the MCP tooling.

File: klea_utils/mcp/errors.py

exception klea_utils.mcp.errors.DocumentConversionError[source]¶

Bases: Exception

Raised when a document file cannot be converted to text.

Carries a user-facing message that tools report through their error result field.

exception klea_utils.mcp.errors.FileEditError[source]¶

Bases: Exception

Raised when a file cannot be read or written by the editing tools.

Carries a user-facing message that the write/edit tool implementations report through their error result field (e.g. a missing file, a binary or non-UTF-8 file, or an atomic-write failure).

exception klea_utils.mcp.errors.PermissionDeniedError[source]¶

Bases: PermissionError

Raised when a tool is denied access to a path.

Tool registration¶

Shared MCP tool registration helpers.

File: klea_utils/mcp/registry.py

klea_utils.mcp.registry.register_tools(mcp: fastmcp.FastMCP, modules: list[ModuleType])[source]¶

A function is registered as a tool when it is decorated with tool_meta() (which attaches ToolInfo metadata). The function name is used as the tool name. Helper functions in the same module that are not decorated are ignored (logged at debug level, so a forgotten decoration is easy to spot). Only functions defined in the given module are registered, so an imported decorated function is not picked up accidentally.

Also installs _DropNullOptionalsMiddleware, which tolerates weak models sending explicit null for optional arguments (treated as “use the default”).

Parameters:

mcp – FastMCP server to register the tools on.
modules – list of modules with tool function definitions

klea_utils.mcp.registry.tool_meta(metadata: ToolInfo)[source]¶

Decorator that attaches ToolInfo metadata to a tool function.

Usage:

@tool_meta(ToolInfo(tags={"bundled", "web"}))
async def web_fetch(ctx: Context, url: str, ...):
    ...

The metadata is read by register_tools() when the function is registered on a FastMCP server (it sets description, title, tags, and meta on the tool if provided). A function is only registered as a tool when it carries this decoration; the function name is used as the tool name.

Parameters:: metadata – ToolInfo to attach to the decorated function.
Returns:: The decorated function, unchanged, with _tool_meta set.

Tool call dispatch¶

Client-side MCP tool-call dispatch with permission gating.

File: klea_utils/mcp/dispatch.py

klea_utils.mcp.dispatch.DEFAULT_TOOL_CALL_TIMEOUT_SECONDS = 900.0¶: Default per-call wall-clock cap for tool calls. A generous backstop that only fires when a tool hangs; individual tools own their semantic timeouts (e.g. run_command), so raise this above any tool’s own maximum.

klea_utils.mcp.dispatch.PermissionResolver¶

Resolves detected outside-path requests to allow/deny decisions. The app supplies one that pauses the run for user approval (interrupt, ADR-0046); with no resolver (the default) every request is denied.

alias of Callable[[list[PathRequest]], PermissionResolution]

klea_utils.mcp.dispatch.TOOL_CALL_TIMEOUT_ENV_VAR = 'KLEA_TOOL_CALL_TIMEOUT'¶: Process environment variable overriding the backstop (seconds). A value of 0 or less disables it.

async klea_utils.mcp.dispatch.dispatch_tool_calls(mcp_client: Any, tool_calls: Sequence[ToolCallSchema | tuple[str, Mapping[str, Any]]], tool_infos: dict[str, ToolInfo] | None = None, project_root: str | None = None, access_level: Literal['read_only', 'full'] = 'full', call_timeout: float | None = None, *, allowed_dirs: Sequence[str] | None = None, allowed_files: Sequence[str] | None = None, permission_resolver: Callable[[list[PathRequest]], PermissionResolution] | None = None, session_id: str | None = None, mcp_session_pool: Any | None = None, client_context: ClientContext | None = None, meta: Mapping[str, Any] | None = None) → list[fastmcp.client.client.CallToolResult][source]¶

Gate and dispatch tool calls against an MCP server.

For each call a gate runs before it reaches the server, each producing a synthetic non-halting error when rejected:

an invalid tool name (empty, or – when tool_infos is given – not in the disclosed catalogue) is rejected without contacting the server, so a hallucinated name cannot be dispatched; and
the permission gate (klea_utils.mcp.path_detect.detect_path_requests()), which finds the outside-project paths a call would touch (declared checkpaths, picker-declared paths, and heuristics); permission_resolver turns those into allow/deny decisions (the app pauses for user approval) and any request left unapproved blocks the call; and
the tool access level (ADR-0037): a tool the level does not permit (read_only permits only explicitly read-only, non-destructive tools) is rejected by check_tool_access().

Allowed calls are dispatched in parallel, and the returned list stays aligned with the input tool_calls order.

Parameters:

mcp_client – MCP client used for call_tool. Its reentrant context is entered/exited by this helper. Typed as Any because fastmcp.Client.call_tool has a complex signature (optional arguments, keyword-only raise_on_error, CallToolResult | ToolTask return) that a structural protocol would not cleanly match; tests substitute a fake implementing the subset used here.
tool_calls – The calls to invoke, as ToolCallSchema (so the picker’s declared paths travel) or legacy (name, args) pairs.
tool_infos – Mapping of tool name to its ToolInfo, carrying both the annotated read_only/destructive capability and (in meta) the checkpaths declarations. None disables both gates (backward compatible); for read_only a tool missing from the map is denied (fail-closed).
project_root – Boundary directory for the path gate. Defaults to the current working directory.
access_level – Active tool access level (ADR-0037).
call_timeout – Per-call wall-clock timeout in seconds, as a backstop against a hung tool. None (the default) resolves from TOOL_CALL_TIMEOUT_ENV_VAR / DEFAULT_TOOL_CALL_TIMEOUT_SECONDS; 0 or less disables the backstop. A timed-out call becomes a synthetic non-halting error.
allowed_dirs – Directories already approved for the session (BaseGraphSchema.allowed_dirs), checked alongside project_root.
allowed_files – Sensitive files already approved for the session (BaseGraphSchema.allowed_files).
permission_resolver – Called with the detected requests before dispatch; returns a PermissionResolution. The app supplies one that pauses for user approval; when it is None the outside paths are denied (the pre-ADR-0007 behaviour) and the sensitive second round is not run.
session_id – Chat-thread identifier used to select a pooled MCP client for this dispatch. When given together with mcp_session_pool the per-session client from mcp_session_pool.get(session_id, client_context) is used instead of mcp_client, so concurrent chats do not share one MCP session. None (the default) keeps the shared mcp_client, which is what tests and non-pooled callers use.
mcp_session_pool – Optional session pool exposing get(session_id, client_context). Ignored when session_id is None.
client_context – Optional client-supplied environment forwarded to the per-session pooled client. Ignored without a pool/session id; the shared mcp_client always uses its own spawn configuration.
meta – Optional metadata forwarded to client.call_tool as its meta argument. Only whether a value was supplied is logged, never the value itself.

Returns:

One CallToolResult per input call, in input order.

klea_utils.mcp.dispatch.resource_key(call: ToolCallSchema, tool_infos: dict[str, ToolInfo] | None) → frozenset[str][source]¶

Return the resource identifiers a tool call touches (ADR-0041).

Best-effort and deterministic: reads the tool’s declared checkpaths (from ToolInfo.checkpaths or the folded ToolInfo.meta['checkpaths']) and normalises the string value of each declared argument lexically with os.path.normpath. The caller uses this to serialise calls that share a resource, so a missed dependency edge cannot cause a lost update.

An empty set means no identifiable resource: the tool is unknown, declares no checkpaths, or the declared argument is absent/not a string. Such a call is left to run concurrently.

Parameters:

call – The bound tool call.
tool_infos – Mapping of tool name to ToolInfo, or None.

Returns:

A frozenset of normalised resource identifiers (possibly empty).

klea_utils.mcp.dispatch.tool_call_timeout_seconds() → float | None[source]¶

Return the per-call tool timeout in seconds, or None when disabled.

Reads TOOL_CALL_TIMEOUT_ENV_VAR; an unset, non-numeric, or non-finite value falls back to DEFAULT_TOOL_CALL_TIMEOUT_SECONDS, while a non-positive value disables the backstop.

Returns:: The timeout ceiling, or None when disabled.

Path permissions¶

Layered path discovery, the sensitive-file matcher, and the payload/response schema for the interactive allow / deny approval loop (ADR-0007). The gate runs in klea_utils.mcp.dispatch before a tool call reaches the server.

Layered path discovery for the client-side tool permission gate (ADR-0007).

The gate needs to know which filesystem paths a tool call would touch before it dispatches. Declared checkpaths alone only cover Klea-authored tools, so discovery is layered and tool-agnostic:

declared – the tool’s declared checkpaths argument values;
expected – paths the tools picker declared for the call;
guess – heuristic: argument names that look path-ish, plus path-like values (a shell command string is split with shlex.split() first).

Only paths that resolve outside the permitted roots (project_root plus any session-approved allowed_dirs) become PathRequest objects. This is advisory: a tool can ignore its arguments and touch any path at runtime; only OS sandboxing confines it (ADR-0007).

See devdocs/system/mcp-permissions.md.

File: klea_utils/mcp/path_detect.py

klea_utils.mcp.path_detect.Confidence¶

Confidence that a discovered string really is a path, strongest first.

alias of Literal[‘declared’, ‘expected’, ‘guess’]

klea_utils.mcp.path_detect.MAX_PATH_DISPLAY_CHARS = 240¶: Display caps for permission prompts. The full path and approval key stay in the request; only the rendered labels are bounded so a pathological guessed string cannot flood the ask.

class klea_utils.mcp.path_detect.PathRequest(*, tool: str, argument: str, raw: str, path: str, directory: str, confidence: ~typing.Literal['declared', 'expected', 'guess'], source: str, kind: ~typing.Literal['outside', 'sensitive'] = 'outside', argument_preview: str = '', covered_paths: list[str] = <factory>)[source]¶

Bases: BaseModel

One path a tool call would touch that needs approval.

raw is the string as discovered; path is its resolved location; directory is the directory to approve (the path itself when it is a directory, else its parent). confidence and source describe how it was found, so the prompt can show a guessed path differently from a declared one.

kind is outside (the path is outside the permitted roots) or sensitive (a credential-bearing file inside them). outside is approved per directory, sensitive per file (approval_key).

argument_preview is a bounded, token-centred preview of the full argument value, used by the approval UI. path and directory remain full values for decisions and auditing; use the display helpers when rendering them.

covered_paths lists every distinct resolved path that maps to this request’s approval_key (all files in an approved directory, or the single sensitive file). It lets the ask show what one session approval covers.

property approval_key: str¶

Return the key a decision for this request is matched on.

The resolved file for a sensitive request, the resolved directory for an outside one.

directory_display() → str[source]¶: Return a bounded display label for the approval directory.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path_display() → str[source]¶: Return a bounded display label for the resolved path.

raw_preview() → str[source]¶: Return a bounded display label for the discovered token.

klea_utils.mcp.path_detect.declared_checkpaths(tool_info: ToolInfo | None) → list[str][source]¶

Return the tool’s declared path-argument names.

Reads ToolInfo.checkpaths or the folded ToolInfo.meta['checkpaths'] (set by register_tools); empty when the tool declares none.

Parameters:: tool_info – Tool metadata, or None.
Returns:: Declared path-argument names.

klea_utils.mcp.path_detect.detect_path_requests(tool: str, arguments: Mapping[str, Any], tool_info: ToolInfo | None = None, *, project_root: str | PathLike | None = None, allowed_dirs: Sequence[str] | None = None, allowed_files: Sequence[str] | None = None, picker_paths: Sequence[str] | None = None, include_sensitive: bool = False) → list[PathRequest][source]¶

Return the paths a tool call would touch that need approval.

Layered discovery (see the module docstring) unions declared checkpaths values, picker-declared paths, and heuristics. Two rounds become requests: paths outside the permitted roots (kind="outside", approved per directory) and, when include_sensitive is set, sensitive files inside them (kind="sensitive", approved per file). Requests are de-duplicated by PathRequest.approval_key, keeping the strongest confidence tier.

Parameters:

tool – Tool name (recorded on each request).
arguments – The arguments the call would pass.
tool_info – Tool metadata (for checkpaths), or None.
project_root – Boundary directory; defaults to the current directory.
allowed_dirs – Additional permitted directories (session approvals).
allowed_files – Sensitive files already approved for the session.
picker_paths – Paths the picker declared for this call, or None.
include_sensitive – Also request sensitive files inside the roots (the permission second round).

Returns:

Approval requests, sorted by approval key.

klea_utils.mcp.path_detect.platform_is_windows() → bool[source]¶

Return whether arguments should be split with Windows quoting rules.

Backslashes are literal path separators on Windows, so POSIX shlex escape processing (a\ b) must not be applied there. Kept as a function so tests can pin either behaviour on any platform.

Returns:: True when running on Windows.

klea_utils.mcp.path_detect.resolve_path_arguments(arguments: Mapping[str, Any], tool_info: ToolInfo | None, *, project_root: str | PathLike | None = None) → dict[str, Any][source]¶

Return arguments with relative filesystem paths made absolute.

The tools run in the MCP subprocess, whose working directory is fixed at spawn, so a per-session project root cannot reach them via chdir. Resolving relative path arguments to absolute against the session root here lets any local tool operate under the root regardless of its own cwd, and works for third-party tools too (it does not depend on the tool being Klea-authored).

Rules:

a declared checkpaths argument is always resolved (string or list[str] value);
other arguments are resolved heuristically when the argument name is path-ish (klea_utils.tool_args.is_path_argument_name()) and the value is a single-token, non-URL/non-flag/non-glob string (multi-token shell command strings are never rewritten);
working_directory (the run_command convention) defaults to the session root when unset;
absolute paths and URLs are left unchanged.

Parameters:

arguments – The call’s bound arguments.
tool_info – The tool’s metadata (for checkpaths), or None.
project_root – Session root; defaults to the process working directory.

Returns:

A copy of arguments with relative path values made absolute.

klea_utils.mcp.path_detect.split_arguments(text: str, *, windows: bool | None = None) → list[str][source]¶

Split text into shell-like argument tokens for the current platform.

On POSIX this is shlex.split() unchanged: quotes group tokens, backslash escapes are honoured, and unbalanced quotes raise ValueError. On Windows the split is non-POSIX, so backslashes stay literal (C:\Users\me\a.txt is a single token); one enclosing quote pair is stripped from each token, and unbalanced quotes still raise ValueError.

Parameters:

text – The argument string to split.
windows – Force Windows (True) or POSIX (False) rules; defaults to platform_is_windows().

Returns:

The argument tokens.

klea_utils.mcp.path_detect.split_shell_tokens(text: str, *, windows: bool | None = None) → list[str][source]¶

Split a shell command into tokens, separating control operators.

Unlike split_arguments() (used for /command arguments, where a ; may be part of a value), this uses shlex’s punctuation mode so ;, |, &, < and > are separate tokens instead of clinging to a path. Quotes still group a token, so a quoted path with spaces or shell characters stays intact. Comments are disabled (as in shlex.split()): # is a literal character, not a comment marker. whitespace_split is enabled so =, : and # do not split a token; only whitespace and the punctuation characters do.

Parameters:

text – The shell command string.
windows – Force Windows (True) or POSIX (False) rules; defaults to platform_is_windows().

Returns:

The argument and operator tokens.

klea_utils.mcp.path_detect.truncate_middle(text: str, max_chars: int = 240) → str[source]¶

Truncate text in the middle, keeping its start and end.

Used for display only. The start (root/directory) and the end (filename) are the useful parts of a path; the omitted span is replaced with an explicit marker so a pathological guessed value is recognisable.

Parameters:

text – The text to bound.
max_chars – Maximum number of visible characters to keep.

Returns:

The text unchanged when short, otherwise a middle-truncated preview with an omission marker.

Session-scoped path-approval helpers.

Pure helpers for the allowed_dirs / allowed_files state used by the path gate (ADR-0007) and the /paths command (ADR-0047): resolve and validate directory approvals, normalise the edited lists, find the entry that covers a path, and format the user-facing listing. Command-specific parsing and messages stay in the app command handler.

See devdocs/system/mcp-permissions.md and system/session-commands.md.

File: klea_utils/mcp/path_approval.py

klea_utils.mcp.path_approval.entry_matches_path(entry: str, resolved: Path) → bool[source]¶: Return whether an approval entry resolves to resolved.

klea_utils.mcp.path_approval.find_covering_dir(allowed_dirs: list[str], resolved: Path) → str | None[source]¶: Return the allowed directory that covers resolved, if any.

klea_utils.mcp.path_approval.find_matching_entry(entries: list[str], resolved: Path) → str | None[source]¶: Return the exact approval entry matching resolved, if any.

klea_utils.mcp.path_approval.format_allowed_paths(project_root: str | PathLike | None, allowed_dirs: list[str], allowed_files: list[str]) → str[source]¶: Return the user-facing permitted-path listing.

klea_utils.mcp.path_approval.is_relative_to(path: Path, parent: Path) → bool[source]¶: Return whether path is inside parent.

klea_utils.mcp.path_approval.normalise_allowed_dirs(project_root: str | PathLike | None, values: list[Any]) → tuple[list[str], list[str]][source]¶

Validate edited directory values.

Parameters:

project_root – The session project root.
values – The submitted directory values.

Returns:

(accepted, ignored); ignored holds the validation errors.

klea_utils.mcp.path_approval.normalise_allowed_files(current: list[str], values: list[Any]) → list[str][source]¶

Keep only existing sensitive-file approvals that remain selected.

The editor is additive-disabled for sensitive files: new values are ignored, removals are honoured.

Parameters:

current – The current allowed_files entries.
values – The submitted values.

Returns:

The remaining entries, in their original order.

klea_utils.mcp.path_approval.normalise_directory(project_root: str | PathLike | None, value: str | PathLike) → tuple[str | None, str][source]¶

Validate one directory approval.

The directory must exist, must not be inside the project root (which is always permitted), and is returned resolved.

Parameters:

project_root – The session project root.
value – The user-supplied directory.

Returns:

(resolved, "") on success, or (None, error).

klea_utils.mcp.path_approval.resolve_project_root(project_root: str | PathLike | None = None) → Path[source]¶

Return the resolved session project root.

Falls back to the current directory when project_root is unset or cannot be resolved.

Parameters:: project_root – The session root, or None.
Returns:: A resolved directory path.

klea_utils.mcp.path_approval.resolve_user_path(value: str | PathLike, base: str | PathLike) → Path[source]¶

Resolve value relative to base without requiring it to exist.

Parameters:

value – The user-supplied path.
base – The directory relative paths resolve against.

Returns:

A resolved path.

Sensitive-file matcher for the client-side permission gate (ADR-0007).

The path-boundary check (path_detect) only asks about paths outside the permitted roots. This module is the second round: a curated, best-effort matcher for files that commonly hold credentials (env files, private keys, cloud config) so the gate can ask before a tool reads one that is inside the project root.

It is advisory: a renamed secret, a secret inline in source, or a directory scan (grep -r) can slip past a filename matcher. Operators can extend the list with the comma-separated KLEA_SENSITIVE_PATTERNS env var (extra fnmatch globs on the basename).

See devdocs/system/mcp-permissions.md.

File: klea_utils/mcp/sensitive.py

klea_utils.mcp.sensitive.SENSITIVE_ENV_VAR = 'KLEA_SENSITIVE_PATTERNS'¶: Env var with extra comma-separated basename globs to treat as sensitive.

klea_utils.mcp.sensitive.is_sensitive(path: str | PathLike) → bool[source]¶

Return whether path is a well-known credential-bearing file.

Matches on the basename (exact names and globs), the suffix, and any directory in the path (_SENSITIVE_DIRS), plus the operator’s extra patterns from SENSITIVE_ENV_VAR.

Parameters:: path – Path to test (need not exist).
Returns:: True when the path looks sensitive.

HITL payload and response schema for interactive path approval (ADR-0007).

The client-side permission gate collects outside-project PathRequest objects, pauses the run with an interrupt carrying build_permission_payload(), and resolves the user’s per-path allow/deny decision with resolve_decisions(). The interrupt mechanism itself is the shared LangGraph one (ADR-0046); this module owns only the permission-specific payload, response schema, and decision mapping.

See devdocs/system/mcp-permissions.md.

File: klea_utils/mcp/permission_hitl.py

klea_utils.mcp.permission_hitl.DEFAULT_QUESTION = 'The agent wants to access paths outside the project directory. Allow or deny each one.'¶: Default question shown with a permission ask.

klea_utils.mcp.permission_hitl.Decision¶

allow for this dispatch only, allow for the session (persisted in allowed_dirs), deny, or dismiss the request as “not a path” (a heuristic false positive; the call proceeds for this dispatch and nothing is remembered).

Type:: A per-path decision

alias of Literal[‘once’, ‘session’, ‘deny’, ‘not_a_path’]

class klea_utils.mcp.permission_hitl.PathDecision(*, key: str, decision: Literal['once', 'session', 'deny', 'not_a_path'] = 'deny')[source]¶

Bases: BaseModel

A user’s decision for one requested path.

key matches a request’s approval_key (the directory for an outside request, the file for a sensitive one); decision is once (this dispatch), session (also persisted for the thread), deny, or not_a_path (dismiss a guessed false positive for this dispatch; the call proceeds).

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class klea_utils.mcp.permission_hitl.PermissionResolution(*, allowed_now: list[str] = <factory>, allowed_session: list[str] = <factory>, denied: list[str] = <factory>, not_a_path: list[str] = <factory>)[source]¶

Bases: BaseModel

Resolved decisions for one permission ask.

Fields hold approval keys (a directory for an outside request, a file for a sensitive one). allowed_now is allowed for this dispatch only; allowed_session is also persisted for the thread; denied is refused (the gate turns each into a non-halting denial); not_a_path is dismissed for this dispatch only – the request was a heuristic false positive (the user is not approving it), so the call proceeds without gating it, and no memory is kept.

effective_keys() → list[str][source]¶: Return the approval keys allowed for the current dispatch.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class klea_utils.mcp.permission_hitl.PermissionResponse(*, action: Literal['answer', 'cancel']='answer', decisions: list[PathDecision] = <factory>)[source]¶

Bases: BaseModel

Resume payload for a permission ask (ADR-0007).

action="cancel" denies every request. action="answer" carries one PathDecision per requested directory; a requested directory with no decision is denied (fail closed).

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

klea_utils.mcp.permission_hitl.build_permission_payload(requests: Sequence[PathRequest], *, question: str = '') → dict[str, Any][source]¶

Build the interrupt payload for a list of path requests.

The payload is a plain mapping (the stream/interrupt convention): kind is "permission", question is the prompt text, and requests is the list of serialized PathRequest objects for the client to render. Each request carries the full path and directory for decisions plus bounded path_display, directory_display, raw_preview and argument_preview fields for the UI; the payload’s raw value is capped to raw_preview.

Parameters:

requests – The outside-path requests to ask about.
question – Custom prompt text; defaults to DEFAULT_QUESTION.

Returns:

The interrupt payload dict.

klea_utils.mcp.permission_hitl.resolve_decisions(requests: Sequence[PathRequest], response: Any) → PermissionResolution[source]¶

Resolve a resume value into allow-now / allow-session / deny key sets.

Uses each request’s approval_key (the directory for an outside request, the file for a sensitive one). A cancel, an invalid response, or a request with no matching decision denies that request (fail closed).

Parameters:

requests – The requests that were asked about.
response – The value returned by the interrupt (a PermissionResponse, a raw mapping, or None).

Returns:

The PermissionResolution.

Tool access levels¶

Tool access levels for MCP tool invocation (ADR-0037).

Classification is derived from the standard MCP ToolAnnotations (readOnlyHint / destructiveHint) carried on ToolInfo, with an optional operator override (ToolAccessOverride). Levels are read_only (only explicitly read-only, non-destructive tools) and full (every tool); an unannotated tool fails closed in read_only. See devdocs/system/mcp-permissions.md and devdocs/adr/0037-tool-access-levels.md.

File: klea_utils/mcp/access.py

klea_utils.mcp.access.AccessLevel¶

Tool invocation access levels (ADR-0037).

alias of Literal[‘read_only’, ‘full’]

klea_utils.mcp.access.DEFAULT_ACCESS_LEVEL: Literal['read_only', 'full'] = 'full'¶

every tool is permitted (the pre-ADR-0037 behaviour).

Type:: Default level

class klea_utils.mcp.access.ToolAccessOverride(*, read_only: bool | None = None, destructive: bool | None = None)[source]¶

Bases: BaseModel

Operator declaration of a tool’s capability (ADR-0037).

Configured per tool under general.tool_access and applied over the tool’s own MCP annotations. Either field may be left None to leave the annotated value unchanged; the operator who writes the config is the trust root for the declaration.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

klea_utils.mcp.access.check_tool_access(tool: str, read_only: bool | None, destructive: bool | None, access_level: Literal['read_only', 'full']) → str | None[source]¶

Return a denial message when tool is not permitted, else None.

Mirrors tool_permits() but returns a human-readable message for the non-halting synthetic error result the caller produces (the ADR-0007 error contract) instead of raising.

Parameters:

tool – Tool name, used in the message.
read_only – Tool’s read-only capability.
destructive – Tool’s destructive capability.
access_level – The active access level.

Returns:

A denial message, or None when the tool is permitted.

klea_utils.mcp.access.filter_tools_info(tools_info: dict[str, dict[str, ToolInfo]], access_level: Literal['read_only', 'full']) → dict[str, dict[str, ToolInfo]][source]¶

Return tools_info with tools disallowed by access_level removed.

Domains are preserved as keys even when all their tools are filtered out. full returns the input unchanged; otherwise a new mapping is built.

Parameters:

tools_info – Per-domain {name: ToolInfo} mapping.
access_level – The active access level.

Returns:

A filtered mapping (the input is not mutated).

klea_utils.mcp.access.resolve_access_level(value: Any) → Literal['read_only', 'full'][source]¶

Coerce a value to a valid access level, defaulting to full.

Used for untrusted inputs (env var strings); an unrecognised value is logged and replaced with DEFAULT_ACCESS_LEVEL.

Parameters:: value – Candidate value (e.g. from a request component or env).
Returns:: The matching level, or DEFAULT_ACCESS_LEVEL.

Apply an operator override over a tool’s annotated capability.

Precedence is explicit tool_access override > tool annotation > None (which fails closed downstream). A None override field leaves the annotated value unchanged.

Parameters:

read_only – Annotated read-only hint, or None.
destructive – Annotated destructive hint, or None.
override – Operator override, or None.

Returns:

The effective (read_only, destructive) pair.

klea_utils.mcp.access.tool_permits(read_only: bool | None, destructive: bool | None, access_level: Literal['read_only', 'full']) → bool[source]¶

Return whether access_level permits invoking this tool.

full permits every tool. read_only permits a tool only when it is explicitly read-only and not destructive; an unannotated tool (None) is not permitted (fail-closed).

Parameters:

read_only – Tool’s read-only capability.
destructive – Tool’s destructive capability.
access_level – The active access level.

Returns:

True when the tool may be invoked.

Strict-safe call schema¶

Strict-safe tool-call schema builder for the tools picker.

Builds the picker’s structured-output schema dynamically from the disclosed tools, so each tool’s parameters become real, typed fields instead of a free-form args object. A free-form object is closed by every provider’s strict structured-output mode (additionalProperties: false), which silently forces args to {}; a discriminated union of per-tool call models avoids that and enforces each tool’s required parameters.

The generated schema is:

ToolPickerOutput - {tool_calls: list[ToolCall]}
ToolCall - {step, reason, call: <union>} (nested)
one <tool>_call model per tool - {tool: Literal["<name>"], <params>}
a NoTool branch - {tool: Literal["no_tool"], reason} for the “no suggested tool can carry out this step” signal.

Type mapping (strict-safe): string/integer/number/boolean map directly; anyOf (including T | null) maps to a union; arrays of supported types map to list[...]; object/unknown parameters fall back to a JSON-encoded str (documented). No generated field is an open object.

File: klea_utils/mcp/call_schema.py

klea_utils.mcp.call_schema.NO_TOOL_TAG = 'no_tool'¶: Discriminator tag for the “no suggested tool fits” branch. _update_state maps it to the picker’s empty-tool failure signal.

klea_utils.mcp.call_schema.build_tool_call_schema(tools: Mapping[str, ToolInfo]) → type[BaseModel][source]¶

Build the picker’s structured-output schema for a tool set.

The schema is a nested discriminated union over the given tools plus a NoTool branch. Tool order is deterministic (sorted by name) so the serialized schema - and therefore the prompt’s Output schema block - is stable for a given tool set (prompt caching).

Parameters:: tools – The disclosed tools ({name: ToolInfo}), already filtered by access level and, for RAG, by the run’s domains.
Returns:: A pydantic model class with a tool_calls field.

Privilege guard¶

Privilege guard for Klea-authored MCP tools.

Klea-authored tools (the bundled server and the NeuroML server) refuse to run when the server process has root privileges, unless the operator explicitly opts in via an environment variable. This is defense-in-depth for deployments that accidentally run the server as uid 0 (including containers, which default to root): a tool running as root can read/write/execute far beyond the project. Third-party MCP servers do not use this guard; their privileges are the operator’s choice (ADR-0007 trust model). See ADR-0038.

File: klea_utils/mcp/privilege.py

klea_utils.mcp.privilege.ALLOW_ROOT_ENV_VAR = 'KLEA_ALLOW_ROOT_TOOLS'¶: Process environment variable that allows Klea-authored tools to run as root. Must be truthy (1/true/yes/on); anything else (or unset) refuses root execution. A process env var, so it must reach the MCP server subprocess (it is inherited from the app’s environment).

klea_utils.mcp.privilege.allow_root_tools() → bool[source]¶

Return whether root execution is explicitly allowed.

Reads ALLOW_ROOT_ENV_VAR; only a truthy value (1/true/yes/on, case-insensitive) opts in.

Returns:: True when the override is set to a truthy value.

klea_utils.mcp.privilege.root_denial_message() → str[source]¶

Return the non-halting denial message for a root-refused tool call.

Returns:: A message naming the override variable.

klea_utils.mcp.privilege.running_as_root() → bool[source]¶

Return whether the current process has root (effective uid 0).

Windows-safe: non-POSIX systems have no geteuid and are treated as non-root.

Returns:: True when os.geteuid() == 0.

klea_utils.mcp.privilege.warn_if_root(logger: Logger) → None[source]¶

Log a warning when the current process runs as root.

Called once at app/server startup so operators are aware that tools will either be refused or execute with full root privileges.

Parameters:: logger – Logger to emit the warning on.

Tool results¶

Helper to map Klea dict results with an error field to MCP isError.

File: klea_utils/mcp/tool_result.py

klea_utils.mcp.tool_result.to_result(result: dict[str, Any]) → fastmcp.tools.ToolResult[source]¶

Return a ToolResult with is_error derived from the error field.

Klea tool implementations are framework-agnostic and signal failure via a non-empty error (or Error for legacy repository tools) string in their returned dict. The MCP spec requires such failures to be reported as isError: true in the CallToolResult (see mcp/types.py:CallToolResult and the spec section server/tools#Error Handling); FastMCP only sets this when the tool explicitly returns ToolResult(is_error=True) or raises ToolError (see fastmcp/tools/base.py:convert_result and mcp/server/lowlevel/server.py:_make_error_result).

This helper bridges the two: it preserves the dict as structured_content (so LLM remediation keeps full context) while marking is_error when the error/Error field is truthy. It intentionally preserves the structured payload rather than raising ToolError, which would strip it.

Parameters:: result – Tool return dict containing at least an error key; legacy tools may use Error (capitalised).
Returns:: ToolResult with is_error reflecting the error field.

HTTP session lifespan¶

Shared FastMCP lifespan helpers.

See devdocs/adr/0005-httpx-single-stack.md.

File: klea_utils/mcp/lifespan.py

klea_utils.mcp.lifespan.make_http_session_lifespan(session_key: str = 'http_session')[source]¶

Create a FastMCP lifespan that provides a shared httpx session.

Tools that need an HTTP session (e.g. klea_utils.mcp.tool_impls.web_fetch) read it from ctx.lifespan_context[<session_key>] in their MCP wrapper. Lifespans are composable with the | operator.

Parameters:: session_key – Lifespan context key under which the session is stored.
Returns:: A FastMCP @lifespan-decorated function.

Shared tool implementations¶

Framework-agnostic tool bodies that apps wrap into FastMCP tools, passing their httpx session via the lifespan context.

Path boundary helpers for the client-side permission gate.

The tool implementations no longer call these; the client-side gate uses resolve_path() / path_is_allowed() / permitted_roots() (klea_utils.mcp.path_detect) and check_path_access() remains a convenience that raises on denial. See devdocs/system/mcp-permissions.md and devdocs/adr/0007-mcp-permissions.md.

Containment is path-based and taken as a snapshot. Two known limitations: a hard link inside the root to an inode whose other names are outside it is not detectable path-wise and is allowed; and a path swapped for an external symlink after the check (TOCTOU) could still be opened. Hardening would need a per-open check (openat/O_NOFOLLOW).

File: klea_utils/mcp/tool_impls/permission.py

klea_utils.mcp.tool_impls.permission.check_path_access(path: str | PathLike, project_root: str | PathLike | None = None, *, allowed_dirs: Sequence[str] | None = None) → None[source]¶

Raise PermissionDeniedError when path is not permitted.

path is allowed when it resolves inside project_root (default: the current working directory) or inside any of allowed_dirs. Both sides are fully resolved first, so .. traversal and symlink escapes outside the boundary are caught.

This is the utility used by the client-side tool-call gate (klea_utils.mcp.path_detect); it is no longer called inside tool implementations (ADR-0007 update 2026-10-08).

Parameters:

path – File or directory path the tool wants to access.
project_root – Boundary directory inside which access is allowed. Defaults to the current working directory.
allowed_dirs – Additional directories permitted for this check (session approvals), or None.

Raises:

PermissionDeniedError – when path resolves outside every root, or cannot be resolved at all (for example a symlink loop).

klea_utils.mcp.tool_impls.permission.path_is_allowed(path: Path, roots: list[Path]) → bool[source]¶

Return whether resolved path is inside any of roots.

Parameters:

path – An already-resolved path.
roots – Resolved boundary roots (see permitted_roots()).

Returns:

True when path is inside at least one root.

klea_utils.mcp.tool_impls.permission.permitted_roots(project_root: str | PathLike | None = None, allowed_dirs: Sequence[str] | None = None) → list[Path][source]¶

Return the resolved roots a path is allowed to be inside.

All are fully resolved, so .. traversal and symlink escapes are caught against every root. Blank or unresolvable allowed_dirs entries are skipped (fail open for the root, not the path).

Parameters:

project_root – Primary boundary; defaults to the current directory.
allowed_dirs – Additional roots, for example directories a user approved for the session (ADR-0007 update 2026-10-08).

Returns:

A list of resolved boundary roots.

klea_utils.mcp.tool_impls.permission.resolve_path(path: str | PathLike) → Path | None[source]¶

Resolve path to an absolute, symlink-free path, or None.

Returns None for an empty path or one that cannot be resolved (for example a symlink loop), so callers can fail closed. A missing target is resolved non-strictly so the boundary check still applies to its location.

Parameters:: path – Path to resolve.
Returns:: The resolved path, or None.

Shared session protocol for MCP tool implementations.

File: klea_utils/mcp/tool_impls/session.py

class klea_utils.mcp.tool_impls.session.SessionLike(*args, **kwargs)[source]¶

Bases: Protocol

Minimal HTTP session interface needed by the shared tool implementations.

Kept structural so tests can substitute a fake and so the implementations do not depend on a specific HTTP client library. Matches the subset of httpx.AsyncClient that the bundled tools use (get for downloads, stream for page fetches).

SSRF (Server-Side Request Forgery) protection for outbound HTTP tools.

File: klea_utils/mcp/tool_impls/ssrf.py

klea_utils.mcp.tool_impls.ssrf.check_ssrf(url: str) → str | None[source]¶

Return an error message if url resolves to a private/internal host.

Resolves the hostname and rejects the request when any resolved address is private, loopback, link-local, reserved, or multicast. Returns None when the request is allowed.

Note

This is the synchronous, blocking variant (uses socket.getaddrinfo directly). Call check_ssrf_async() from async code to avoid stalling the event loop.

Parameters:: url – Absolute URL to check.
Returns:: An error message describing the denial, or None when the URL is allowed.

async klea_utils.mcp.tool_impls.ssrf.check_ssrf_async(url: str, timeout: float = 5.0) → str | None[source]¶

Async wrapper around check_ssrf() that offloads DNS to a thread.

socket.getaddrinfo is blocking; running it in asyncio.to_thread keeps the event loop responsive and adds a timeout.

Parameters:

url – Absolute URL to check.
timeout – Seconds to wait for DNS before returning a timeout error.

Returns:

Error message or None when allowed.

klea_utils.mcp.tool_impls.ssrf.is_private_or_reserved(ip: IPv4Address | IPv6Address) → bool[source]¶

Return True for addresses an SSRF guard should refuse to fetch.

Blocks loopback, private (RFC1918/ULA), link-local (incl. the cloud metadata address 169.254.169.254), reserved, and multicast ranges.

Parameters:: ip – Address to classify.

Honest Klea User-Agent strings for the shared MCP tool implementations.

Each tool family identifies itself with a distinct prefix (klea-web-fetch/, klea-paper-search/), so a server operator can tell which Klea subsystem is calling and polite-pool/abuse handling stays attributed. The Klea version is read once from package metadata.

File: klea_utils/mcp/tool_impls/user_agent.py

klea_utils.mcp.tool_impls.user_agent.klea_user_agent(prefix: str) → str[source]¶

Return an honest <prefix><klea_utils version> User-Agent string.

Parameters:: prefix – Tool-family prefix, e.g. "klea-web-fetch/". Keep the trailing slash.
Returns:: The User-Agent string, cached per prefix.

Web fetch implementation for Klea MCP tools.

File: klea_utils/mcp/tool_impls/web_fetch.py

klea_utils.mcp.tool_impls.web_fetch.honest_user_agent() → str[source]¶

Return the honest client User-Agent for web fetches.

E.g. klea-web-fetch/<version>; shared by the download and repository tools that reuse this identity (see klea_utils.mcp.tool_impls.user_agent.klea_user_agent()).

async klea_utils.mcp.tool_impls.web_fetch.web_fetch(session: SessionLike | None, url: str, timeout: float = 30.0, max_chars: int = 100000, retries: int = 3, max_download_bytes: int = 5000000, allow_internal_hosts: bool = False) → dict[str, Any][source]¶

Fetch a URL and return its text content.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool that supplies session from their lifespan context (see klea_utils.mcp.lifespan).

Transient failures (timeouts, connection errors, HTTP 5xx/429) are retried with exponential backoff. HTTP 4xx errors are returned as error results and not retried. The raw response body is capped at max_download_bytes and the returned text at max_chars; each cap is reported via its own flag.

Parameters:

session – HTTP session to use for the request. None when no session is available.
url – HTTP or HTTPS URL to fetch.
timeout – Request timeout in seconds.
max_chars – Maximum number of characters of content to return.
retries – Number of attempts for transient failures.
max_download_bytes – Maximum number of raw response bytes to read.
allow_internal_hosts – Skip the SSRF guard (requests to loopback, private, link-local, or reserved addresses).

Returns:

dict with url, status_code, content_type, content, truncated, download_truncated, error.

File listing implementation for Klea MCP tools.

See devdocs/system/mcp-permissions.md.

File: klea_utils/mcp/tool_impls/list_files.py

klea_utils.mcp.tool_impls.list_files.list_files(path: str = '.', max_depth: int | None = None, pattern: str = '*', include_files: bool = True, include_directories: bool = True, recursive: bool = False, max_results: int = 100, project_root: str | None = None) → dict[str, Any][source]¶

List files and directories with filtering and metadata.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool (see klea_utils.mcp.registry).

Parameters:

path – Directory path to list. Must be relative to current working directory and cannot contain ‘..’ for security. Defaults to the current directory (".").
max_depth – Maximum directory depth to traverse. 1 lists the immediate entries inside path, 2 also descends one directory deeper, and so on. None for unlimited.
pattern – Space separated file patterns to filter based on file type.
include_files – Whether to include files in results.
include_directories – Whether to include directories in results.
recursive – If True, traverse subdirectories recursively.
max_results – Maximum number of entries to return.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with files, error, truncated, note. note is non-empty when the result was cut short by max_results (it then carries the truncation message) and/or when the requested filter matched nothing and the fallback below returned an unfiltered listing instead.

klea_utils.mcp.tool_impls.list_files.missing_target_note(path: str, directory: str, nearby: list[str]) → str[source]¶

Return a “target missing; nearby entries” note for a file-tool result.

Parameters:

path – The requested (missing) path.
directory – The nearest existing directory, or "".
nearby – Entry names found in directory.

Returns:

A one-line, human-readable note.

klea_utils.mcp.tool_impls.list_files.nearby_entries(path: str | Path, project_root: str | None = None, max_results: int = 20) → tuple[str, list[str]][source]¶

Return the nearest existing directory and its entries for path.

Diagnostic helper for the file tools: when a requested target is missing, the tool includes this so the caller can see what actually exists (the same “show what is there” idea as the list_files fallback, reused). Walks up from the target’s parent to the nearest existing ancestor and lists its immediate entries.

Parameters:

path – The requested (possibly missing) path.
project_root – Retained and passed through to list_files().
max_results – Maximum number of entries to return.

Returns:

(directory, names), or ("", []) when no existing ancestor directory could be found or listed.

File reading implementation for Klea MCP tools.

File: klea_utils/mcp/tool_impls/read_file.py

klea_utils.mcp.tool_impls.read_file.read_file(path: str = '.', offset: int = 1, char_offset: int = 0, limit: int | None = 500, max_chars: int = 20000, max_bytes: int = 104857600, line_numbers: bool = True, project_root: str | None = None) → dict[str, Any][source]¶

Read a file and return a slice of its text content.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool (see klea_utils.mcp.server.bundled_tools).

Files are converted to plain text first: HTML is stripped with BeautifulSoup, and office documents/PDF/EPUB/CSV are converted to Markdown with the anydoc library; anything else is read as plain text. For document formats the offsets/limits apply to that converted text, and the returned line_end/total_lines let the caller continue reading a large document in pages.

Plain text must be valid UTF-8 (a leading UTF-8 BOM is stripped). A binary file (containing a NUL byte) or a file that is not valid UTF-8 is refused with an error instead of being returned as garbled text. Lines are delimited by \n after normalising CRLF and lone-CR endings, so the line numbers match editors and wc -l (str.splitlines would also break on form feeds and Unicode separators).

Parameters:

path – File path to read. Defaults to the current directory ("."), which is not a file, so a missing path yields the nearby-entries fallback.
offset – 1-indexed line to start reading from. A negative value counts from the end: -1 is the last line, -N the Nth from the end. Returned line numbers are then negative relative to EOF, and a tail request that reaches past the start clamps to the first line.
char_offset – 0-indexed character position within offset’s line to start from (default 0). Only needed to continue inside a single line longer than max_chars: the returned next_char_offset gives the resume position.
limit – Maximum number of lines to return. None removes only the line limit; the server-owned max_chars budget still applies, so the result can still be truncated (with next_offset). For a negative offset, this counts forward from the tail start.
max_chars – Character budget for the response (server-owned backstop; not exposed to the model), applied after the line slice. The effective value is reported as char_budget.
max_bytes – Per-call scan cap in bytes (server-owned). Plain text is streamed, so the scan stops after this many bytes and the page is returned truncated with a hint; HTML/office/PDF conversions and a file that grows mid-read are still bounded by it.
line_numbers – Prefix each returned line with its line number (default). Set False to return the raw line text, e.g. to copy a span into an edit tool’s old_string.
project_root – Base directory used to locate nearby entries when the target is missing; the boundary itself is enforced by the client-side gate (ADR-0007 update 2026-10-08).

Returns:

dict with path, content, line_start, line_end, total_lines, truncated, truncation_reason, char_budget, next_offset, next_char_offset, error, note (plus nearby on a missing-file error). A truncated result is a success (error empty) that carries the line range actually returned and next_offset for continuing; nearby/note are populated on a missing/not-a-file error so the caller can see what exists instead. total_lines is exact when the read reaches the relevant end (EOF for a forward read, the file start for a tail read); a page that stops early reports it as None. line_start/line_end (and next_offset) are negative for a tail read, with -1 as the last line. char_budget is the effective server character cap, and truncation_reason is "char_budget", "line_limit" or "scan_cap" when truncated (empty otherwise), so a limit=None read that still reports truncated shows why. A truncated page of a large file also carries a note suggesting grep to locate the region.

File-name search for Klea MCP tools.

File: klea_utils/mcp/tool_impls/find_files.py

klea_utils.mcp.tool_impls.find_files.DEFAULT_MAX_RESULTS = 100¶: Default cap on the number of file paths returned.

async klea_utils.mcp.tool_impls.find_files.find_files(pattern: str = '*', path: str = '.', include_ignored: bool = False, max_results: int = 100, project_root: str | None = None) → dict[str, Any][source]¶

List files under a directory whose path matches a glob pattern.

Uses the pinned ripgrep binary when it is installed (the [search] extra), otherwise falls back to find_files_inhouse(). With ripgrep the default honours .gitignore; pass include_ignored to list ignored files too. The in-house fallback has no .gitignore support, so include_ignored has no effect there.

Parameters:

pattern – Glob pattern matched against file paths; "*" lists every file.
path – Directory to search, relative to project_root.
include_ignored – When True, also list .gitignore-ignored files (ripgrep backend only).
max_results – Maximum number of file paths to return.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with files, truncated, error, note. note carries a truncation message when the result was cut short by max_results, and is empty otherwise.

klea_utils.mcp.tool_impls.find_files.find_files_inhouse(pattern: str = '*', path: str = '.', max_results: int = 100, project_root: str | None = None) → dict[str, Any][source]¶

List files under a directory whose path matches a glob pattern.

This is the fallback used when the pinned ripgrep binary is not installed. It has no .gitignore support: only the fixed SKIP_DIRECTORIES set is excluded, so gitignored files are listed.

The pattern is matched against each file’s path relative to path (fnmatch, where * also crosses directory separators, so *.py matches at any depth). Results are in traversal order and not sorted.

Parameters:

pattern – Glob pattern to match against file paths; "*" lists every file.
path – Directory to search.
max_results – Maximum number of file paths to return.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with files, truncated, error, note. note carries a truncation message when the result was cut short by max_results, and is empty otherwise.

Regular-expression content search for Klea MCP tools.

File: klea_utils/mcp/tool_impls/grep.py

klea_utils.mcp.tool_impls.grep.DEFAULT_MAX_FILES = 5000¶: Default cap on the number of files scanned (bounds worst-case runtime).

klea_utils.mcp.tool_impls.grep.DEFAULT_MAX_FILE_BYTES = 5000000¶: Default cap on the size of an individual file considered for searching.

klea_utils.mcp.tool_impls.grep.DEFAULT_MAX_RESULTS = 100¶: Default cap on the number of matching lines returned.

async klea_utils.mcp.tool_impls.grep.grep(pattern: str, path: str = '.', include: str | None = None, case_sensitive: bool = True, include_ignored: bool = False, context_lines: int = 2, max_results: int = 100, project_root: str | None = None) → dict[str, Any][source]¶

Search files under a directory for lines matching a regular expression.

Uses the pinned ripgrep binary when it is installed (the [search] extra), otherwise falls back to grep_inhouse(). With ripgrep the default honours .gitignore; pass include_ignored to search ignored files too. The in-house fallback has no .gitignore support, so include_ignored has no effect there (ignored files are searched regardless) and only the fixed skip-directory set is excluded.

Parameters:

pattern – Regular expression to search for.
path – Directory to search, relative to project_root.
include – Optional space separated glob patterns; when given, only files whose path (relative to path) matches one are searched.
case_sensitive – Whether matching is case sensitive.
include_ignored – When True, also search .gitignore-ignored files (ripgrep backend only).
context_lines – Lines of context before and after each match, clamped to 0..7.
max_results – Maximum number of matching lines to return. Context lines do not count toward this cap.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with pattern, path, match hunks, truncated, files_scanned, error.

klea_utils.mcp.tool_impls.grep.grep_inhouse(pattern: str, path: str = '.', include: str | None = None, case_sensitive: bool = True, context_lines: int = 2, max_results: int = 100, max_files: int = 5000, max_file_bytes: int = 5000000, max_chars: int = 100000, project_root: str | None = None) → dict[str, Any][source]¶

Search files using the in-house walker (no external tools).

This is the fallback used when the pinned ripgrep binary is not installed. It has no .gitignore support: only the fixed SKIP_DIRECTORIES set is excluded, so gitignored files are searched.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap grep() in an MCP tool (see klea_utils.mcp.server.bundled_tools).

The walk skips version control internals, virtual environments, and tool caches, and never follows symlinks (see klea_utils.mcp.tool_impls.walk). Binary files (those containing a NUL byte) and files larger than max_file_bytes are skipped.

Parameters:

pattern – Regular expression to search for.
path – Directory to search, relative to project_root.
include – Optional space separated glob patterns; when given, only files whose path (relative to path) matches one are searched.
case_sensitive – Whether matching is case sensitive.
context_lines – Lines of context before and after each match, clamped to 0..7.
max_results – Maximum number of matching lines to return. Context lines do not count toward this cap.
max_files – Maximum number of files to scan.
max_file_bytes – Maximum size of a file considered for searching.
max_chars – Maximum total characters of matching and context lines to return.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with pattern, path, match hunks, truncated, files_scanned, error.

Shared filesystem walker for read-only search tools.

File: klea_utils/mcp/tool_impls/walk.py

klea_utils.mcp.tool_impls.walk.SKIP_DIRECTORIES: frozenset[str] = frozenset({'.git', '.hg', '.klea-cache', '.mypy_cache', '.nox', '.pytest_cache', '.ruff_cache', '.svn', '.tox', '.venv', '__pycache__', 'node_modules', 'venv'})¶: Directory names never descended into by default. These are version control internals, virtual environments, and tool caches: searching them is almost never what the caller wants, and some (.git, caches) can be large enough to dominate a scan. Matched against the directory basename at every depth; the search root itself is always traversed, even when it matches, since the caller explicitly asked for it.

klea_utils.mcp.tool_impls.walk.iter_files(root: str | PathLike, *, skip_dirs: Collection[str] = frozenset({'.git', '.hg', '.klea-cache', '.mypy_cache', '.nox', '.pytest_cache', '.ruff_cache', '.svn', '.tox', '.venv', '__pycache__', 'node_modules', 'venv'}), project_root: str | PathLike | None = None) → Iterator[Path][source]¶

Yield the files under root, skipping caches and symlinks.

Framework-agnostic helper shared by the read-only search tools (grep, find_files). This is a depth-first walk using os.scandir with symlinks disabled at every step:

directories named in skip_dirs are not descended into;
directory symlinks are not followed (so a link pointing outside root cannot pull external files into the results);
file symlinks are skipped, for the same boundary reason.

The root itself is always traversed, even when its name is in skip_dirs: the caller explicitly asked for it. Unreadable subdirectories are logged and skipped rather than aborting the walk.

This function performs no result capping; callers bound the number of files scanned and matches returned.

Parameters:

root – Directory to walk.
skip_dirs – Directory basenames to skip at any depth.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Raises:

OSError – when root cannot be scanned (e.g. it is a file or does not exist).

Yields:

pathlib.Path for each regular file found.

ripgrep backend for the read-only search tools.

File: klea_utils/mcp/tool_impls/rg_backend.py

klea_utils.mcp.tool_impls.rg_backend.DEFAULT_SEARCH_TIMEOUT_SECONDS = 30.0¶: Default wall-clock cap for one ripgrep invocation.

klea_utils.mcp.tool_impls.rg_backend.RG_DISTRIBUTION = 'astral-dev-toolchain-ripgrep'¶: Distribution that provides a pinned ripgrep binary. Built and published by Astral (https://astral.sh) from a pinned upstream ripgrep commit; see the [search] extra in setup.cfg. We deliberately do not probe PATH: only this known, pinned artifact is used, so the read-only guarantee of the search tools does not depend on the host environment.

klea_utils.mcp.tool_impls.rg_backend.resolve_rg() → str | None[source]¶

Return the path to the pinned ripgrep executable, or None.

Locates the binary via the installed RG_DISTRIBUTION distribution’s RECORD (importlib.metadata), so it works whether the wheel places the executable in the environment’s scripts directory or ships it as package data. No PATH lookup is performed.

The result is cached for the process lifetime: the installed set of distributions does not change while the server runs. Tests can reset it with resolve_rg.cache_clear().

Returns:: Absolute path to the executable when the distribution is installed and provides an rg/rg.exe file; None otherwise.

async klea_utils.mcp.tool_impls.rg_backend.rg_files(rg: str, pattern: str = '*', path: str = '.', include_ignored: bool = False, max_results: int = 100, timeout_seconds: float = 30.0, project_root: str | None = None) → dict[str, Any][source]¶

List files matching a glob with ripgrep (rg --files).

Parameters:

rg – Resolved ripgrep executable (see resolve_rg()).
pattern – Glob pattern matched against file paths; "*" lists all.
path – Directory to search.
include_ignored – When True, list .gitignore-ignored files.
max_results – Maximum number of file paths to return.
timeout_seconds – Wall-clock cap for the invocation.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with files, truncated, error, note. note carries a truncation message when the result was cut short by max_results, and is empty otherwise.

async klea_utils.mcp.tool_impls.rg_backend.rg_grep(rg: str, pattern: str, path: str = '.', include: str | None = None, case_sensitive: bool = True, include_ignored: bool = False, context_lines: int = 2, max_results: int = 100, max_chars: int = 100000, timeout_seconds: float = 30.0, project_root: str | None = None) → dict[str, Any][source]¶

Search file contents with ripgrep, returning grep-shaped results.

Parameters:

rg – Resolved ripgrep executable (see resolve_rg()).
pattern – Regular expression to search for.
path – Directory to search.
include – Space separated glob patterns restricting files searched.
case_sensitive – Whether matching is case sensitive.
include_ignored – When True, search .gitignore-ignored files.
context_lines – Lines of context before and after each match, clamped to 0..7.
max_results – Maximum number of matching lines to return. Context lines do not count toward this cap.
max_chars – Maximum total characters of matching and context lines to return.
timeout_seconds – Wall-clock cap for the invocation.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with pattern, path, match hunks, truncated, files_scanned, error.

Shared filesystem helpers for the file-editing tools.

File: klea_utils/mcp/tool_impls/file_ops.py

klea_utils.mcp.tool_impls.file_ops.BOM = '\ufeff'¶: Byte-order mark, preserved when present.

klea_utils.mcp.tool_impls.file_ops.DEFAULT_MAX_BYTES = 104857600¶: Default cap on the size of a file the editing tools will read or write.

klea_utils.mcp.tool_impls.file_ops.MAX_DIFF_CHARS = 20000¶: Maximum size of the unified diff returned in a tool result.

klea_utils.mcp.tool_impls.file_ops.TRUNCATION_NOTE = 'Results truncated: showing the first {max_results} entries; more exist and were not listed.'¶: Note appended to a result cut short by max_results.

class klea_utils.mcp.tool_impls.file_ops.TextFile(text: str, bom: bool, newline: str, mode: int)[source]¶

Bases: object

A text file’s content plus the write-time properties to preserve.

bom: bool¶: Whether the file began with a UTF-8 BOM (re-applied on write).

mode: int¶: st_mode captured at read time, used to preserve the file mode.

newline: str¶: Dominant line ending, LF or CRLF; re-applied on write.

text: str¶: File text with any leading BOM stripped.

klea_utils.mcp.tool_impls.file_ops.apply_newline(text: str, newline: str) → str[source]¶

Re-apply newline to LF-normalised text.

Parameters:

text – LF-only text.
newline – Target line ending ("\n" or "\r\n").

Returns:

Text with newline endings.

klea_utils.mcp.tool_impls.file_ops.detect_newline(text: str) → str[source]¶

Return the dominant line ending in text.

Parameters:: text – Text to inspect.
Returns:: "\r\n" when the text contains CRLF, else "\n".

klea_utils.mcp.tool_impls.file_ops.diff_counts(old_text: str, new_text: str) → tuple[int, int][source]¶

Return the number of added and removed lines between two texts.

Parameters:

old_text – Original text.
new_text – Updated text.

Returns:

(additions, deletions).

klea_utils.mcp.tool_impls.file_ops.diff_payload(old_text: str, new_text: str, path: str = '') → tuple[str, int, int][source]¶

Return (diff, additions, deletions) for a change.

The diff is truncated to MAX_DIFF_CHARS.

Parameters:

old_text – Original text.
new_text – Updated text.
path – Label used for both sides of the diff.

Returns:

Unified diff text plus the added and removed line counts.

klea_utils.mcp.tool_impls.file_ops.is_binary(data: bytes) → bool[source]¶

Return whether data looks binary (contains a NUL byte).

Parameters:: data – Raw file bytes.
Returns:: True when a NUL byte is present.

klea_utils.mcp.tool_impls.file_ops.join_bom(text: str, bom: bool) → str[source]¶

Prefix text with a BOM when bom is set.

Parameters:

text – Text without a BOM.
bom – Whether to add the BOM.

Returns:

Text, optionally BOM-prefixed.

klea_utils.mcp.tool_impls.file_ops.normalize_newlines(text: str) → str[source]¶

Return text with CRLF endings converted to LF.

Parameters:: text – Text to normalise.
Returns:: Text with \r\n replaced by \n.

klea_utils.mcp.tool_impls.file_ops.read_whole_text(path: str | PathLike, *, max_bytes: int = 104857600) → TextFile[source]¶

Read a UTF-8 text file, preserving BOM, newline and mode.

Parameters:

path – File to read.
max_bytes – Maximum size accepted.

Raises:

FileEditError – when the path is missing, not a regular file, too large, binary, not valid UTF-8, or unreadable.

Returns:

A TextFile with normalised metadata.

klea_utils.mcp.tool_impls.file_ops.split_bom(text: str) → tuple[str, bool][source]¶

Split a leading BOM off text.

Parameters:: text – Decoded text that may start with a BOM.
Returns:: (text_without_bom, had_bom).

klea_utils.mcp.tool_impls.file_ops.split_lines(text: str) → list[str][source]¶

Split text into lines using the shared tool definition.

Lines end at \n after normalising CRLF and lone-CR endings, and a single trailing newline does not add an empty line (matching editors and wc -l). This deliberately avoids str.splitlines(), which also breaks on form feeds, vertical tabs, the \x1c-\x1e separators and Unicode line separators, so its line numbers disagree with the rest of the file toolchain.

Parameters:: text – File text (any newline convention).
Returns:: The lines, without line endings.

klea_utils.mcp.tool_impls.file_ops.unified_diff(old_text: str, new_text: str, path: str = '') → str[source]¶

Return a unified diff between two texts.

Parameters:

old_text – Original text.
new_text – Updated text.
path – Label used for both sides of the diff.

Returns:

Unified diff text (empty when the texts are identical).

klea_utils.mcp.tool_impls.file_ops.write_whole_text(path: str | PathLike, text: str, *, bom: bool = False, newline: str = '\n', mode: int | None = None, create_parents: bool = True) → None[source]¶

Atomically write text to path, preserving BOM/newline/mode.

The content is written to a temporary file in the target directory and os.replace``d into place, so a failure leaves any existing file untouched. An existing file's mode is preserved when *mode* is ``None; a new file gets DEFAULT_FILE_MODE.

Parameters:

path – Destination file.
text – Full file content (LF endings).
bom – Whether to prefix the UTF-8 BOM.
newline – Line ending to apply ("\n" or "\r\n").
mode – Octal mode to apply, or None to preserve/default.
create_parents – Create missing parent directories.

Raises:

FileEditError – when the destination is a directory or the write fails.

Whole-file write implementation for the Klea MCP tools.

See devdocs/system/edit-format-strategy.md; ADR-0039.

File: klea_utils/mcp/tool_impls/write_file.py

klea_utils.mcp.tool_impls.write_file.write_file(path: str, content: str, project_root: str | None = None) → dict[str, Any][source]¶

Create or overwrite a file with content.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool (see klea_utils.mcp.server.bundled_tools).

The write is atomic (a temporary file is replaced into place). An existing file keeps its mode; a UTF-8 BOM is preserved when either the old or the new content has one; the existing file’s line ending is kept and the new content normalised to it. Missing parent directories are created. The result carries a unified diff against the previous content.

Parameters:

path – File path to write.
content – Complete file content.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).

Returns:

dict with path, created, bytes_written, additions, deletions, diff, error.

Exact search/replace edit implementation for the Klea MCP tools (ADR-0039).

See devdocs/system/edit-format-strategy.md.

File: klea_utils/mcp/tool_impls/edit_file.py

klea_utils.mcp.tool_impls.edit_file.edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False, project_root: str | None = None) → dict[str, Any][source]¶

Replace an exact span of an existing file.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool (see klea_utils.mcp.server.bundled_tools).

Matching normalises line endings and, when an exact match fails, tries the bounded replacer chain in klea_utils.mcp.tool_impls.edit_replacers (line-trimmed, block-anchor, whitespace-normalised, indentation-flexible, context-aware). The file’s own line ending, BOM and mode are preserved on write, which is atomic. Unless replace_all is set, the matched span must be unique.

Parameters:

path – File path to edit.
old_string – Exact text to replace (must not be empty).
new_string – Replacement text (must differ from old_string).
replace_all – Replace every occurrence rather than requiring one.
project_root – Base directory used to locate nearby entries when the target is missing; the boundary itself is enforced by the client-side gate (ADR-0007 update 2026-10-08).

Returns:

dict with path, replacements, additions, deletions, matcher, diff, error, nearby, note. nearby/note are populated on a missing/not-a-file error so the caller can see what exists instead.

Replacer chain for the search/replace edit tool (ADR-0039).

A replacer takes the file content and the model’s old_string and yields candidate substrings of the actual content that are meant to match. The chain is tried in order, simplest first, so an exact match always wins and fuzzy matching is only reached when exact matching fails. Every candidate is subject to the uniqueness and proportional-match checks in apply_edit().

The matchers are adapted from opencode’s tool/edit.ts (which in turn credits Cline and the Gemini CLI).

File: klea_utils/mcp/tool_impls/edit_replacers.py

klea_utils.mcp.tool_impls.edit_replacers.MAX_MATCH_LINES = 20000¶: Above this many lines, only the exact matcher is tried. The fallback matchers are Python-level and cost O(lines x old_string lines); bounding them keeps an edit on a very large file from stalling the server. Exact matching is C-level so it stays available.

klea_utils.mcp.tool_impls.edit_replacers.MULTIPLE_CANDIDATES_SIMILARITY = 0.65¶: Similarity threshold for the best of several block-anchor candidates.

klea_utils.mcp.tool_impls.edit_replacers.REPLACERS: tuple[tuple[str, Callable[[str, str], Iterator[str]]], ...] = (('exact', <function simple_replacer>), ('line-trimmed', <function line_trimmed_replacer>), ('block-anchor', <function block_anchor_replacer>), ('whitespace-normalised', <function whitespace_normalized_replacer>), ('indentation-flexible', <function indentation_flexible_replacer>), ('context-aware', <function context_aware_replacer>))¶: The replacer chain, tried in order.

klea_utils.mcp.tool_impls.edit_replacers.Replacer¶

(content, old_string) -> candidate spans of content.

Type:: A replacer

alias of Callable[[str, str], Iterator[str]]

klea_utils.mcp.tool_impls.edit_replacers.SINGLE_CANDIDATE_SIMILARITY = 0.65¶: Similarity threshold for a single block-anchor candidate.

klea_utils.mcp.tool_impls.edit_replacers.apply_edit(content: str, old_string: str, new_string: str, replace_all: bool = False) → tuple[str, int, str, str][source]¶

Apply old_string -> new_string using the replacer chain.

Returns the updated content, the number of replacements, the name of the replacer that matched, and an error message (empty on success).

Parameters:

content – File content (LF-normalised).
old_string – Span to replace (LF-normalised).
new_string – Replacement span (LF-normalised).
replace_all – Replace every occurrence of the matched span.

Returns:

(updated, replacements, matcher, error).

klea_utils.mcp.tool_impls.edit_replacers.block_anchor_replacer(content: str, old_string: str) → Iterator[str][source]¶

Yield blocks anchored on first/last line with similar middle lines.

Recovers an edit when the model’s old_string is mostly right but one or more middle lines drifted slightly (a typo, a changed value), so exact and line-trimmed matching both fail.

How it works:

The block’s first and last trimmed lines are treated as distinctive, correct anchors; only blocks of at least three lines (anchors plus at least one middle) are considered.
Candidate blocks are found by pairing each content line equal to the first anchor with the next content line (at or after i + 2) equal to the last anchor. A pair is kept only when its length is within 25% of the search block’s length, so a nearby same-anchor block is not mistaken for the target.
Each candidate is scored by the average per-line similarity of its middle lines against the search block’s middle lines (difflib.SequenceMatcher). A lone candidate must clear SINGLE_CANDIDATE_SIMILARITY; with several, the best-scoring candidate wins if it clears MULTIPLE_CANDIDATES_SIMILARITY.

Only the candidate’s location comes from the anchors: the text yielded is the file’s real block, so the replacement preserves the file’s actual indentation and content.

Parameters:

content – File content (LF-normalised).
old_string – Span to find.

Yields:

The best matching block, if any.

klea_utils.mcp.tool_impls.edit_replacers.context_aware_replacer(content: str, old_string: str) → Iterator[str][source]¶

Yield blocks whose anchors match and whose middle lines mostly match.

A complementary fallback to block_anchor_replacer(), not simply a looser version of it. Both use the first and last trimmed lines as anchors, but they accept different kinds of drift:

block_anchor_replacer() lets the block length differ (by up to 25%) and scores every middle line by fuzzy similarity. It fits a block where each middle line is slightly wrong (for example a typo per line).
this replacer requires the exact same number of lines and at least half of the middle lines to match exactly. It fits a block where some middle lines are entirely different or replaced, but the surrounding context (and the anchors) is intact.

Neither subsumes the other: a block whose middle lines are all ~80% similar scores above the anchor threshold but has almost no exact matches, while a block that is half exact and half unrelated averages ~0.5 similarity (below the anchor threshold) yet passes the half-exact rule here.

How it works: old_string must have at least three lines. For each line matching the first anchor, the next last-anchor line at or after i + 2 is located; because the block length must equal the search length, that hit only counts when it is exactly at i + size - 1. The middle lines are compared for exact equality (after trimming) and the block is yielded when at least half of the comparable middle lines match (or when there are none to compare).

Unlike block_anchor_replacer(), which returns the single best candidate, this yields every qualifying block; the uniqueness check in apply_edit() then decides whether the edit is safe.

Parameters:

content – File content (LF-normalised).
old_string – Span to find.

Yields:

Matching blocks as they appear in content.

klea_utils.mcp.tool_impls.edit_replacers.indentation_flexible_replacer(content: str, old_string: str) → Iterator[str][source]¶

Yield blocks matching old_string once common indentation is removed.

Fixes a block that was shifted left or right as a whole.

Parameters:

content – File content (LF-normalised).
old_string – Span to find.

Yields:

Matching blocks as they appear in content.

klea_utils.mcp.tool_impls.edit_replacers.is_disproportionate_match(span: str, old_string: str) → bool[source]¶

Return whether span is far larger than old_string.

Guards against a loose replacer selecting a much larger region than the model asked to replace.

Parameters:

span – Candidate span found in the content.
old_string – The model’s requested span.

Returns:

True when the match should be refused.

klea_utils.mcp.tool_impls.edit_replacers.line_trimmed_replacer(content: str, old_string: str) → Iterator[str][source]¶

Yield blocks whose lines match old_string’s lines when trimmed.

Fixes blocks that were re-indented or had per-line whitespace changed.

Parameters:

content – File content (LF-normalised).
old_string – Span to find.

Yields:

Matching blocks as they appear in content.

klea_utils.mcp.tool_impls.edit_replacers.simple_replacer(content: str, old_string: str) → Iterator[str][source]¶

Yield old_string unchanged (exact match).

Parameters:

content – File content (LF-normalised).
old_string – Span to find.

Yields:

old_string.

klea_utils.mcp.tool_impls.edit_replacers.whitespace_normalized_replacer(content: str, old_string: str) → Iterator[str][source]¶

Yield spans matching old_string after whitespace collapsing.

Fixes tab/space and repeated-space differences on single or multiple lines.

Parameters:

content – File content (LF-normalised).
old_string – Span to find.

Yields:

Matching spans as they appear in content.

File download implementation for Klea MCP tools.

See devdocs/system/mcp-permissions.md.

File: klea_utils/mcp/tool_impls/download_file.py

async klea_utils.mcp.tool_impls.download_file.download_file(session: SessionLike | None, url: str, file_path: str | Path, params: dict[str, Any] | None = None, timeout: float | Timeout = 30.0, retries: int = 3, project_root: str | None = None, allow_internal_hosts: bool = False, max_download_bytes: int = 104857600) → Path | None[source]¶

Download a URL to file_path (overwriting) and return the path.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool that supplies session from their lifespan context (see klea_utils.mcp.lifespan). Note that since this overwrites, this should not be exposed directly as a tool; use a wrapper around this.

The request carries an honest User-Agent and is subject to the shared SSRF guard (refusing private/loopback hosts unless allow_internal_hosts is set). The raw response body is written as bytes, so binary files (PDFs, office documents) survive intact.

Transient failures (timeouts, connection errors, HTTP 5xx/429) are retried with exponential backoff. Returns None when the download fails (non-2xx response, no session available, an SSRF denial, or the target path is denied by the permission check).

Parameters:

session – HTTP session to use for the request. None when no session is available.
url – HTTP or HTTPS URL to download.
file_path – Destination file path (existing files are overwritten).
params – Optional query parameters for the request.
timeout – Request timeout in seconds.
retries – Number of attempts for transient failures.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).
allow_internal_hosts – Skip the SSRF guard (requests to loopback, private, link-local, or reserved addresses).
max_download_bytes – Maximum bytes to download; larger responses are aborted and the download is treated as failed to avoid OOM.

Returns:

The written Path, or None on failure.

async klea_utils.mcp.tool_impls.download_file.download_file_to_cache(session: SessionLike | None, url: str, cache_dir: str | Path, file_name: str, params: dict[str, Any] | None = None, timeout: float | Timeout = 30.0, retries: int = 3, allow_internal_hosts: bool = False, max_download_bytes: int = 104857600) → Path | None[source]¶

Download a URL into cache_dir as file_name and return the path.

Convenience wrapper around download_file() for callers that keep a per-app cache directory (see klea_utils.paths.get_cache_dir). The permission boundary is cache_dir itself: this helper may write inside it and nowhere else.

Parameters:

session – HTTP session to use for the request. None when no session is available.
url – HTTP or HTTPS URL to download.
cache_dir – Directory in which to store the downloaded file.
file_name – File name under cache_dir (existing files overwritten).
params – Optional query parameters for the request.
timeout – Request timeout in seconds.
retries – Number of attempts for transient failures.
allow_internal_hosts – Skip the SSRF guard (requests to loopback, private, link-local, or reserved addresses).
max_download_bytes – Maximum bytes to download; larger responses are aborted.

Returns:

The written Path, or None on failure.

async klea_utils.mcp.tool_impls.download_file.download_files(session: SessionLike | None, files: list[dict[str, Any]], target_dir: str | Path, max_concurrency: int = 3, timeout: float | Timeout = 30.0, retries: int = 3, max_download_bytes: int = 104857600) → dict[str, Any][source]¶

Download a list of files into target_dir, bounded in concurrency.

Framework-agnostic helper that drives download_file() for each entry in files, as returned by the repository source list functions (entries carry path and download_url). Relative path values are preserved under target_dir (parent directories are created as needed), and writes stay confined to target_dir.

target_dir is an explicit destination directory – it may be the current project, a working subfolder, or a cache directory – so the downloaded files are immediately usable where the caller asked for them.

Downloads run with bounded concurrency (an asyncio.Semaphore) so a large dataset does not hammer the source server. Individual failures are recorded per file and do not abort the rest of the batch, so a caller (e.g. an LLM) can retry the failed paths.

Parameters:

session – HTTP session to use for the requests. None when no session is available.
files – File entries to download; each needs path (relative target path) and download_url.
target_dir – Destination directory under which the files are written.
max_concurrency – Maximum number of downloads in flight.
timeout – Request timeout in seconds per download.
retries – Number of attempts for transient failures per download.
max_download_bytes – Maximum bytes per file; larger files are treated as failed to avoid OOM.

Returns:

dict with results (one entry per file: path plus saved_to on success or error on failure) and a top-level error (only set when the whole batch fails unexpectedly).

Shell command execution implementation for Klea MCP tools (ADR-0038).

Timeout and cancellation signal the command’s process group (SIGTERM, then SIGKILL). A descendant that detaches from the group (setsid/setpgid, as some build tools do) cannot be reached by a portable group signal and may survive.

See devdocs/system/mcp-permissions.md.

File: klea_utils/mcp/tool_impls/run_command.py

klea_utils.mcp.tool_impls.run_command.DEFAULT_MAX_OUTPUT_CHARS = 100000¶: Default cap on captured characters per stream (stdout, stderr).

klea_utils.mcp.tool_impls.run_command.DEFAULT_MAX_TIMEOUT_SECONDS = 600.0¶: Default ceiling for the per-command timeout, overridable via MAX_TIMEOUT_ENV_VAR (mirrors opencode’s 10-minute bash cap).

klea_utils.mcp.tool_impls.run_command.DEFAULT_TIMEOUT_SECONDS = 30.0¶: Default per-command timeout in seconds.

klea_utils.mcp.tool_impls.run_command.KILL_GRACE_SECONDS = 3.0¶: Grace period between SIGTERM and SIGKILL when killing a timed-out process group.

klea_utils.mcp.tool_impls.run_command.MAX_TIMEOUT_ENV_VAR = 'KLEA_RUN_COMMAND_MAX_TIMEOUT'¶: Process environment variable that raises/lowers the timeout ceiling (seconds). A process env var, not an env-file/config field: it is an operational bound (like KLEA_LOG_LEVEL), not app configuration.

klea_utils.mcp.tool_impls.run_command.max_timeout_seconds() → float[source]¶

Return the effective timeout ceiling in seconds.

Reads MAX_TIMEOUT_ENV_VAR from the process environment (always a string) and casts it to a float; an unset, non-numeric, non-finite, or non-positive value falls back to DEFAULT_MAX_TIMEOUT_SECONDS with a warning.

Returns:: The ceiling for timeout_seconds.

async klea_utils.mcp.tool_impls.run_command.run_command(command: str, working_directory: str | None = None, timeout_seconds: float = 30.0, max_output_chars: int = 100000, project_root: str | None = None) → dict[str, Any][source]¶

Run a shell command and return its exit status and captured output.

Framework-agnostic implementation shared across Klea MCP servers. Apps wrap this in an MCP tool (see klea_utils.mcp.server.bundled_tools).

This is not a sandbox (ADR-0038): the command can ignore working_directory (cd /, absolute paths) and has the host user’s filesystem, process, and network authority. The command inherits the server’s environment. Access control relies on the client-side path gate (working_directory and paths in the command, ADR-0007) plus the tool’s destructive / open_world annotations and the access level (ADR-0037).

The process is run with start_new_session so it can be killed as a group on timeout, with stdin closed so an interactive command cannot hang. Timeout and cancellation terminate the whole process group, but a command that detaches its children (setsid/daemonising) can leave processes running that the group signal cannot reach.

Parameters:

command – Shell command string to run.
working_directory – Directory to run in. Defaults to project_root (or the current working directory when project_root is None).
timeout_seconds – Seconds to wait before killing the process group. Must not exceed max_timeout_seconds().
max_output_chars – Maximum characters captured per stream.
project_root – Default base for working_directory when it is not given; not a permission check (the client-side gate enforces the boundary).

Returns:

dict with command, working_directory, returncode, stdout, stderr, truncated, error. A non-zero returncode is a normal command result (many tools signal conditions with it); error is set only when the command could not be run or was killed (timeout, denied working directory, spawn failure).

Read-only sqlite query implementation for Klea MCP tools.

File: klea_utils/mcp/tool_impls/sqlite_query.py

klea_utils.mcp.tool_impls.sqlite_query.DEFAULT_LIMIT = 50¶: Default maximum number of rows returned by sqlite_query().

klea_utils.mcp.tool_impls.sqlite_query.DEFAULT_MAX_OPS = 25000000¶: Default total sqlite VM-instruction budget for a single statement (the max_ops argument). Generous so normal queries are never affected; it only bounds runaway or pathological (LLM-generated) SELECTs.

klea_utils.mcp.tool_impls.sqlite_query.MAX_LIMIT = 200¶: Hard cap on the number of rows returned, whatever the caller asks for.

klea_utils.mcp.tool_impls.sqlite_query.PROGRESS_TICK = 100000¶

the handler fires roughly every PROGRESS_TICK virtual-machine instructions.

Type:: Granularity of the progress-handler ticks

klea_utils.mcp.tool_impls.sqlite_query.SQLITE_SUFFIXES = frozenset({'.db', '.sqlite', '.sqlite3'})¶: File extensions accepted as sqlite databases.

klea_utils.mcp.tool_impls.sqlite_query.sqlite_query(db_path: str, sql: str, limit: int = 50, project_root: str | None = None, max_ops: int | None = 25000000) → dict[str, Any][source]¶

Run a read-only SELECT query against a sqlite database.

Framework-agnostic implementation shared across Klea MCP servers (and directly usable by agents). The database is opened read-only (mode=ro plus PRAGMA query_only), so nothing can modify it even if validation is bypassed. Only a single SELECT statement is allowed; other statements, statement separators, and unknown tables/columns return a clear error rather than raising.

Use when:

Answering a question from structured, tabular data stored in a sqlite database (exact values, LIKE patterns, ranges, aggregations).
The data is not searchable as free text (that is what the vector/BM25 stores are for).

Do not use for:

Free-text / semantic search over document collections.
Anything that writes; this tool is read-only by construction.

Example: sqlite_query(db_path=”data.db”, sql=”SELECT name, year FROM papers WHERE journal = ‘Nature’ ORDER BY year DESC”)

Parameters:

db_path – Path to the sqlite database file (.sqlite/.sqlite3/.db), relative to the project root.
sql – A single SELECT statement (a trailing semicolon is allowed).
limit – Maximum number of rows to return (clamped to MAX_LIMIT).
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).
max_ops – Total sqlite virtual-machine instruction budget for the statement, bounding how much CPU a query may consume. None disables the budget. Non-integer or non-positive values fall back to the default.

Returns:

Dictionary with db_path, columns, rows, row_count, truncated, error.

klea_utils.mcp.tool_impls.sqlite_query.sqlite_schema(db_path: str, project_root: str | None = None, max_ops: int | None = 25000000) → dict[str, Any][source]¶

Inspect the tables and columns of a sqlite database.

Framework-agnostic implementation shared across Klea MCP servers (and directly usable by agents). Returns the list of tables/views and, for each, its columns with their declared types, so a caller can write a valid query. Intended for agent-style callers that can chain an inspection step before querying; a RAG wrapper of sqlite_query() usually embeds the schema in the tool description instead, since the RAG runs tools in a single pass and cannot take a separate look-up step.

Use when:

You need to know what tables and columns a database has before writing a query.

Do not use for:

Querying data (use sqlite_query instead).

Example: sqlite_schema(db_path=”data.db”)

Parameters:

db_path – Path to the sqlite database file (.sqlite/.sqlite3/.db), relative to the project root.
project_root – Retained for callers; the boundary is enforced by the client-side gate, not here (ADR-0007 update 2026-10-08).
max_ops – Total sqlite virtual-machine instruction budget for the introspection statements. None disables the budget.

Returns:

Dictionary with db_path, tables, error. tables maps each table/view name to {"columns": [{"name": ..., "type": ...}, ...]}.

Paper search sources¶

Framework-agnostic sources that search scholarly APIs and return normalised PaperRecord objects. klea_utils.mcp.tool_impls.papers.search.PaperSearchResolver rotates and falls back across them, and the search_papers bundled tool wraps it.

Academic paper search across several scholarly APIs.

Peer reviewed sources are preferred. The general sources (OpenAlex, Crossref, and Semantic Scholar when SEMANTIC_SCHOLAR_API_KEY is set) are queried in round-robin order across calls, like klea_utils.biblio.doi.DoiResolver, so no single API takes every request; when the primary is rate limited, fails or finds nothing, the next one is tried. Life sciences searches try Europe PMC, then PubMed, before the general sources. Preprints (arXiv, Europe PMC preprints) are only searched when asked for, and are listed after the peer reviewed results.

File: klea_utils/mcp/tool_impls/papers/search.py

class klea_utils.mcp.tool_impls.papers.search.PaperSearchResolver(rotation: Iterator[int] | None = None)[source]¶

Bases: object

Search academic papers across a pool of scholarly APIs.

async search(session: SessionLike | None, query: str, domain: str = 'general', preprints: bool = False, max_results: int = 10) → dict[str, Any][source]¶

Search academic papers by keywords.

Parameters:

session – HTTP session to use for the requests.
query – Keywords to search for.
domain – "general" or "life-sciences".
preprints – Also search preprints, listed after peer reviewed results.
max_results – Maximum results from the peer reviewed sources, and separately from the preprint sources.

Returns:

Dictionary with query, domain, preprints, results, error and note. note lists sources that failed, if any.

async klea_utils.mcp.tool_impls.papers.search.search_papers(session: SessionLike | None, query: str, domain: str = 'general', preprints: bool = False, max_results: int = 10) → dict[str, Any][source]¶

Search academic papers by keywords.

Thin wrapper over the shared PaperSearchResolver; this is the entry point the bundled tool calls.

Parameters:

session – HTTP session to use for the requests.
query – Keywords to search for.
domain – "general" or "life-sciences".
preprints – Also search preprints, listed after peer reviewed results.
max_results – Maximum results from the peer reviewed sources, and separately from the preprint sources.

Returns:

Dictionary with query, domain, preprints, results, error and note. note lists sources that failed, if any.

Normalised paper record returned by every paper search source.

File: klea_utils/mcp/tool_impls/papers/record.py

klea_utils.mcp.tool_impls.papers.record.MAX_ABSTRACT_CHARS = 2000¶: Abstracts longer than this are cut, so a page of results stays small enough for the model’s context.

class klea_utils.mcp.tool_impls.papers.record.PaperRecord(*, title: str | None = None, authors: list[str] = [], year: int | None = None, journal: str | None = None, abstract: str | None = None, doi: str | None = None, url: str | None = None, peer_reviewed: bool = True, source: str = '')[source]¶

Bases: BiblioRecord

A search hit: the DOI resolver’s record plus search-specific fields.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

peer_reviewed: bool¶: False for preprints, so the model can tell the user the paper has not been peer reviewed.

source: str¶: Name of the service that returned the record.

klea_utils.mcp.tool_impls.papers.record.clean_doi(raw: str | None) → str | None[source]¶

Strip a https://doi.org/ or doi: prefix from an API-provided DOI.

Unlike klea_utils.biblio.normalize.normalize_doi() this keeps everything after the prefix: DOI suffixes may contain / (for example 10.1088/0957-4484/24/38/383001).

klea_utils.mcp.tool_impls.papers.record.from_biblio(record: BiblioRecord, *, url: str | None, peer_reviewed: bool, source: str, doi: str | None = None) → PaperRecord[source]¶

Build a PaperRecord from a BiblioRecord.

Parameters:

record – Record produced by one of the DOI resolver normalisers.
url – Link to the paper’s landing page.
peer_reviewed – Whether the paper is peer reviewed.
source – Name of the service that returned the record.
doi – DOI to use instead of record.doi.

klea_utils.mcp.tool_impls.papers.record.is_preprint(record: BiblioRecord) → bool[source]¶

Whether record comes from a preprint server, by its venue.

Some sources tag preprints as journal articles (Semantic Scholar, and SSRN in Crossref), so their peer reviewed filters alone are not enough. The DOI is not checked: Semantic Scholar merges a preprint with its published version and can keep the preprint DOI, while the venue is then the journal.

klea_utils.mcp.tool_impls.papers.record.truncate_abstract(abstract: str | None) → str | None[source]¶: Cut abstract to MAX_ABSTRACT_CHARS, marking the cut.

Errors for the academic paper search layer.

File: klea_utils/mcp/tool_impls/papers/errors.py

exception klea_utils.mcp.tool_impls.papers.errors.PaperSourceError[source]¶

Bases: Exception

Raised when a paper source cannot be queried (HTTP error, 429, bad body).

The search falls back to the next source when this is raised.

Shared HTTP helpers for the paper search sources.

File: klea_utils/mcp/tool_impls/papers/http.py

class klea_utils.mcp.tool_impls.papers.http.Throttle(min_interval: float)[source]¶

Bases: object

Keep at least min_interval seconds between requests to one service.

Used for services that ask for a gap between requests (arXiv, Semantic Scholar and PubMed); each module explains its own gap.

async klea_utils.mcp.tool_impls.papers.http.get_json(session: SessionLike | None, url: str, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None) → Any[source]¶

GET url and return the decoded JSON body.

Raises:: PaperSourceError – on any failure or an unreadable body.

async klea_utils.mcp.tool_impls.papers.http.get_text(session: SessionLike | None, url: str, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None) → str[source]¶

GET url and return the body as text (for the XML APIs).

Raises:: PaperSourceError – on any failure.

OpenAlex paper search source.

File: klea_utils/mcp/tool_impls/papers/openalex.py

klea_utils.mcp.tool_impls.papers.openalex.PEER_REVIEWED_FILTER = 'type:article,primary_location.source.type:journal'¶: OpenAlex search also returns preprints, datasets etc.; this keeps only journal articles.

async klea_utils.mcp.tool_impls.papers.openalex.search(session: SessionLike | None, query: str, limit: int) → list[PaperRecord][source]¶

Search OpenAlex for peer reviewed journal articles.

Raises:: PaperSourceError – when the request fails or is rate limited.

Crossref paper search source.

File: klea_utils/mcp/tool_impls/papers/crossref.py

klea_utils.mcp.tool_impls.papers.crossref.PEER_REVIEWED_FILTER = 'type:journal-article'¶: Crossref search also returns preprints (“posted-content”), datasets etc.; this keeps journal articles.

async klea_utils.mcp.tool_impls.papers.crossref.search(session: SessionLike | None, query: str, limit: int) → list[PaperRecord][source]¶

Search Crossref for peer reviewed journal articles.

Raises:: PaperSourceError – when the request fails or is rate limited.

Semantic Scholar paper search source.

The search endpoint answers HTTP 429 to nearly every request without an API key, so this source is only used when SEMANTIC_SCHOLAR_API_KEY is set.

File: klea_utils/mcp/tool_impls/papers/semantic_scholar.py

klea_utils.mcp.tool_impls.papers.semantic_scholar.api_key() → str | None[source]¶: Return the Semantic Scholar API key, or None when unset.

async klea_utils.mcp.tool_impls.papers.semantic_scholar.search(session: SessionLike | None, query: str, limit: int) → list[PaperRecord][source]¶

Search Semantic Scholar for peer reviewed journal articles.

Raises:: PaperSourceError – when no key is set, or the request fails or is rate limited.

Europe PMC paper search source (life sciences).

Europe PMC indexes PubMed (SRC:MED) and preprint servers such as bioRxiv, medRxiv and Research Square (SRC:PPR), so one free, keyless API covers both peer reviewed papers and preprints.

File: klea_utils/mcp/tool_impls/papers/europepmc.py

async klea_utils.mcp.tool_impls.papers.europepmc.search(session: SessionLike | None, query: str, limit: int, preprints: bool = False) → list[PaperRecord][source]¶

Search Europe PMC for PubMed papers, or preprints when preprints.

Raises:: PaperSourceError – when the request fails or is rate limited.

PubMed paper search source (life sciences), via NCBI E-utilities.

Used as the fallback for Europe PMC: esearch finds the PubMed IDs, then efetch returns the records (with abstracts) as XML.

File: klea_utils/mcp/tool_impls/papers/pubmed.py

klea_utils.mcp.tool_impls.papers.pubmed.PEER_REVIEWED_FILTER = 'NOT preprint[pt]'¶: PubMed also indexes some preprints; this keeps only peer reviewed records.

async klea_utils.mcp.tool_impls.papers.pubmed.search(session: SessionLike | None, query: str, limit: int) → list[PaperRecord][source]¶

Search PubMed for peer reviewed papers.

Raises:: PaperSourceError – when a request fails, is rate limited, or returns an unreadable body.

arXiv paper search source (general preprints).

File: klea_utils/mcp/tool_impls/papers/arxiv.py

async klea_utils.mcp.tool_impls.papers.arxiv.search(session: SessionLike | None, query: str, limit: int) → list[PaperRecord][source]¶

Search arXiv preprints.

Raises:: PaperSourceError – when the request fails, is rate limited, or the body is not valid Atom XML.

Repository sources¶

Framework-agnostic functions that list the versions and files of archival repositories (GitHub, FigShare, DANDI Archive, BioModels). The returned file lists carry direct download_url values that can be fed to klea_utils.mcp.tool_impls.download_file.download_files() (or the single-file download_file implementation), e.g. by wrapping them into MCP tools.

Shared helpers for the repository source implementations.

File: klea_utils/mcp/tool_impls/repositories/sources.py

klea_utils.mcp.tool_impls.repositories.sources.REQUEST_TIMEOUT = Timeout(timeout=30.0)¶: Request timeout for the JSON API calls, in seconds.

Error classes for the repository source implementations.

File: klea_utils/mcp/tool_impls/repositories/errors.py

exception klea_utils.mcp.tool_impls.repositories.errors.RepositorySourceError[source]¶

Bases: Exception

Raised when a repository source cannot be queried.

Carries a user-facing message that the public functions report through their error result field.

GitHub repository source implementation.

File: klea_utils/mcp/tool_impls/repositories/github.py

klea_utils.mcp.tool_impls.repositories.github.GITHUB_API_BASE = 'https://api.github.com/repos'¶: GitHub REST API base for the /repos collection.

klea_utils.mcp.tool_impls.repositories.github.PAGE_SIZE = 100¶: Page size for the branches/tags listing endpoints.

klea_utils.mcp.tool_impls.repositories.github.RAW_BASE = 'https://raw.githubusercontent.com'¶: Base URL for direct raw file downloads.

async klea_utils.mcp.tool_impls.repositories.github.github_list_files(session: SessionLike | None, url: str, version: str | None = None) → dict[str, Any][source]¶

List the files in a GitHub repository at a given version.

The version is a git branch or a tag. When a name exists as both a branch and a tag, a branch is assumed (git ref resolution precedence). If version is omitted, the repository’s default branch is used.

Use when:

Getting the file list of a GitHub repository so files can be downloaded.

Parameters:

url – GitHub repository URL (https://github.com/<owner>/<repo>).
version – Branch or tag to list. Defaults to the default branch.

Returns:

Dictionary with source, url, version, files (path, name, download_url, size), and error.

async klea_utils.mcp.tool_impls.repositories.github.github_list_versions(session: SessionLike | None, url: str) → dict[str, Any][source]¶

List the available versions (branches and tags) of a GitHub repository.

A GitHub version is a git branch or a tag; both are merged into a single list. When a name exists as both a branch and a tag, it is listed once.

Use when:

Discovering which branches/tags a GitHub repository offers before listing its files.

Parameters:: url – GitHub repository URL (https://github.com/<owner>/<repo>).
Returns:: Dictionary with source, url, versions, and an empty files list.

FigShare repository source implementation.

File: klea_utils/mcp/tool_impls/repositories/figshare.py

klea_utils.mcp.tool_impls.repositories.figshare.FIGSHARE_API_BASE = 'https://api.figshare.com/v2'¶: FigShare API v2 base URL. Even for institutional FigShare instances (e.g. rdr.ucl.ac.uk) the article IDs and the API endpoint are shared.

klea_utils.mcp.tool_impls.repositories.figshare.MAX_PAGES = 100¶: Cap on the number of pages fetched, to bound runaway loops.

klea_utils.mcp.tool_impls.repositories.figshare.PAGE_SIZE = 1000¶: Page size for the article files endpoint (the API maximum is 1000).

async klea_utils.mcp.tool_impls.repositories.figshare.figshare_list_files(session: SessionLike | None, url: str, version: str | None = None) → dict[str, Any][source]¶

List the files of a FigShare article.

FigShare serves the same file list for every version of an article (the files endpoint is not versioned); the version argument is accepted for a uniform API but only labels the result. When version is omitted, the article’s current version is reported.

Use when:

Getting the file list of a FigShare article so files can be downloaded.

Parameters:

url – FigShare article URL (e.g. https://figshare.com/articles/dataset/<title>/<article_id>).
version – Version label for the result. Defaults to the article’s current version.

Returns:

Dictionary with source, url, version, files (path, name, download_url, size), and error.

async klea_utils.mcp.tool_impls.repositories.figshare.figshare_list_versions(session: SessionLike | None, url: str) → dict[str, Any][source]¶

List the available versions of a FigShare article.

Use when:

Discovering which versions a FigShare article offers before listing its files.

Parameters:: url – FigShare article URL (e.g. https://figshare.com/articles/dataset/<title>/<article_id>).
Returns:: Dictionary with source, url, versions, and an empty files list.

DANDI Archive repository source implementation.

File: klea_utils/mcp/tool_impls/repositories/dandi.py

klea_utils.mcp.tool_impls.repositories.dandi.DANDI_API_BASE = 'https://api.dandiarchive.org/api'¶: DANDI Archive REST API base URL.

klea_utils.mcp.tool_impls.repositories.dandi.DANDI_HOSTS = ('dandiarchive.org', 'www.dandiarchive.org')¶: Hosts that serve DANDI Archive.

klea_utils.mcp.tool_impls.repositories.dandi.DRAFT_VERSION = 'draft'¶: The default version label, used when none is specified.

klea_utils.mcp.tool_impls.repositories.dandi.MAX_FILES = 10000¶: Cap on the total number of files collected, to bound runaway recursion.

klea_utils.mcp.tool_impls.repositories.dandi.MAX_PAGES = 100¶: Cap on the number of pages fetched per folder, to bound runaway loops.

klea_utils.mcp.tool_impls.repositories.dandi.PAGE_SIZE = 100¶: Page size for the assets/paths endpoint.

async klea_utils.mcp.tool_impls.repositories.dandi.dandi_list_files(session: SessionLike | None, url: str, version: str | None = None) → dict[str, Any][source]¶

List the files of a DANDI dandiset at a given version.

The file tree is walked recursively via the assets/paths endpoint: entries with an asset are files, entries without one are folders that are descended into. When version is omitted, the draft version is used.

Use when: - Getting the file list of a DANDI dandiset so files can be downloaded.

Parameters:

url – DANDI dandiset URL (https://dandiarchive.org/dandiset/<id>).
version – Version to list (e.g. draft or a published version). Defaults to draft.

Returns:

Dictionary with source, url, version, files (path, name, download_url, size), and error.

async klea_utils.mcp.tool_impls.repositories.dandi.dandi_list_versions(session: SessionLike | None, url: str) → dict[str, Any][source]¶

List the available versions of a DANDI dandiset.

The list includes the working draft version as well as published versions.

Use when:

Discovering which versions a DANDI dandiset offers before listing its files.

Parameters:: url – DANDI dandiset URL (https://dandiarchive.org/dandiset/<id>).
Returns:: Dictionary with source, url, versions, and an empty files list.

BioModels repository source implementation.

File: klea_utils/mcp/tool_impls/repositories/biomodels.py

klea_utils.mcp.tool_impls.repositories.biomodels.BIOMODELS_API_BASE = 'https://www.biomodels.org'¶: Canonical BioModels API base. The legacy www.ebi.ac.uk/biomodels host redirects here, so all API calls go to this base regardless of the host used in the model URL.

klea_utils.mcp.tool_impls.repositories.biomodels.BIOMODELS_HOSTS = ('biomodels.org', 'www.biomodels.org')¶: Hosts that serve BioModels model pages.

async klea_utils.mcp.tool_impls.repositories.biomodels.biomodels_list_files(session: SessionLike | None, url: str, version: str | None = None) → dict[str, Any][source]¶

List the files of a BioModels model at a given revision.

The file list combines the main and additional file groups of the model record. When version is omitted, the latest revision is used.

Use when: - Getting the file list of a BioModels model so files can be downloaded.

Parameters:

url – BioModels model URL (e.g. https://www.biomodels.org/MODEL0912160000).
version – Revision number (as a string) to list. Defaults to the latest revision.

Returns:

Dictionary with source, url, version, files (path, name, download_url, size), and error.

async klea_utils.mcp.tool_impls.repositories.biomodels.biomodels_list_versions(session: SessionLike | None, url: str) → dict[str, Any][source]¶

List the available revisions (versions) of a BioModels model.

Use when:

Discovering which revisions a BioModels model offers before listing its files.

Parameters:: url – BioModels model URL (e.g. https://www.biomodels.org/MODEL0912160000).
Returns:: Dictionary with source, url, versions, and an empty files list.

Bundled tools server¶

The bundled tools server (auto-launched by apps as a stdio subprocess and exposed standalone via the klea-mcp CLI) and its configuration.

class klea_utils.mcp.server.config.BundledToolsConfig(*, enabled: bool = True, include_tags: set[str] = <factory>, exclude_tags: set[str] = <factory>)[source]¶

Bases: BaseModel

Configuration for the shared bundled tools server.

Read by the app orchestrators (BaseLangGraph) to decide whether the bundled tools server is wired into the graph and, when it is, which of its tools are exposed. Tools are filtered by tag (see klea_utils.mcp.server.bundled_tools for the bundled tag vocabulary); the tags are applied to the stdio config entry so fastmcp’s TransformingStdioMCPServer enforces them.

Pydantic-only on purpose: config loads must not pull in fastmcp.

exclude_tags: set[str]¶: Tools carrying any of these tags are hidden.

include_tags: set[str]¶: Only tools carrying at least one of these tags are exposed.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Bundled tools server for Klea.

Provides the common Klea tools (web fetch, file list/read, download) as an MCP server. Apps auto-launch this module over stdio (see BaseLangGraph._bundled_server_config) so users get the common tools with no extra setup; the same server can also be run standalone over HTTP via the klea-mcp CLI for remote deployments.

Tool implementations live in klea_utils.mcp.tool_impls and the FastMCP wrappers in klea_utils.mcp.server.bundled_tools; this module wires them onto a FastMCP server.

File: klea_utils/mcp/server/bundled.py

klea_utils.mcp.server.bundled.bundle_server = 'KleaBundled'¶: The bundled FastMCP server instance. Apps embed this module as a stdio subprocess (python -m klea_utils.mcp.server.bundled); tests and the klea-mcp CLI use it directly.

klea_utils.mcp.server.bundled.main(transport: str = <typer.models.OptionInfo object>, port: int = <typer.models.OptionInfo object>) → None[source]¶

Run the bundled tools server.

Accessed via the klea-mcp entry point. --transport http serves the same pre-registered tools over HTTP so a remote client (e.g. a RAG deployment that runs Klea and the bundled server on different hosts) can point its mcp_servers config at this server’s URL.

FastMCP tool wrappers for the shared Klea bundled tools.

File: klea_utils/mcp/server/bundled_tools.py

klea_utils.mcp.server.bundled_tools.BUNDLED_TAG = 'bundled'¶: Common tags carried by every bundled tool, so “enable the common set” is a single include_tags: [“bundled”] in the app config.

klea_utils.mcp.server.bundled_tools.STORES_TAG = 'stores'¶: Tag marking the tool as searching the deployment’s document stores.

async klea_utils.mcp.server.bundled_tools.download_file(ctx: fastmcp.Context, url: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], file_path: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])]) → fastmcp.tools.ToolResult[source]¶

Download a URL to a local file.

Use this tool to fetch binary or text resources from the web and save them to disk for later reading or processing.

Use when: - Downloading a file such as a PDF, dataset, or archive. - Saving remote content locally before inspecting it.

Do not use for: - Reading a web page as text (use the web fetch tool instead).

Example: download_file(url=”https://example.com/paper.pdf”, file_path=”paper.pdf”)

Parameters:

url – HTTP or HTTPS URL to download.
file_path – Destination path, relative to the working directory. Existing files are overwritten.

Returns:

Dictionary with the saved path, or an error on failure.

async klea_utils.mcp.server.bundled_tools.edit_file(path: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], old_string: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Exact text to replace', metadata=[MinLen(min_length=1)])], new_string: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Replacement text')], replace_all: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Replace every occurrence instead of requiring one')] = False) → fastmcp.tools.ToolResult[source]¶

Replace an exact span of text in an existing file.

Use this tool for targeted changes. To supply old_string, first read the file with read_file(line_numbers=false) so the text matches exactly, including whitespace and indentation.

Use when:

Changing a small part of an existing file.
Updating one occurrence, or every occurrence with replace_all.

Do not use for:

Creating a file or replacing its entire content (use the write file tool instead).
Reading a file (use the read file tool instead).

Example: edit_file(path=”a.py”, old_string=”x = 1”, new_string=”x = 2”)

Parameters:

path – File path to edit, relative to the project directory.
old_string – Exact text to replace; must occur once unless replace_all is set.
new_string – Replacement text.
replace_all – Replace every occurrence (default: require a unique match).

Returns:

Dictionary with path, replacements, additions, deletions, matcher, diff, error. On a missing/not-a-file error, nearby (entries in the nearest existing directory) and note are populated so the caller can see what exists instead.

async klea_utils.mcp.server.bundled_tools.find_files(pattern: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Glob pattern matched against file paths, e.g. '*.py' or '**/*.py'. Use '*' to list every file", metadata=[MinLen(min_length=1)])] = '*', path: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Directory to search. Must be relative to the current working directory and cannot contain '..' for security", metadata=[MinLen(min_length=1)])] = '.', include_ignored: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Also list files ignored by .gitignore. Default false, which skips ignored files (and skips .git/cache directories always)')] = False, max_results: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum number of file paths to return', metadata=[Ge(ge=1), Le(le=10000)])] = 100) → fastmcp.tools.ToolResult[source]¶

Find files by path pattern anywhere under a directory.

Use this tool to locate files by name or extension when you do not know where they are in the tree. It returns file paths only, without contents or directory entries.

Use when:

Finding files by name or extension across a project tree.
Discovering where a file lives before reading it.

Do not use for:

Listing the contents of a specific directory, including its subdirectories (use the list files tool instead).
Searching inside file contents (use the grep tool instead).

Example: find_files(pattern="**/*.py")

Parameters:

pattern – Glob pattern matched against file paths; "*" lists all.
path – Directory to search, relative to the project directory.
include_ignored – Also list .gitignore-ignored files.
max_results – Maximum number of file paths to return.

Returns:

Dictionary with files (paths relative to the search directory), truncated, error.

async klea_utils.mcp.server.bundled_tools.grep(pattern: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], path: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Directory to search. Must be relative to the current working directory and cannot contain '..' for security", metadata=[MinLen(min_length=1)])] = '.', include: Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description="Space separated glob patterns to restrict which files are searched, e.g. '*.py' or '*.py *.md'. Omit to search all files")] = None, case_sensitive: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether the search is case sensitive')] = True, include_ignored: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Also search files ignored by .gitignore. Default false, which skips ignored files (and skips .git/cache directories always)')] = False, context_lines: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Lines of context before and after each match (0-7). The default of 2 usually shows enough surrounding code to avoid a follow-up read_file; raise it for denser blocks', metadata=[Ge(ge=0), Le(le=7)])] = 2, max_results: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum number of matching lines to return', metadata=[Ge(ge=1), Le(le=1000)])] = 100) → fastmcp.tools.ToolResult[source]¶

Search file contents for lines matching a regular expression.

Use this tool to find where a symbol, string, or pattern occurs in the files under a directory.

Use when:

Locating a definition, call site, or error message in the codebase.
Searching the contents of many files without reading each one.

Do not use for:

Listing files by name without searching their contents (use the list files tool instead).
Reading a whole file (use the read file tool instead).

Example: grep(pattern="def main", include="*.py")

Parameters:

pattern – Regular expression to search for.
path – Directory to search, relative to the project directory.
include – Space separated glob patterns restricting which files are searched.
case_sensitive – Whether the search is case sensitive.
include_ignored – Also search .gitignore-ignored files.
context_lines – Lines of context before and after each match (0-7).
max_results – Maximum number of matching lines to return.

Returns:

Dictionary with pattern, path, match hunks, truncated, files_scanned, error. Each hunk is {path, start_line, end_line, lines} with one {line_number, line, match} entry per line, so callers get the surrounding code without a separate read.

async klea_utils.mcp.server.bundled_tools.list_files(path: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Directory path to list. Must be relative to current working directory and cannot contain '..' for security. Defaults to '.'", metadata=[MinLen(min_length=1)])] = '.', max_depth: Annotated[int | None, FieldInfo(annotation=NoneType, required=True, description="Maximum directory depth to traverse. 'None' for unlimited")] = None, pattern: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="\n Space separated file patterns to filter based on files type.\n Correct: '*.py'\n Correct: '*.md'\n Correct: '*.py *.md'\n ")] = '*', include_files: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether to include files in results')] = True, include_directories: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether to include directories in results')] = True, recursive: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='If True, traverse subdirectories recursively')] = False, max_results: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum number of entries to return', metadata=[Ge(ge=1), Le(le=10000)])] = 100) → fastmcp.tools.ToolResult[source]¶

List files and directories with filtering and metadata.

Use this tool to explore the local file system structure and find specific files.

Use when:

Discovering what files exist in the working directory.
Finding files by name, type, or location.

Do not use for:

Reading a file’s contents (use the read file tool instead).
Finding a file anywhere under a directory tree by path pattern (use the find files tool instead).

If the pattern or the include_* filters match nothing but the directory is not empty, the unfiltered listing is returned instead and note explains why, so the response still shows what is there.

Example: list_files(path=".", pattern="*.py", recursive=True)

Parameters:

path – Directory path to list. Must be relative to the current working directory and cannot contain ‘..’ for security. Defaults to the current directory (‘.’).
max_depth – Maximum directory depth to traverse. ‘None’ for unlimited.
pattern – Space separated file patterns to filter files by type.
include_files – Whether to include files in results.
include_directories – Whether to include directories in results.
recursive – If True, traverse subdirectories recursively.
max_results – Maximum number of entries to return.

Returns:

Dictionary with list of files, truncated flag, error, and note (empty unless a filter matched nothing and the unfiltered listing was substituted).

async klea_utils.mcp.server.bundled_tools.read_file(path: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="File path to read. Must be relative to current working directory and cannot contain '..' for security. Defaults to '.'", metadata=[MinLen(min_length=1)])] = '.', offset: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='1-indexed line to start reading from. Negative values count from the end: -1 is the last line, -N is the Nth line from the end')] = 1, char_offset: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='0-indexed character position within the start line; only needed to continue inside a line longer than the character cap', metadata=[Ge(ge=0)])] = 0, limit: Annotated[int | None, FieldInfo(annotation=NoneType, required=True, description='Maximum number of lines to return (>=1). None removes the line limit, but the server character budget still caps the response', metadata=[Ge(ge=1)])] = 500, line_numbers: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Prefix each returned line with its line number (default true). Set false to get raw line text, e.g. to copy into an edit')] = True) → fastmcp.tools.ToolResult[source]¶

Read a file and return a slice of its text content.

Use this tool to inspect source files, logs, or documents as plain text. Document formats (PDF, office files) are converted to Markdown first.

Use when:

You need to see the contents of a file in the project.
You want to page through a large file by line numbers.
You need the end of a large file, such as the tail of a build log: use a negative offset (offset=-100) instead of paging from the start.

Read file chunks instead of complete files. Use the grep tool to locate the region you need first, then read around it with offset/limit.

Do not use for:

Listing a directory (use the list files tool instead).
Fetching remote content (use the web fetch tool instead).

Examples:

read_file(path="README.md", offset=1, limit=100)
read_file(path="build.log", offset=-100)

Parameters:

path – File path to read. Must be relative to the current working directory and cannot contain ‘..’ for security. Defaults to the current directory (‘.’), which is not a file, so an omitted path yields the nearby-entries fallback instead of a validation error.
offset – 1-indexed line to start reading from. A negative value counts from the end: -1 is the last line, -N is the Nth line from the end.
char_offset – 0-indexed character position within the start line; used only to continue inside a single line longer than the character cap (see next_char_offset in the result).
limit – Maximum number of lines to return. None removes only the line limit; the server character budget still caps the response, so a None read can still be truncated with next_offset. For a negative offset, this counts forward from the tail start.
line_numbers – Prefix each line with its line number; set false for raw text to copy into an edit.

The response size is always capped by a server-side character budget, reported as char_budget; a large file is read by paging (offset/limit) or from the end rather than returned in one call.

Returns:: Dictionary with content, line range, total_lines, truncated, truncation_reason, char_budget, next_offset, next_char_offset, error and note. line_start/line_end and next_offset are negative for a tail read, with -1 as the last line. total_lines is null when a streamed read stops before the relevant end (EOF for a forward read, the file start for a tail read); a tail that reaches past the start clamps to the first line and reports the exact total. A partial tail that reaches EOF has truncated: false, next_offset: null and line_end: -1. char_budget is the effective server character cap, and truncation_reason is char_budget, line_limit or scan_cap when truncated (empty otherwise), so a limit=None read shows why it was cut. A truncated read is a success (empty error) that normally ends on a line boundary and carries next_offset (and an explanatory note) so the caller can continue. A single line longer than the character cap is paged by characters: continue with the same offset and the returned next_char_offset. On a missing/not-a-file error, nearby (entries in the nearest existing directory) and note are populated so the caller can see what exists instead.

async klea_utils.mcp.server.bundled_tools.run_command(command: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Shell command string to run. Pipes, '&&' and redirection are supported. Scope the command to the requested target; do not run a broader command or suite than asked (for example run the one named test file, not the whole suite).", metadata=[MinLen(min_length=1)])], working_directory: Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Directory to run the command in; must be inside the project directory. Defaults to the project directory.')] = None, timeout_seconds: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Seconds before the command is killed. Defaults to 30; the maximum is set by the KLEA_RUN_COMMAND_MAX_TIMEOUT environment variable (default 600).', metadata=[Ge(ge=1)])] = 30.0, max_output_chars: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum characters captured per output stream', metadata=[Ge(ge=1)])] = 100000) → fastmcp.tools.ToolResult[source]¶

Run a shell command and return its exit status and output.

Use this tool to inspect or act on the workspace: print the working directory, list processes, run a build or test, or invoke a script.

Use when: - You need a fact about the environment (working directory, host, date). - You need to run a build, test, or script in the project.

Do not use for: - Reading a file (use the read file tool instead). - Listing a directory (use the list files tool instead).

A non-zero exit code is a normal result, not an error: many tools signal conditions with it (diff/grep return 1 for differences/no match, a failing test returns non-zero). Judge the outcome from returncode and stdout; error is set only when the command could not be run or was killed.

A timed-out or cancelled command has its process group killed, but a command that detaches its children (for example daemonises or calls setsid) may leave processes running; check with ps/pgrep and clean up if needed.

Example: run_command(command=”pwd && ls”, working_directory=”.”)

Parameters:

command – Shell command string to run.
working_directory – Directory to run in; must be inside the project.
timeout_seconds – Seconds before the command is killed.
max_output_chars – Maximum characters captured per output stream.

Returns:

Dictionary with command, working_directory, returncode, stdout, stderr, truncated, error.

async klea_utils.mcp.server.bundled_tools.search_papers(ctx: fastmcp.Context, query: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="Keywords to search for, e.g. 'hodgkin huxley'", metadata=[MinLen(min_length=1)])], domain: Annotated[Literal['general', 'life-sciences'], FieldInfo(annotation=NoneType, required=True, description="'general' for all fields, or 'life-sciences' to search biomedical literature (Europe PMC, PubMed) first")] = 'general', preprints: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Also search preprints (not peer reviewed). Set only when the user asks for preprints, or the peer reviewed results are not enough')] = False, max_results: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Maximum number of peer reviewed results, and separately of preprint results', metadata=[Ge(ge=1), Le(le=25)])] = 10) → fastmcp.tools.ToolResult[source]¶

Search academic papers by keywords.

Use this tool to find published research papers on a topic, with their title, authors, year, journal, DOI, link and abstract. Peer reviewed papers are searched first; preprints are only included when preprints is true, and are marked with peer_reviewed: false.

Use when: - Finding papers or references on a scientific topic. - Looking for the source of a model, method or result.

Do not use for: - Reading a paper’s full text (use the web fetch or download tool). - General web pages, news or docs (use the web search tool instead).

When you cite a preprint, tell the user it has not been peer reviewed.

Example: search_papers(query=”hodgkin huxley”, domain=”life-sciences”)

Parameters:

query – Keywords to search for.
domain – ‘general’ or ‘life-sciences’.
preprints – Also search preprints, listed after the peer reviewed results.
max_results – Maximum results from the peer reviewed sources and preprints.

Returns:

Dictionary with query, domain, preprints, results (title, authors, year, journal, abstract, doi, url, peer_reviewed, source), error and note.

async klea_utils.mcp.server.bundled_tools.search_stores(ctx: fastmcp.Context, query: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Search string.')] = '', domain: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Domain to search, or empty to list the catalogue.')] = '', mode: Annotated[str, FieldInfo(annotation=NoneType, required=True, description="One of 'hybrid', 'vector', 'bm25' or 'catalogue'.")] = 'hybrid', top_k: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Documents fetched per store', metadata=[Ge(ge=1), Le(le=50)])] = 5, rerank: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Re-rank the fused results with the cross-encoder')] = True, metadata_filter: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Optional metadata filter in the store filter DSL')] = None, max_chars: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Character budget for the returned reference material', metadata=[Ge(ge=1), Le(le=200000)])] = 20000) → fastmcp.tools.ToolResult[source]¶

Search the deployment’s configured document stores for one domain.

Use this tool to answer a question from the deployment’s document collections: it queries the domain’s vector and BM25 stores, fuses their results, re-ranks them and returns character-budgeted reference material.

Use when: - Answering a question from the deployment’s document collections. - Looking up material held in a vector store or BM25 corpus.

Do not use for: - Reading a local file (use the read file tool instead). - General web pages (use the web search tool instead).

Call it with mode ‘catalogue’ (or an empty domain) to list the domains with the filter fields each accepts.

Parameters:

query – Search string; required unless mode is ‘catalogue’.
domain – Domain to search, or empty to list the catalogue.
mode – One of ‘hybrid’, ‘vector’, ‘bm25’ or ‘catalogue’.
top_k – Documents fetched per store (1-50).
rerank – Re-rank the fused results with the cross-encoder.
metadata_filter – Optional metadata filter in the store filter DSL.
max_chars – Character budget for the returned reference material.

Returns:

Dictionary with query, domain, mode, top_k, results, count, content, catalogue, truncated, error and note.

async klea_utils.mcp.server.bundled_tools.web_fetch(ctx: fastmcp.Context, url: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1.0), Le(le=120.0)])] = 30.0, max_chars: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1), Le(le=1000000)])] = 100000) → fastmcp.tools.ToolResult[source]¶

Fetch a URL and return its text content.

Use this tool to read web pages, docs, or other HTTP resources.

Use when: - Reading a page or document from the web. - Checking a URL that a user or another tool referenced.

Do not use for: - Downloading a file to disk (use the download file tool instead).

Example: web_fetch(url=”https://example.com”)

Parameters:

url – HTTP or HTTPS URL to fetch.
timeout – Request timeout in seconds.
max_chars – Maximum number of characters of content to return.

Returns:

Dictionary with url, status_code, content_type, content, truncated, error.

async klea_utils.mcp.server.bundled_tools.web_search(ctx: fastmcp.Context, query: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], max_results: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1), Le(le=20)])] = 8) → fastmcp.tools.ToolResult[source]¶

Search the web and return ranked results with snippets.

Use this tool to discover current information, documentation, or sources online, then read a chosen result with the web fetch tool.

Use when: - You need to find information on the web (discovery). - You want links and snippets before fetching a page.

Do not use for: - Reading the content of a specific URL (use the web fetch tool instead). - Downloading a file to disk (use the download file tool instead).

Example: web_search(query=”NeuroML specification”)

Parameters:

query – Free-text search query.
max_results – Maximum number of results to return (1-20).

Returns:

Dictionary with query, provider, results (list of title, url, snippet, published, score), and error.

async klea_utils.mcp.server.bundled_tools.write_file(path: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], content: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Complete content to write to the file')]) → fastmcp.tools.ToolResult[source]¶

Create a new file or overwrite an existing one with the given content.

Use this tool to create a file, or to replace a file’s entire contents. The write is atomic and an existing file’s mode is preserved.

Use when:

Creating a new file.
Replacing the whole content of a small file.

Do not use for:

Reading a file (use the read file tool instead).
Listing a directory (use the list files tool instead).
Making a small change to an existing file (use the edit file tool instead).

Example: write_file(path=”notes.txt”, content=”hello world”)

Parameters:

path – File path to write, relative to the project directory. Missing parent directories are created.
content – Complete file content.

Returns:

Dictionary with path, created, bytes_written, additions, deletions, diff, error.