Recipes and the Tool Layer#

The machinery that exposes the package to an LLM agent: the analysis-recipe framework, the @tool_spec decorator and its registry, the settings singleton, the subprocess isolation used when a tool call arrives over an untrusted transport, and the MCP server that publishes the registered tools. The narrative overview is in Architecture overview.

Analysis recipes#

Kevin Dunn, 2010-2026. MIT License.

Reusable analysis-recipe framework.

An analysis recipe is a predefined, step-by-step workflow an LLM agent can follow when a user’s request matches a known analytical scenario (intake, panel processing, relating to covariates, …). Recipes reference existing agent tools by name so the agent chains calls deterministically instead of improvising the order.

The framework is package-wide and domain-agnostic: any subpackage may define its own recipes and register them. The sensory subpackage is the first consumer (see process_improve.sensory.recipes).

Adding recipes for a subpackage#

Create process_improve/<subpackage>/recipes.py.
Build AnalysisRecipe instances and pass each to register_recipe().
Add "process_improve.<subpackage>.recipes" to _RECIPE_MODULES below so discover_recipes() imports it.

No changes to the agent tool layer are required: the single, general select_analysis_recipe tool matches across every registered recipe.

class process_improve.recipes.AnalysisRecipe(key, title, summary, domain, cue_phrases, inputs_needed, stages, status='available')[source]#

Bases: object

A reusable, multi-step analysis workflow for the agent.

Parameters:

key (str)
title (str)
summary (str)
domain (str)
cue_phrases (list[str])
inputs_needed (list[str])
stages (list[RecipeStep])
status (str)

key#

Unique snake_case identifier.

Type:: str

title#

Human-readable name.

Type:: str

summary#

One-paragraph description of what the recipe does and when to use it.

Type:: str

domain#

The subpackage the recipe belongs to (e.g. "sensory").

Type:: str

cue_phrases#

Lower-case substrings; each one found in a user’s request scores the recipe one point during matching.

Type:: list of str

inputs_needed#

What the agent must resolve from the user before running, each with a short example.

Type:: list of str

stages#

The ordered steps. Empty for a planned (not yet available) recipe.

Type:: list of RecipeStep

status#

"available" (default) or "planned" for parked future work.

Type:: str

to_payload()[source]#

Serialise to a JSON-friendly dict the agent can consume.

Return type:: dict[str, Any]

class process_improve.recipes.RecipeStep(order, directive, tools=<factory>, arg_hints=<factory>)[source]#

Bases: object

One step in an analysis recipe the agent should execute.

Parameters:

order (int)
directive (str)
tools (list[str])
arg_hints (dict[str, str])

order#

1-based position of the step in the recipe.

Type:: int

directive#

Natural-language instruction telling the agent what to do.

Type:: str

tools#

Names of agent tools this step may call (empty for prose-only steps such as interpretation or data assembly).

Type:: list of str

arg_hints#

Optional {parameter: "where the value comes from"} hints, for example {"score_min": "0", "mode": "observational"}.

Type:: dict

process_improve.recipes.get_recipe(key)[source]#

Return the registered recipe with key, or None.

Parameters:: key (str)
Return type:: AnalysisRecipe | None

process_improve.recipes.list_recipes()[source]#

Return every registered recipe (registration order).

Return type:: list[AnalysisRecipe]

process_improve.recipes.register_recipe(recipe)[source]#

Raises:: ValueError – If a recipe with the same key is already registered.
Parameters:: recipe (AnalysisRecipe)
Return type:: AnalysisRecipe

process_improve.recipes.select_analysis_recipe(spec)[source]#

Return the best-matching recipe payload plus the full catalogue.

Parameters:: spec (_RecipeQuery)
Return type:: dict

process_improve.recipes.select_recipe(query)[source]#

Return the best-matching recipe for query, or None.

Scoring is intentionally simple: each cue phrase that appears as a substring of the canonicalised query contributes one point. The highest-scoring recipe wins (ties broken by registration order); a minimum score of one is required to match. Replace with embedding similarity if the catalog grows beyond a few dozen recipes.

Parameters:: query (str)
Return type:: AnalysisRecipe | None

Tool specifications and registry#

Kevin Dunn, 2010-2026. MIT License.

Tool-call-first infrastructure for process-improve.

Provides the @tool_spec decorator, a global registry of all decorated functions, and helpers for Anthropic-compatible tool-use integrations.

Quick start#

Import the decorated tools and pass the specs to the Anthropic client:

import anthropic
from process_improve.tool_spec import get_tool_specs, execute_tool_call

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    tools=get_tool_specs(),
    messages=[{"role": "user", "content": "Are there outliers in [1,2,3,100]?"}],
)

# Dispatch tool calls from the response
for block in response.content:
    if block.type == "tool_use":
        result = execute_tool_call(block.name, block.input)

exception process_improve.tool_spec.ToolInputInvalidError(message, *, details=None)[source]#

Bases: ToolSafetyError

Input failed structural validation (unexpected types, bad shape).

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

exception process_improve.tool_spec.ToolInputTooLargeError(message, *, details=None)[source]#

Bases: ToolSafetyError

Input exceeded an allowed size limit (cells, string length, depth).

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

exception process_improve.tool_spec.ToolMemoryExceededError(message, *, details=None)[source]#

Bases: ToolSafetyError

Subprocess was killed, most likely by the memory limit.

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

exception process_improve.tool_spec.ToolSafetyError(message, *, details=None)[source]#

Bases: Exception

Base class for safety-related tool-execution failures.

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

to_dict()[source]#

Return a JSON-serialisable representation of this error.

Return type:: dict[str, Any]

exception process_improve.tool_spec.ToolTimeoutError(message, *, details=None)[source]#

Bases: ToolSafetyError

Tool call exceeded the wall-clock timeout.

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

process_improve.tool_spec.clean(value)[source]#

Recursively convert numpy scalars / arrays to plain Python types.

All tools.py modules should call clean(result) before returning so that every tool output is JSON-serialisable.

Parameters:: value (Any)
Return type:: Any

process_improve.tool_spec.discover_tools()[source]#

Import all tools.py modules to populate the tool registry.

This is called lazily on the first get_tool_specs() invocation. It is safe to call multiple times (subsequent calls are no-ops).

Return type:: None

process_improve.tool_spec.execute_tool_call(tool_name, tool_input)[source]#

Dispatch a single tool call from an Anthropic tool_use content block.

The input dict is validated via input_model.model_validate(tool_input). Unknown keys raise ToolInputInvalidError (closes the SEC-15 confirmed=True kwarg-injection at the schema layer). The parsed pydantic model is passed to the tool function as a single positional argument.

Parameters:

tool_name (str) – The name field from the tool_use block.
tool_input (dict[str, Any]) – The input dict from the tool_use block.

Returns:

Whatever the tool function returns (typically a JSON-serialisable dict).

Return type:

Any

Raises:

ValueError – If tool_name is not in the registry.
ToolInputInvalidError – If the input fails pydantic validation.

process_improve.tool_spec.get_tool_specs(names=None, category=None)[source]#

Return tool specs in the format expected by the Anthropic tools= parameter.

Parameters:

names (list[str] | None) – Optional allow-list of tool names to include. When None (default) all registered tools are returned.
category (str | None) – Optional category filter (e.g. "univariate"). When provided, only tools whose category matches are returned.

Returns:

Each dict has keys "name", "description", and "input_schema" as required by the Anthropic API. Tools that opt in via rng= on the decorator also carry an "rng" key describing their reproducibility contract; see tool_spec().

Return type:

list[dict]

process_improve.tool_spec.safe_execute_tool_call(tool_name, tool_input, *, timeout=None, max_cells=None, max_string=None, max_depth=None, memory_mb=None, executor=None)[source]#

Execute a tool call with input validation, timeout, and memory cap.

Parameters:

tool_name (str) – Same meaning as process_improve.tool_spec.execute_tool_call().
tool_input (dict[str, Any]) – Same meaning as process_improve.tool_spec.execute_tool_call().
timeout (float | None) – Wall-clock seconds. On overrun the runaway worker is force-terminated (terminate() then kill()) so it cannot keep holding a CPU, and ToolTimeoutError is raised.
max_cells (int | None) – Input-size limits. See validate_input().
max_string (int | None) – Input-size limits. See validate_input().
max_depth (int | None) – Input-size limits. See validate_input().
memory_mb (int | None) – RSS cap applied to the worker subprocess via RLIMIT_AS (POSIX). On overrun the subprocess dies and ToolMemoryExceededError is raised.
executor (ProcessPoolExecutor | None) – Optional caller-provided pool. When None (default) a PRIVATE pool is created for this call and torn down afterwards, so each call runs in a fresh worker with isolated process-global state and reclaimed memory, and concurrent calls (e.g. from the threaded MCP server) never share or tear down each other’s workers. A caller-provided executor is never recycled or terminated by this function - the caller owns its lifecycle.

Raises:

ToolInputInvalidError, ToolInputTooLargeError: – Synchronous rejection before any subprocess work. ToolInputInvalidError also covers JSON-schema violations (wrong type, out-of-bounds value, bad enum, missing required key, or an unknown parameter).
ToolTimeoutError: – Wall-clock overrun.
ToolMemoryExceededError: – Worker subprocess died unexpectedly (likely OOM).
ValueError: – Unknown tool name (propagated from execute_tool_call).

Return type:

Any

process_improve.tool_spec.tool_spec(name, description, *, input_model, examples='', category='', rng=None)[source]#

Mark a function as an agent-callable tool.

Parameters:

name (str) – Unique tool name exposed to the LLM (snake_case).
description (str) – Natural-language description of what the tool does and when to use it.
input_model (type[BaseModel]) –
A pydantic.BaseModel subclass describing the tool’s input. The model is the single source of truth for both the function’s Python signature and the MCP JSON Schema (derived via BaseModel.model_json_schema()). Every input model must set model_config = ConfigDict(extra="forbid") so unknown keys are rejected (closes the SEC-15 kwarg-injection vector at the schema layer).

The decorated function then receives the parsed model as its single positional argument:
```
class FooInput(BaseModel):
    model_config = ConfigDict(extra="forbid")
    values: list[float] = Field(..., min_length=2)

@tool_spec(name="foo", description="...", input_model=FooInput)
def foo(spec: FooInput) -> dict:
    return {"n": len(spec.values)}
```
examples (str) – Optional natural-language -> tool-call mappings appended to description so the LLM sees worked examples.
category (str) – Optional category string (e.g. "univariate"). Used for filtering with get_tool_specs().
rng (dict[str, Any] | None) – Optional reproducibility metadata; see _validate_rng_metadata() for the contract.

Returns:

The original function, with an attached _tool_spec dict and _input_model attribute.

Return type:

Callable

Configuration#

Kevin Dunn, 2010-2026. MIT License.

Central configuration for process-improve.

Closes ENG-09 (#291) – “configuration sprawl: env vars at import time, magic numbers, no central config” – and ENG-27 (#309) – “tool_safety.py reads env vars at import time”.

The settings singleton is the single place every other module reads configurable knobs from. Each knob:

has a documented default,
can be overridden by the corresponding environment variable (PROCESS_IMPROVE_*), and
is read on first access, not at import time. Tests can call Settings.reload() to re-read after mutating the environment; callers can override a single knob in code via settings.tool_timeout = 5.0 (the setter writes through and bypasses the cache).

Usage#

>>> from process_improve.config import settings
>>> settings.tool_timeout
10.0
>>> settings.mcp_safe_mode
False

Override via environment:

PROCESS_IMPROVE_TOOL_TIMEOUT=30 python -m process_improve.mcp_server

Override in code (e.g. in a test):

from process_improve.config import settings
monkeypatch.setenv("PROCESS_IMPROVE_TOOL_TIMEOUT", "5")
settings.reload()
assert settings.tool_timeout == 5.0

Why a class rather than module-level globals?#

Module-level globals are read once at import; a test that mutates os.environ after import has no effect. The Settings class caches values on first access and exposes reload() for the test case. That matches what every other module’s behaviour should be.

The class deliberately does not depend on pydantic-settings (or any new third-party package). pydantic is already a hard dep (ENG-04 is the open decision about whether to commit to it everywhere); a plain class with explicit env-var reads keeps that decision unfettered.

process_improve.config.DEFAULTS: Final[dict[str, Any]] = {'dataset_fetch_timeout': 30.0, 'max_cells': 1000000, 'max_depth': 10, 'max_factors_combinatorial': 15, 'max_formula_chars': 4096, 'max_formula_terms': 100, 'max_matrix_cols': 500, 'max_matrix_rows': 10000, 'max_memory_mb': 1024, 'max_regression_points': 5000, 'max_string': 100000, 'mcp_safe_mode': False, 'tool_timeout': 10.0}#: Default knob values. Kept here so Settings.reload() knows what to fall back to when an env var is unset, and so DEFAULTS is the single place to read the canonical defaults from.

process_improve.config.ENV_VAR_NAMES: Final[dict[str, str]] = {'dataset_fetch_timeout': 'PROCESS_IMPROVE_DATASET_FETCH_TIMEOUT', 'max_cells': 'PROCESS_IMPROVE_MAX_CELLS', 'max_depth': 'PROCESS_IMPROVE_MAX_DEPTH', 'max_factors_combinatorial': 'PROCESS_IMPROVE_MAX_FACTORS_COMBINATORIAL', 'max_formula_chars': 'PROCESS_IMPROVE_MAX_FORMULA_CHARS', 'max_formula_terms': 'PROCESS_IMPROVE_MAX_FORMULA_TERMS', 'max_matrix_cols': 'PROCESS_IMPROVE_MAX_MATRIX_COLS', 'max_matrix_rows': 'PROCESS_IMPROVE_MAX_MATRIX_ROWS', 'max_memory_mb': 'PROCESS_IMPROVE_MAX_MEMORY_MB', 'max_regression_points': 'PROCESS_IMPROVE_MAX_REGRESSION_POINTS', 'max_string': 'PROCESS_IMPROVE_MAX_STRING', 'mcp_safe_mode': 'PROCESS_IMPROVE_MCP_SAFE_MODE', 'tool_timeout': 'PROCESS_IMPROVE_TOOL_TIMEOUT'}#: Mapping from knob name to environment-variable name. tool_safety’s original env-var contract is preserved verbatim so existing deployments keep working without modification.

class process_improve.config.Settings[source]#

Bases: object

Single-instance configuration store.

Every attribute is a knob; reads are cached after the first access. Call reload() after mutating os.environ (typically inside a test fixture); call override() to set a single knob from code.

property tool_timeout: float#: Wall-clock seconds budget for a single tool call.

property dataset_fetch_timeout: float#

Wall-clock seconds budget for downloading one remote sample dataset.

Bounds the urlopen call in process_improve._remote_data.fetch_remote_bytes(), so a black-holing host raises the module’s documented RuntimeError instead of hanging the caller indefinitely (#508).

property max_cells: int#: Maximum number of numeric leaves anywhere in a tool input payload.

property max_string: int#: Maximum length of any single string in a tool input payload.

property max_depth: int#: Maximum nesting depth of any tool input payload.

property max_memory_mb: int#: Per-subprocess RSS cap for tool execution (MiB).

property mcp_safe_mode: bool#

Whether the MCP server should treat its transport as untrusted.

When True, every tool call goes through process_improve.tool_safety.safe_execute_tool_call() (validation, subprocess isolation, memory cap).

property max_factors_combinatorial: int#: Maximum k for combinatorial design generators (ff2n, fullfact, simplex centroid / lattice). Default 15 caps 2**k rows at ~32 KiB of memory per cell.

property max_regression_points: int#: Maximum len(x) / len(y) for the O(N^2) regression kernels (repeated_median_slope etc.).

property max_matrix_rows: int#: Maximum row count for data / x_data matrix inputs to fit_pca / fit_pls / detect_multivariate_outliers.

property max_matrix_cols: int#: Maximum column count for matrix inputs to the multivariate tools.

property max_formula_chars: int#: Maximum length (chars) of a model-formula string accepted by fit_linear_model and analyze_experiment.

property max_formula_terms: int#: Maximum number of terms after patsy expansion of a model formula.

reload()[source]#

Drop the cache so the next attribute access re-reads from env.

Return type:: None

as_dict()[source]#

Return a snapshot of every knob’s current value.

Triggers a read of every knob (populating the cache); useful in --verbose startup banners and for printing the effective configuration.

Return type:: dict[str, Any]

process_improve.config.settings: Settings = <process_improve.config.Settings object>#: The single module-level Settings instance. Every other module imports this object directly.

Tool-call safety#

Kevin Dunn, 2010-2026. MIT License.

Safe execution wrapper for process-improve tool calls.

Adds the four guard rails needed to expose the tool registry over an untrusted transport (public MCP server, hosted REST API, etc.):

Input-size validation (reject oversize arrays/strings before work).
Wall-clock timeout via subprocess isolation.
Memory cap per subprocess (POSIX; best-effort on Windows).
Structured error types so callers can distinguish failure modes.

The in-process process_improve.tool_spec.execute_tool_call() is left untouched for callers that trust their input (notebooks, tests, the stdio MCP server running on the user’s own machine). Hosted callers should use safe_execute_tool_call() instead.

Configuration#

Every knob is read lazily from process_improve.config.settings, which in turn picks up the corresponding PROCESS_IMPROVE_* environment variable on first access (ENG-09 / ENG-27). Tests can override a knob in-process via settings.tool_timeout = 5.0; CI deployments can still set env vars at startup. See process_improve/config.py for the canonical defaults table.

exception process_improve.tool_safety.ToolSafetyError(message, *, details=None)[source]#

Bases: Exception

Base class for safety-related tool-execution failures.

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

to_dict()[source]#

Return a JSON-serialisable representation of this error.

Return type:: dict[str, Any]

exception process_improve.tool_safety.ToolInputTooLargeError(message, *, details=None)[source]#

Bases: ToolSafetyError

Input exceeded an allowed size limit (cells, string length, depth).

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

exception process_improve.tool_safety.ToolInputInvalidError(message, *, details=None)[source]#

Bases: ToolSafetyError

Input failed structural validation (unexpected types, bad shape).

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

exception process_improve.tool_safety.ToolTimeoutError(message, *, details=None)[source]#

Bases: ToolSafetyError

Tool call exceeded the wall-clock timeout.

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

exception process_improve.tool_safety.ToolMemoryExceededError(message, *, details=None)[source]#

Bases: ToolSafetyError

Subprocess was killed, most likely by the memory limit.

Parameters:

message (str)
details (dict[str, Any] | None)

Return type:

None

process_improve.tool_safety.validate_input(tool_input, *, max_cells=None, max_string=None, max_depth=None, scalar_caps=None)[source]#

Raise ToolInputTooLargeError if tool_input breaks any limit.

Parameters:

tool_input (dict[str, Any]) – The input dict that would be passed as keyword arguments to the tool function.
max_cells (int | None) – Maximum number of numeric leaves anywhere in the payload.
max_string (int | None) – Maximum length of any single string value.
max_depth (int | None) – Maximum nesting depth for dicts/lists.
scalar_caps (dict[str, float] | None) – Override the default per-key numeric caps (see _SCALAR_CAPS).

Return type:

None

process_improve.tool_safety.get_pool(memory_mb=None, max_workers=1)[source]#

Return a lazily-initialised module-level ProcessPoolExecutor.

The pool is recreated if memory_mb changes (e.g. tests override it). Passing None resolves the cap from settings.max_memory_mb at call time (ENG-09 / ENG-27). Thread-safe.

Parameters:

memory_mb (int | None)
max_workers (int)

Return type:

ProcessPoolExecutor

process_improve.tool_safety.shutdown_pool()[source]#

Shut down the module-level pool, if any. Safe to call repeatedly.

Worker processes are force-terminated first so a runaway task cannot keep holding a CPU after the executor is torn down. Thread-safe.

Return type:: None

process_improve.tool_safety.safe_execute_tool_call(tool_name, tool_input, *, timeout=None, max_cells=None, max_string=None, max_depth=None, memory_mb=None, executor=None)[source]#

Execute a tool call with input validation, timeout, and memory cap.

Parameters:

tool_name (str) – Same meaning as process_improve.tool_spec.execute_tool_call().
tool_input (dict[str, Any]) – Same meaning as process_improve.tool_spec.execute_tool_call().
timeout (float | None) – Wall-clock seconds. On overrun the runaway worker is force-terminated (terminate() then kill()) so it cannot keep holding a CPU, and ToolTimeoutError is raised.
max_cells (int | None) – Input-size limits. See validate_input().
max_string (int | None) – Input-size limits. See validate_input().
max_depth (int | None) – Input-size limits. See validate_input().
memory_mb (int | None) – RSS cap applied to the worker subprocess via RLIMIT_AS (POSIX). On overrun the subprocess dies and ToolMemoryExceededError is raised.
executor (ProcessPoolExecutor | None) – Optional caller-provided pool. When None (default) a PRIVATE pool is created for this call and torn down afterwards, so each call runs in a fresh worker with isolated process-global state and reclaimed memory, and concurrent calls (e.g. from the threaded MCP server) never share or tear down each other’s workers. A caller-provided executor is never recycled or terminated by this function - the caller owns its lifecycle.

Raises:

ToolInputInvalidError, ToolInputTooLargeError: – Synchronous rejection before any subprocess work. ToolInputInvalidError also covers JSON-schema violations (wrong type, out-of-bounds value, bad enum, missing required key, or an unknown parameter).
ToolTimeoutError: – Wall-clock overrun.
ToolMemoryExceededError: – Worker subprocess died unexpectedly (likely OOM).
ValueError: – Unknown tool name (propagated from execute_tool_call).

Return type:

Any

MCP server#

Requires the optional mcp extra; the docs environment installs every extra, so autodoc imports it directly rather than mocking it.

Kevin Dunn, 2010-2026. MIT License.

MCP (Model Context Protocol) server for process-improve.

Exposes all @tool_spec-decorated functions as MCP tools, making them instantly available to Claude Desktop, Cursor, VS Code Copilot, and any other MCP-compatible client.

Usage#

Run directly:

python -m process_improve.mcp_server

Or via the installed entry-point:

process-improve-mcp

Configuration for Claude Desktop (claude_desktop_config.json):

{
    "mcpServers": {
        "process-improve": {
            "command": "process-improve-mcp"
        }
    }
}

process_improve.mcp_server.create_server()[source]#

Create the MCP server with every @tool_spec tool registered.

Returns:: A server whose list_tools() publishes, for each registered tool, the same input_schema that process_improve.tool_spec.get_tool_specs() reports.
Return type:: MCPServer

process_improve.mcp_server.main()[source]#

Entry point for the MCP server.

Return type:: None