tools

`dsi_search_tools` ¶

`load_dsi_tool(db_path, run_path='', master_db_folder='')` ¶

Load a DSI object from the path and add information to the context for the llm to use.

Parameters:

Name	Type	Description	Default
`db_path`	`str`	the path to the DSI object to load	required
`run_path`	`str`	the path this code is being run from	`''`
`master_db_folder`	`str`	the folder containing the master database, used to resolve relative paths when loading new databases	`''`

Returns:

Name	Type	Description
`str`	`str`	message indicating success or failure

Source code in src/ursa/tools/dsi_search_tools.py

@tool
def load_dsi_tool(
    db_path: str, run_path: str = "", master_db_folder: str = ""
) -> str:
    """Load a DSI object from the path and add information to the context for the llm to use.

    Args:
        db_path (str): the path to the DSI object to load
        run_path (str): the path this code is being run from
        master_db_folder (str): the folder containing the master database, used to resolve relative paths when loading new databases

    Returns:
        str: message indicating success or failure
    """
    if no_DSI:
        return "Optional dependence [dsi] not installed. Tool will not work."

    master_database_previously_set = True
    if master_db_folder == "":
        master_database_previously_set = False
        # the ai is loading the master database for the first time
        master_database_path, master_db_folder = _get_db_abs_path(
            db_path, run_path
        )
        data_path = master_database_path.strip()
    else:
        p = Path(db_path).expanduser()
        if not p.is_absolute():
            _db_path = str(Path(master_db_folder).expanduser() / p)
        else:
            _db_path = str(p)

        data_path = _db_path.strip()

    if not _check_db_valid(data_path):
        return f"Failed to load DSI database at: {data_path}. Please check the path and ensure it points to a valid DSI .db file."

    try:
        _, _db_schema, _db_description = _get_db_info(data_path)
        _current_db_abs_path = data_path
        schema_text = json.dumps(_db_schema, indent=2, ensure_ascii=False)

        if master_database_previously_set is False:
            return f"""
- Current working database path (current_db_abs_path): {_current_db_abs_path}
- Master database path (master_database_path): {master_database_path}
- Master database folder (master_db_folder): {master_db_folder}
- Current database schema: {schema_text}
- Database description: {_db_description}
""".strip()
        else:
            return f"""
- Current working database path (current_db_abs_path): {_current_db_abs_path}
- Current database schema: {schema_text}
- Database description: {_db_description}
""".strip()

    except Exception as e:  # noqa: BLE001
        return f"Failed to load database information: {e}"

`query_dsi_tool(query_str, db_path)` ¶

Execute a SQL query on a DSI object

Arg

query_str (str): the SQL query to run on DSI object db_path (str): the absolute path to the DSI database to query

Returns:

Name	Type	Description
`collection`	`dict`	the results of the query

Source code in src/ursa/tools/dsi_search_tools.py

@tool
def query_dsi_tool(query_str: str, db_path: str) -> dict:
    """Execute a SQL query on a DSI object

    Arg:
        query_str (str): the SQL query to run on DSI object
        db_path (str): the absolute path to the DSI database to query

    Returns:
        collection: the results of the query
    """
    if no_DSI:
        return {
            "error": "Optional dependency [dsi] not installed. Tool will not work."
        }

    _store = None
    try:
        with redirect_stdout(_NULL), redirect_stderr(_NULL):
            _store = DSI(db_path, check_same_thread=False)
            df = _store.query(query_str, collection=True)

        if df is None:
            return {}
        return df.to_dict(orient="records")

    except Exception:  # noqa: BLE001
        return {}

    finally:
        if _store is not None:
            try:
                with redirect_stdout(_NULL), redirect_stderr(_NULL):
                    _store.close()
            except Exception:
                LOGGER.exception("Error closing DSI store")

`experience_tools` ¶

`edit_experience(old_content, new_content, filename, runtime)` ¶

Replace the first occurrence of old_content with new_content in a markdown experience file.

Parameters:

Name	Type	Description	Default
`old_content`	`str`	Text fragment to search for.	required
`new_content`	`str`	Replacement text fragment.	required
`filename`	`str`	Markdown filename inside the experiences directory.	required

Returns:

Type	Description
`str`	Success or failure message describing the edit operation.

Source code in src/ursa/tools/experience_tools.py

@tool
def edit_experience(
    old_content: str,
    new_content: str,
    filename: str,
    runtime: ToolRuntime[AgentContext],
) -> str:
    """Replace the first occurrence of old_content with new_content in a markdown experience file.

    Args:
        old_content: Text fragment to search for.
        new_content: Replacement text fragment.
        filename: Markdown filename inside the experiences directory.

    Returns:
        Success or failure message describing the edit operation.
    """
    try:
        filename = validate_ascii(filename)
    except AsciiValidationError as exc:
        return ascii_validation_message("filename", exc)
    experience_file = _experience_path(filename, runtime)
    events = ToolEvents.from_runtime("edit_experience", runtime)
    events.emit(
        "Editing experience",
        stage="edit",
        phase="start",
        path=str(experience_file),
    )

    try:
        content = read_text_file(experience_file)
    except FileNotFoundError:
        events.emit(
            "Experience file not found",
            stage="edit",
            phase="error",
            path=str(experience_file),
        )
        return f"Failed: {filename} not found."
    except ValueError as exc:
        events.emit(
            "Failed to read experience",
            stage="edit",
            phase="error",
            path=str(experience_file),
            error=str(exc),
        )
        return f"Failed to edit {filename}: {exc}"
    except OSError as exc:
        events.emit(
            "Failed to read experience",
            stage="edit",
            phase="error",
            path=str(experience_file),
            error=str(exc),
        )
        return f"Failed to edit {filename}: Could not read file: {exc}"

    if old_content not in content:
        events.emit(
            "No changes made",
            stage="edit",
            phase="end",
            path=str(experience_file),
            reason="'old_content' not found in experience file.",
        )
        return f"No changes made to {filename}: 'old_content' not found in experience file."

    updated = content.replace(old_content, new_content, 1)

    diff = "".join(
        difflib.unified_diff(
            content.splitlines(keepends=True),
            updated.splitlines(keepends=True),
            fromfile=filename,
            tofile=filename,
        )
    )

    try:
        with open(experience_file, "w", encoding="utf-8") as f:
            f.write(updated)
    except OSError as exc:
        events.emit(
            "Failed to edit experience",
            stage="edit",
            phase="error",
            path=str(experience_file),
            error=str(exc),
        )
        return f"Failed to edit {filename}: {exc}"

    events.emit(
        "Experience updated",
        stage="edit",
        phase="end",
        path=str(experience_file),
        artifact=event_artifact(
            diff,
            "text/x-diff",
            metadata={"title": "Experience diff", "path": filename},
        ),
    )

    if (store := runtime.store) is not None:
        store.put(
            ("den", "experience_edit"),
            _relative_to_den(experience_file, runtime),
            {
                "modified": time.time(),
                "tool_call_id": runtime.tool_call_id,
                "thread_id": runtime.config.get("metadata", {}).get(
                    "thread_id", None
                ),
                "append": False,
            },
        )

    return f"Experience file {_relative_to_den(experience_file, runtime)} updated successfully."

`list_experiences(runtime)` ¶

List available markdown experience files stored in the experiences directory.

This helps identify which experience files can be read back into context later.

Returns:

Type	Description
`str`	A newline-separated list of available experience filenames, or a message if none exist.

Source code in src/ursa/tools/experience_tools.py

@tool
def list_experiences(runtime: ToolRuntime[AgentContext]) -> str:
    """List available markdown experience files stored in the experiences directory.

    This helps identify which experience files can be read back into context later.

    Returns:
        A newline-separated list of available experience filenames, or a message if none exist.
    """
    experiences_dir = _experiences_dir(runtime)
    events = ToolEvents.from_runtime("list_experiences", runtime)
    files = sorted(
        path.name
        for path in experiences_dir.iterdir()
        if path.is_file() and path.suffix.lower() == ".md"
    )

    if not files:
        events.emit("No experience files found", stage="list", count=0)
        return "No experience files found in experiences/."

    events.emit(
        "Experience files listed",
        stage="list",
        count=len(files),
        artifact=event_artifact(
            "\n".join(files),
            "text/plain",
            metadata={"title": "Experiences"},
        ),
    )
    return "Available experience files:\n" + "\n".join(
        f"- {name}" for name in files
    )

`read_experience(filename, runtime)` ¶

Read a markdown experience file to recall previously stored context.

This tool loads previously stored notes, experiences, and lessons learned from the experiences directory so they can be brought back into context for the current task.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Markdown filename to read from the experiences directory.	required

Returns:

Type	Description
`str`	The contents of the requested experience file.

Source code in src/ursa/tools/experience_tools.py

@tool
def read_experience(
    filename: str,
    runtime: ToolRuntime[AgentContext],
) -> str:
    """Read a markdown experience file to recall previously stored context.

    This tool loads previously stored notes, experiences, and lessons learned from the
    experiences directory so they can be brought back into context for the current task.

    Args:
        filename: Markdown filename to read from the experiences directory.

    Returns:
        The contents of the requested experience file.
    """
    try:
        filename = validate_ascii(filename)
    except AsciiValidationError as exc:
        return ascii_validation_message("filename", exc)
    experience_file = _experience_path(filename, runtime)
    events = ToolEvents.from_runtime("read_experience", runtime)
    if not experience_file.exists() or not experience_file.is_file():
        events.emit(
            "Experience file not found",
            stage="read",
            phase="error",
            path=str(experience_file),
        )
        return (
            "Experience file not found: "
            f"{_relative_to_den(experience_file, runtime)}"
        )

    try:
        content = read_text_from_file(experience_file)
    except (OSError, ValueError) as exc:
        events.emit(
            "Failed to read experience",
            stage="read",
            phase="error",
            path=str(experience_file),
            error=str(exc),
        )
        return f"Failed to read {filename}: {exc}"
    events.emit(
        "Experience read",
        stage="read",
        phase="end",
        path=str(experience_file),
        artifact=file_artifact(experience_file, title="Experience read"),
    )
    return content

`write_experience(filename, content, runtime, append=True)` ¶

Write or append information to markdown files for later recall.

Use it for lessons learned, task-specific notes, observations, summaries, and other context worth preserving in the future.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Markdown filename to write inside the experiences directory. Must be a .md file!	required
`content`	`str`	Text content to store.	required
`append`	`bool`	If True, append content to the file. If False, overwrite the file.	`True`

Returns:

Type	Description
`str`	Confirmation message describing the write operation.

Source code in src/ursa/tools/experience_tools.py

@tool
def write_experience(
    filename: str,
    content: str,
    runtime: ToolRuntime[AgentContext],
    append: bool = True,
) -> str:
    """Write or append information to markdown files for later recall.

    Use it for lessons learned, task-specific notes, observations, summaries, and other
    context worth preserving in the future.

    Args:
        filename: Markdown filename to write inside the experiences directory. Must be a .md file!
        content: Text content to store.
        append: If True, append content to the file. If False, overwrite the file.

    Returns:
        Confirmation message describing the write operation.
    """
    try:
        filename = validate_ascii(filename)
    except AsciiValidationError as exc:
        return ascii_validation_message("filename", exc)
    experience_file = _experience_path(filename, runtime)
    events = ToolEvents.from_runtime("write_experience", runtime)
    experience_file.parent.mkdir(parents=True, exist_ok=True)

    existing_text = ""
    if experience_file.exists():
        existing_text = read_text_from_file(experience_file)

    text_to_write = content.rstrip() + "\n"
    if append and existing_text:
        separator = "\n\n" if not existing_text.endswith("\n\n") else ""
        updated_text = existing_text + separator + text_to_write
        action = "updated"
    elif append:
        updated_text = text_to_write
        action = "created"
    else:
        updated_text = text_to_write
        action = "overwritten"

    try:
        with events.range(
            "write",
            "Writing experience",
            done="Experience written",
            error="Failed to write experience",
            path=str(experience_file),
        ) as span:
            with open(experience_file, "w", encoding="utf-8") as f:
                f.write(updated_text)
            span.update(
                artifact=file_artifact(
                    experience_file, title="Experience written"
                )
            )
    except OSError as exc:
        return f"Failed to write {filename}: {exc}"

    if (store := runtime.store) is not None:
        store.put(
            ("den", "experience_edit"),
            _relative_to_den(experience_file, runtime),
            {
                "modified": time.time(),
                "tool_call_id": runtime.tool_call_id,
                "thread_id": runtime.config.get("metadata", {}).get(
                    "thread_id", None
                ),
                "append": append,
            },
        )

    return f"Experience file {_relative_to_den(experience_file, runtime)} {action} successfully."

`feasibility_checker` ¶

`heuristic_feasibility_check(constraints, variable_name, variable_type, variable_bounds, samples=10000)` ¶

A tool for checking feasibility of the constraints.

Parameters:

Name	Type	Description	Default
`constraints`	`Annotated[list[str], "List of strings like 'x0+x1<=5'"]`	list of strings like 'x0 + x1 <= 5', etc.	required
`variable_name`	`Annotated[list[str], "List of strings like 'x0', 'x1', etc."]`	list of strings containing variable names used in constraint expressions.	required
`variable_type`	`Annotated[list[str], "List of strings like 'real', 'integer', 'boolean', etc."]`	list of strings like 'real', 'integer', 'boolean', etc.	required
`variable_bounds`	`Annotated[list[list[float]], "List of (lower bound, upper bound) tuples for x0, x1, ...'"]`	list of (lower, upper) tuples for x0, x1, etc.	required
`samples`	`Annotated[int, 'Number of random sample. Default 10000']`	number of random samples, default value 10000	`10000`

Returns:

Type	Description
`tuple[str]`	A string indicating whether a feasible solution was found.

Source code in src/ursa/tools/feasibility_checker.py

@tool(parse_docstring=True)
def heuristic_feasibility_check(
    constraints: Annotated[list[str], "List of strings like 'x0+x1<=5'"],
    variable_name: Annotated[
        list[str], "List of strings like 'x0', 'x1', etc."
    ],
    variable_type: Annotated[
        list[str], "List of strings like 'real', 'integer', 'boolean', etc."
    ],
    variable_bounds: Annotated[
        list[list[float]],
        "List of (lower bound, upper bound) tuples for x0, x1, ...'",
    ],
    samples: Annotated[int, "Number of random sample. Default 10000"] = 10000,
) -> tuple[str]:
    """
    A tool for checking feasibility of the constraints.

    Args:
        constraints: list of strings like 'x0 + x1 <= 5', etc.
        variable_name: list of strings containing variable names used in constraint expressions.
        variable_type: list of strings like 'real', 'integer', 'boolean', etc.
        variable_bounds: list of (lower, upper) tuples for x0, x1, etc.
        samples: number of random samples, default value 10000

    Returns:
        A string indicating whether a feasible solution was found.
    """

    symbols = sp.symbols(variable_name)

    # Build a dict mapping each name to its Symbol, for parsing
    locals_map = {name: sym for name, sym in zip(variable_name, symbols)}

    # Parse constraints into Sympy Boolean expressions
    parsed_constraints = []
    try:
        for expr in constraints:
            parsed = parse_expr(
                expr,
                local_dict=locals_map,
                transformations=standard_transformations,
                evaluate=False,
            )
            parsed_constraints.append(parsed)
    except Exception as e:
        return f"Error parsing constraints: {e}"

    # Sampling loop
    n = len(parsed_constraints)
    funcs = [
        sp.lambdify(symbols, c, modules=["math", "numpy"])
        for c in parsed_constraints
    ]
    constraint_satisfied = np.zeros(n, dtype=int)
    for _ in range(samples):
        point = {}
        for i, sym in enumerate(symbols):
            typ = variable_type[i].lower()
            low, high = variable_bounds[i]
            if typ == "integer":
                value = random.randint(int(low), int(high))
            elif typ in ("real", "continuous"):
                value = random.uniform(low, high)
            elif typ in ("boolean", "logical"):
                value = random.choice([False, True])
            else:
                raise ValueError(
                    f"Unknown type {variable_type[i]} for variable {variable_name[i]}"
                )
            point[sym] = value

        # Evaluate all constraints at this point
        try:
            vals = [point[s] for s in symbols]
            cons_satisfaction = [
                bool(np.asarray(f(*vals)).all()) for f in funcs
            ]
            if all(cons_satisfaction):
                # Found a feasible point
                readable = {str(k): round(v, 3) for k, v in point.items()}
                return f"Feasible solution found: {readable}"
            else:
                constraint_satisfied += np.array(cons_satisfaction)
        except Exception as e:
            return f"Error evaluating constraint at point {point}: {e}"

    rates = constraint_satisfied / samples  # fraction satisfied per constraint
    order = np.argsort(rates)  # lowest (most violated) first

    lines = []
    for rank, idx in enumerate(order, start=1):
        expr_text = constraints[
            idx
        ]  # use the original string; easier to read than str(sympy_expr)
        sat = constraint_satisfied[idx]
        lines.append(
            f"[C{idx + 1}] {expr_text} — satisfied {sat:,}/{samples:,} ({sat / samples:.1%}), "
            f"violated {1 - sat / samples:.1%}"
        )

    return (
        f"No feasible solution found after {samples:,} samples. Most violated constraints (low→high satisfaction):\n "
        + "\n  ".join(lines)
    )

`feasibility_tools` ¶

Unified feasibility checker with heuristic pre-check and exact auto-routing.

Backends (imported lazily and used only if available): - PySMT (cvc5/msat/yices/z3) for SMT-style logic, disjunctions, and nonlinear constructs. - OR-Tools CP-SAT for strictly linear integer/boolean instances with integer coefficients. - OR-Tools CBC (pywraplp) for linear MILP/LP (mixed real + integer, or pure LP). - SciPy HiGHS (linprog) for pure continuous LP feasibility.

Install any subset you need

pip install pysmt && pysmt-install --cvc5 # or --z3/--msat/--yices pip install ortools pip install scipy pip install numpy

This file exposes a single LangChain tool: feasibility_check_auto.

`feasibility_check_auto(constraints, variable_name, variable_type, variable_bounds, prefer_smt_solver='cvc5', heuristic_enabled=True, heuristic_first=True, heuristic_samples=2000, heuristic_seed=None, heuristic_unbounded_radius_real=1000.0, heuristic_unbounded_radius_int=10 ** 6, numeric_tolerance=1e-08)` ¶

Unified feasibility checker with heuristic pre-check and exact auto-routing.

Performs an optional randomized feasibility search. If no witness is found (or the heuristic is disabled), the function auto-routes to an exact backend based on the detected problem structure (PySMT for SMT/logic/nonlinear, OR-Tools CP-SAT for linear integer/boolean, OR-Tools CBC for MILP/LP, or SciPy HiGHS for pure LP).

Parameters:

Name	Type	Description	Default
`constraints`	`Annotated[list[str], "Constraint strings like 'x0 + 2*x1 <= 5' or '(x0<=3) \| (x1>=2)'"]`	Constraint strings such as "x0 + 2*x1 <= 5" or "(x0<=3) \| (x1>=2)".	required
`variable_name`	`Annotated[list[str], ['x0', 'x1', ...]]`	Variable names, e.g., ["x0", "x1"].	required
`variable_type`	`Annotated[list[str], ['real' \| 'integer' \| 'boolean', ...]]`	Variable types aligned with `variable_name`. Each must be one of "real", "integer", or "boolean".	required
`variable_bounds`	`Annotated[list[list[Optional[float]]], '[(low, high), ...] (use None for unbounded)']`	Per-variable [low, high] bounds aligned with `variable_name`. Use None to denote an unbounded side.	required
`prefer_smt_solver`	`Annotated[str, "SMT backend if needed: 'cvc5'\|'msat'\|'yices'\|'z3'"]`	SMT backend name used by PySMT ("cvc5", "msat", "yices", or "z3").	`'cvc5'`
`heuristic_enabled`	`Annotated[bool, 'Run a fast randomized search first?']`	Whether to run the heuristic sampler.	`True`
`heuristic_first`	`Annotated[bool, 'Try heuristic before exact routing']`	If True, run the heuristic before exact routing; if False, run it after.	`True`
`heuristic_samples`	`Annotated[int, 'Samples for heuristic search']`	Number of heuristic samples.	`2000`
`heuristic_seed`	`Annotated[Optional[int], 'Seed for reproducibility']`	Random seed for reproducibility.	`None`
`heuristic_unbounded_radius_real`	`Annotated[float, 'Sampling range for unbounded real vars']`	Sampling radius for unbounded real variables.	`1000.0`
`heuristic_unbounded_radius_int`	`Annotated[int, 'Sampling range for unbounded integer vars']`	Sampling radius for unbounded integer variables.	`10 ** 6`
`numeric_tolerance`	`Annotated[float, 'Tolerance for relational checks (Eq/Lt/Le/etc.)']`	Tolerance used in relational checks (e.g., Eq, Lt, Le).	`1e-08`

Returns:

Type	Description
`str`	A message indicating the chosen backend and the feasibility result. On success,
`str`	includes an example model (assignment). On infeasibility, includes a short
`str`	diagnostic or solver status.

Raises:

Type	Description
`ValueError`	If constraints cannot be parsed or an unsupported variable type is provided.

Source code in src/ursa/tools/feasibility_tools.py

@tool(parse_docstring=True)
def feasibility_check_auto(
    constraints: Annotated[
        list[str],
        "Constraint strings like 'x0 + 2*x1 <= 5' or '(x0<=3) | (x1>=2)'",
    ],
    variable_name: Annotated[list[str], "['x0','x1',...]"],
    variable_type: Annotated[list[str], "['real'|'integer'|'boolean', ...]"],
    variable_bounds: Annotated[
        list[list[Optional[float]]],
        "[(low, high), ...] (use None for unbounded)",
    ],
    prefer_smt_solver: Annotated[
        str, "SMT backend if needed: 'cvc5'|'msat'|'yices'|'z3'"
    ] = "cvc5",
    heuristic_enabled: Annotated[
        bool, "Run a fast randomized search first?"
    ] = True,
    heuristic_first: Annotated[
        bool, "Try heuristic before exact routing"
    ] = True,
    heuristic_samples: Annotated[int, "Samples for heuristic search"] = 2000,
    heuristic_seed: Annotated[Optional[int], "Seed for reproducibility"] = None,
    heuristic_unbounded_radius_real: Annotated[
        float, "Sampling range for unbounded real vars"
    ] = 1e3,
    heuristic_unbounded_radius_int: Annotated[
        int, "Sampling range for unbounded integer vars"
    ] = 10**6,
    numeric_tolerance: Annotated[
        float, "Tolerance for relational checks (Eq/Lt/Le/etc.)"
    ] = 1e-8,
) -> str:
    """Unified feasibility checker with heuristic pre-check and exact auto-routing.

    Performs an optional randomized feasibility search. If no witness is found (or the
    heuristic is disabled), the function auto-routes to an exact backend based on the
    detected problem structure (PySMT for SMT/logic/nonlinear, OR-Tools CP-SAT for
    linear integer/boolean, OR-Tools CBC for MILP/LP, or SciPy HiGHS for pure LP).

    Args:
        constraints: Constraint strings such as "x0 + 2*x1 <= 5" or "(x0<=3) | (x1>=2)".
        variable_name: Variable names, e.g., ["x0", "x1"].
        variable_type: Variable types aligned with `variable_name`. Each must be one of
            "real", "integer", or "boolean".
        variable_bounds: Per-variable [low, high] bounds aligned with `variable_name`.
            Use None to denote an unbounded side.
        prefer_smt_solver: SMT backend name used by PySMT ("cvc5", "msat", "yices", or "z3").
        heuristic_enabled: Whether to run the heuristic sampler.
        heuristic_first: If True, run the heuristic before exact routing; if False, run it after.
        heuristic_samples: Number of heuristic samples.
        heuristic_seed: Random seed for reproducibility.
        heuristic_unbounded_radius_real: Sampling radius for unbounded real variables.
        heuristic_unbounded_radius_int: Sampling radius for unbounded integer variables.
        numeric_tolerance: Tolerance used in relational checks (e.g., Eq, Lt, Le).

    Returns:
        A message indicating the chosen backend and the feasibility result. On success,
        includes an example model (assignment). On infeasibility, includes a short
        diagnostic or solver status.

    Raises:
        ValueError: If constraints cannot be parsed or an unsupported variable type is provided.
    """
    # 1) Parse
    try:
        symbols, sympy_cons = _parse_constraints(constraints, variable_name)
    except Exception as e:
        return f"Parse error: {e}"

    # 2) Heuristic (optional)
    if heuristic_enabled and heuristic_first:
        try:
            h_model = _heuristic_feasible(
                sympy_cons,
                symbols,
                variable_name,
                variable_type,
                variable_bounds,
                samples=heuristic_samples,
                seed=heuristic_seed,
                tol=numeric_tolerance,
                unbounded_radius_real=heuristic_unbounded_radius_real,
                unbounded_radius_int=heuristic_unbounded_radius_int,
            )
            if h_model is not None:
                return f"[backend=heuristic] Feasible (sampled witness). Example solution: {h_model}"
        except Exception:
            # Ignore heuristic issues and continue to exact route
            pass

    # 3) Classify & route
    info = _classify(sympy_cons, symbols, variable_type)

    # SMT needed or nonlinear / non-conj
    if info["requires_smt"] or not info["all_linear"]:
        res = _solve_with_pysmt(
            sympy_cons,
            symbols,
            variable_name,
            variable_type,
            variable_bounds,
            solver_name=prefer_smt_solver,
        )
        # Optional heuristic after exact if requested
        if (
            heuristic_enabled
            and not heuristic_first
            and any(
                kw in res.lower()
                for kw in ("unknown", "not installed", "unsupported", "failed")
            )
        ):
            h_model = _heuristic_feasible(
                sympy_cons,
                symbols,
                variable_name,
                variable_type,
                variable_bounds,
                samples=heuristic_samples,
                seed=heuristic_seed,
                tol=numeric_tolerance,
                unbounded_radius_real=heuristic_unbounded_radius_real,
                unbounded_radius_int=heuristic_unbounded_radius_int,
            )
            if h_model is not None:
                return f"[backend=heuristic] Feasible (sampled witness). Example solution: {h_model}"
        return res

    # Linear-only path: collect atomic conjuncts
    conjuncts: list[sp.Expr] = []
    for c in sympy_cons:
        atoms, _ = _flatten_conjunction(c)
        conjuncts.extend(atoms)

    has_int, has_bool, has_real = (
        info["has_int"],
        info["has_bool"],
        info["has_real"],
    )

    # Pure LP (continuous only)
    if not has_int and not has_bool and has_real:
        res = _solve_with_highs_lp(
            conjuncts, symbols, variable_name, variable_bounds
        )
        if "not installed" in res.lower():
            res = _solve_with_cbc_milp(
                conjuncts,
                symbols,
                variable_name,
                variable_type,
                variable_bounds,
            )
        if (
            heuristic_enabled
            and not heuristic_first
            and any(kw in res.lower() for kw in ("failed", "unknown"))
        ):
            h_model = _heuristic_feasible(
                sympy_cons,
                symbols,
                variable_name,
                variable_type,
                variable_bounds,
                samples=heuristic_samples,
                seed=heuristic_seed,
                tol=numeric_tolerance,
                unbounded_radius_real=heuristic_unbounded_radius_real,
                unbounded_radius_int=heuristic_unbounded_radius_int,
            )
            if h_model is not None:
                return f"[backend=heuristic] Feasible (sampled witness). Example solution: {h_model}"
        return res

    # All integer/boolean → CP-SAT first (if integer coefficients), else CBC MILP
    if (has_int or has_bool) and not has_real:
        res = _solve_with_cpsat_integer_boolean(
            conjuncts, symbols, variable_name, variable_type, variable_bounds
        )
        if (
            any(
                kw in res
                for kw in (
                    "routing to MILP/LP",
                    "handles linear conjunctions only",
                )
            )
            or "not installed" in res.lower()
        ):
            res = _solve_with_cbc_milp(
                conjuncts,
                symbols,
                variable_name,
                variable_type,
                variable_bounds,
            )
        return res

    # Mixed reals + integers → CBC MILP
    res = _solve_with_cbc_milp(
        conjuncts, symbols, variable_name, variable_type, variable_bounds
    )

    # Optional heuristic after exact (if backend missing/failing)
    if (
        heuristic_enabled
        and not heuristic_first
        and any(
            kw in res.lower() for kw in ("not installed", "failed", "status:")
        )
    ):
        h_model = _heuristic_feasible(
            sympy_cons,
            symbols,
            variable_name,
            variable_type,
            variable_bounds,
            samples=heuristic_samples,
            seed=heuristic_seed,
            tol=numeric_tolerance,
            unbounded_radius_real=heuristic_unbounded_radius_real,
            unbounded_radius_int=heuristic_unbounded_radius_int,
        )
        if h_model is not None:
            return f"[backend=heuristic] Feasible (sampled witness). Example solution: {h_model}"

    return res

`fm_base_tool` ¶

`TorchModuleTool` ¶

Bases: BaseModel, Generic[Input, Output, ModelInput, ModelOutput]

A helper class for exposing a PyTorch model as an MCP tool for inference. Provides default methods for running the following pipeline:

Preprocess a sequence of Inputs into a ModelInput
Pass ModelInput through the PyTorch model getting ModelOutput
Postprocess ModelOutput into a suitable sequence of Outputs

Complex models (i.e. multi-GPU) may not be fully supported by this class.

Source code in src/ursa/tools/fm_base_tool.py

class TorchModuleTool(
    BaseModel, Generic[Input, Output, ModelInput, ModelOutput]
):
    """
    A helper class for exposing a PyTorch model as an MCP tool for inference.
    Provides default methods for running the following pipeline:

    1. Preprocess a sequence of `Inputs` into a `ModelInput`
    2. Pass `ModelInput` through the PyTorch model getting `ModelOutput`
    3. Postprocess `ModelOutput` into a suitable sequence of `Outputs`

    Complex models (i.e. multi-GPU) may not be fully supported by this class.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    fm: torch.nn.Module
    """ The underlying PyTorch model used for inference """

    name: str = None
    """ A short name for the foundation model """

    description: str
    """ What the foundation model does / how to use it """

    args_schema: type[Input]
    """ The input schema for the model """

    output_schema: type[Output]
    """ The output_schema for the model """

    batch_size: int = 1
    """ Inputs to the model will be batched into set of at most this size """

    device: torch.device = Field(default_factory=default_device)
    """ The accelerator on which the model is placed """

    def preprocess(self, input: Sequence[Input]) -> ModelInput:
        """
        Convert tool input into the form accepted by the model
        The input will be of type `list[args_schema]` with a length
        of `batch_size`

        Defaults to `torch.data.default_collate`
        """

        return default_collate(list(input))

    def _forward(self, model_inputs: ModelInput) -> ModelOutput:
        """Process a batch of observations with the model"""
        return self.fm(model_inputs).to("cpu")

    def postprocess(self, model_output: ModelOutput) -> Iterable[Output]:
        """Postprocess the model's raw output into a relevant tool output format"""
        yield from model_output

    @classmethod
    def from_pretrained(
        cls, pretrained_model_name_or_path: str, **kwargs
    ) -> "TorchModuleTool":
        """Instantiate tool from a pretrained checkpoint either on disk,
        or automatically downloaded from a repository (i.e. HuggingFace)
        """
        raise NotImplementedError()

    def model_post_init(self, __context) -> None:
        # Move the model to the indicated device
        self.fm = self.fm.to(self.device)

        # Default to the class name
        if self.name is None:
            self.name = self.__class__.__name__

    @final
    def batch(self, inputs: list[Input], **kwargs) -> list[Output]:
        return list(self.batch_as_completed(inputs, **kwargs))

    @final
    def batch_as_completed(
        self,
        inputs: list[Input],
        max_concurency: int | None = None,
    ) -> Iterable[Output]:
        n = max_concurency or self.batch_size
        for batch in batched(inputs, n=n):
            from torch import inference_mode

            with inference_mode():
                batch = self.preprocess(batch)
                y = self._forward(batch)
                yield from self.postprocess(y)

    @final
    def __call__(self, input: Input):
        with torch.inference_mode():
            batch = self.preprocess([input])
            y = self._forward(batch)
            return next(iter(self.postprocess(y)))

    @final
    def __to_fastmcp(self) -> FastMCPTool:
        field_definitions = {
            field: (field_info.annotation, field_info)
            for field, field_info in self.args_schema.model_fields.items()
        }
        arg_model = create_model(
            f"{self.name}arguments",
            **field_definitions,
            __base__=ArgModelBase,
        )
        fn_metadata = FuncMetadata(
            arg_model=arg_model,
            output_model=self.output_schema,
            output_schema=self.output_schema.model_json_schema(),
        )

        async def fn(**input) -> self.output_schema:
            x = self.args_schema(**input)
            return self(x)

        return FastMCPTool(
            fn=fn,
            name=self.name,
            description=self.description,
            parameters=self.args_schema.model_json_schema(),
            fn_metadata=fn_metadata,
            is_async=True,
        )

    @final
    def add_to_fastmcp(self, server: FastMCP) -> FastMCPTool:
        """Add `self` as a tool to `server`"""
        fasttool = self.__to_fastmcp()

        if fasttool.name not in server._tool_manager._tools:
            server._tool_manager._tools[fasttool.name] = fasttool

        elif server._tool_manager.warn_on_duplicate_tools:
            logging.warning(f"Tool already exists: {fasttool.name}")

        return fasttool

`args_schema` `instance-attribute` ¶

The input schema for the model

`batch_size = 1` `class-attribute` `instance-attribute` ¶

Inputs to the model will be batched into set of at most this size

`description` `instance-attribute` ¶

What the foundation model does / how to use it

`device = Field(default_factory=default_device)` `class-attribute` `instance-attribute` ¶

The accelerator on which the model is placed

`fm` `instance-attribute` ¶

The underlying PyTorch model used for inference

`name = None` `class-attribute` `instance-attribute` ¶

A short name for the foundation model

`output_schema` `instance-attribute` ¶

The output_schema for the model

`add_to_fastmcp(server)` ¶

Add self as a tool to server

Source code in src/ursa/tools/fm_base_tool.py

@final
def add_to_fastmcp(self, server: FastMCP) -> FastMCPTool:
    """Add `self` as a tool to `server`"""
    fasttool = self.__to_fastmcp()

    if fasttool.name not in server._tool_manager._tools:
        server._tool_manager._tools[fasttool.name] = fasttool

    elif server._tool_manager.warn_on_duplicate_tools:
        logging.warning(f"Tool already exists: {fasttool.name}")

    return fasttool

`from_pretrained(pretrained_model_name_or_path, **kwargs)` `classmethod` ¶

Instantiate tool from a pretrained checkpoint either on disk, or automatically downloaded from a repository (i.e. HuggingFace)

Source code in src/ursa/tools/fm_base_tool.py

@classmethod
def from_pretrained(
    cls, pretrained_model_name_or_path: str, **kwargs
) -> "TorchModuleTool":
    """Instantiate tool from a pretrained checkpoint either on disk,
    or automatically downloaded from a repository (i.e. HuggingFace)
    """
    raise NotImplementedError()

`postprocess(model_output)` ¶

Postprocess the model's raw output into a relevant tool output format

Source code in src/ursa/tools/fm_base_tool.py

def postprocess(self, model_output: ModelOutput) -> Iterable[Output]:
    """Postprocess the model's raw output into a relevant tool output format"""
    yield from model_output

`preprocess(input)` ¶

Convert tool input into the form accepted by the model The input will be of type list[args_schema] with a length of batch_size

Defaults to torch.data.default_collate

Source code in src/ursa/tools/fm_base_tool.py

def preprocess(self, input: Sequence[Input]) -> ModelInput:
    """
    Convert tool input into the form accepted by the model
    The input will be of type `list[args_schema]` with a length
    of `batch_size`

    Defaults to `torch.data.default_collate`
    """

    return default_collate(list(input))