MCP utilities¶

Shared machinery for building MCP servers and clients used by Klea apps: metadata schemas, tool registration, an httpx session lifespan, path permission checks, 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 = '')[source]¶

Bases: BaseModel

A single tool call selected by a tools picker node.

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

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

Bases: BaseModel

The structured output of a tools picker node: a list of tool calls.

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.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.

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

async klea_utils.mcp.dispatch.dispatch_tool_calls(mcp_client: Any, tool_calls: list[tuple[str, dict[str, Any]]], tools_meta: dict[str, dict[str, Any]] | None = None, project_root: str | None = None) → list[fastmcp.client.client.CallToolResult][source]¶

Gate and dispatch tool calls against an MCP server.

For each (tool, args) pair, the tool’s meta (from tools_meta) is checked with check_tool_arguments_permissions() before the call reaches the server; denied calls never touch the server and instead produce a synthetic non-halting error result. 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 – (tool name, arguments) pairs to invoke.
tools_meta – Mapping of tool name to the tool’s meta dict (e.g. {t.name: t.meta for t in mcp_tools}). Path arguments declared under checkpaths are permission-checked client-side.
project_root – Boundary directory for the permission gate. Defaults to the current working directory.

Returns:

One CallToolResult per input call, in input order.

HTTP session lifespan¶

Shared FastMCP lifespan helpers.

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 permission checking for file-accessing MCP tools.

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) → None[source]¶

Raise PermissionDeniedError when path is not permitted.

The permission layer is currently a stub: path is allowed only when it resolves inside project_root (default: the current working directory). Both sides are fully resolved first, so .. traversal and symlink escapes outside the boundary are caught. Anything else is denied with no way to grant access yet.

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.

Raises:

PermissionDeniedError – when path resolves outside the boundary.

klea_utils.mcp.tool_impls.permission.check_tool_arguments_permissions(tool_meta: dict[str, Any] | None, arguments: dict[str, Any], project_root: str | PathLike | None = None) → list[str][source]¶

Check the path arguments a tool call would pass against the boundary.

Reads the checkpaths key from tool_meta (the meta dict of an MCP tool, populated by register_tools from ToolInfo.checkpaths). For each declared argument name that is present in arguments, the value is checked with check_path_access(). Unlike check_path_access(), this never raises: denied paths are collected and returned as human-readable messages so the caller (the tool caller node) can turn them into a non-halting error result without invoking the tool.

Values that are not strings or path-like (e.g. an int) are skipped with a warning, so a mistyped declaration cannot crash the gate.

Parameters:

tool_meta – Tool meta dict (Tool.meta from mcp_tools), or None/empty when the tool declares nothing.
arguments – The arguments dict the caller intends to pass to the tool.
project_root – Boundary directory for the permission check. Defaults to the current working directory.

Returns:

List of denial messages; empty when all declared paths are permitted (or no checkpaths are declared).

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.

Web fetch implementation for Klea MCP tools.

File: klea_utils/mcp/tool_impls/web_fetch.py

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.

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.
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 – Boundary directory for the permission check. Defaults to the current working directory.

Returns:

dict with files, error, truncated.

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, limit: int | None = 2000, max_chars: int = 100000, max_bytes: int = 104857600, 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.

Parameters:

path – File path to read.
offset – 1-indexed line to start reading from.
limit – Maximum number of lines to return. None reads to the end of the file.
max_chars – Hard cap on characters returned, applied after the line slice.
max_bytes – Maximum file size in bytes to read; larger files are refused with an error.
project_root – Boundary directory for the permission check. Defaults to the current working directory.

Returns:

dict with path, content, line_start, line_end, total_lines, truncated, error.

File download implementation for Klea MCP tools.

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 – Boundary directory for the permission check. Defaults to the current working directory.
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).

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.

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.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", 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).

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.
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, and error.

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", metadata=[MinLen(min_length=1)])], offset: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='1-indexed line to start reading from', metadata=[Ge(ge=1)])] = 1, limit: Annotated[int | None, FieldInfo(annotation=NoneType, required=True, description="Maximum number of lines to return. 'None' for end of file")] = 2000, max_chars: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Hard cap on characters of content to return', metadata=[Ge(ge=1)])] = 100000) → 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.

Do not use for: - Listing a directory (use the list files tool instead). - Fetching remote content (use the web fetch tool instead).

Example: read_file(path=”README.md”, offset=1, limit=100)

Parameters:

path – File path to read. Must be relative to the current working directory and cannot contain ‘..’ for security.
offset – 1-indexed line to start reading from.
limit – Maximum number of lines to return. None reads to the end.
max_chars – Hard cap on characters of content to return.

Returns:

Dictionary with content, line range, total_lines, truncated, error.

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.