tools

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

read_file_tool

read_file(filename, state)

Reads in a file with a given filename into a string. Can read in PDF or files that are text/ASCII. Uses a PDF parser if the filename ends with .pdf (case-insensitive)

Parameters:

Name	Type	Description	Default
filename	str	string filename to read in	required

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

@tool
def read_file(filename: str, state: Annotated[dict, InjectedState]) -> str:
    """
    Reads in a file with a given filename into a string. Can read in PDF
    or files that are text/ASCII. Uses a PDF parser if the filename ends
    with .pdf (case-insensitive)

    Args:
        filename: string filename to read in
    """
    workspace_dir = state["workspace"]
    full_filename = os.path.join(workspace_dir, filename)

    print("[READING]: ", full_filename)
    try:
        if full_filename.lower().endswith(".pdf"):
            file_contents = read_pdf_text(full_filename)
        else:
            file_contents = read_text_file(full_filename)
    except Exception as e:
        print(f"[Error]: {e}")
        file_contents = f"[Error]: {e}"
    return file_contents

run_command_tool

run_command(query, state)

Execute a shell command in the workspace and return its combined output.

Runs the specified command using subprocess.run in the given workspace directory, captures stdout and stderr, enforces a maximum character budget, and formats both streams into a single string. KeyboardInterrupt during execution is caught and reported.

Parameters:

Name	Type	Description	Default
query	str	The shell command to execute.	required
state	Annotated[dict, InjectedState]	A dict with injected state; must include the 'workspace' path.	required

Returns:

Type	Description
str	A formatted string with "STDOUT:" followed by the truncated stdout and
str	"STDERR:" followed by the truncated stderr.

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

@tool
def run_command(query: str, state: Annotated[dict, InjectedState]) -> str:
    """Execute a shell command in the workspace and return its combined output.

    Runs the specified command using subprocess.run in the given workspace
    directory, captures stdout and stderr, enforces a maximum character budget,
    and formats both streams into a single string. KeyboardInterrupt during
    execution is caught and reported.

    Args:
        query: The shell command to execute.
        state: A dict with injected state; must include the 'workspace' path.

    Returns:
        A formatted string with "STDOUT:" followed by the truncated stdout and
        "STDERR:" followed by the truncated stderr.
    """
    workspace_dir = state["workspace"]

    print("RUNNING: ", query)
    try:
        result = subprocess.run(
            query,
            text=True,
            shell=True,
            timeout=60000,
            capture_output=True,
            cwd=workspace_dir,
        )
        stdout, stderr = result.stdout, result.stderr
    except KeyboardInterrupt:
        print("Keyboard Interrupt of command: ", query)
        stdout, stderr = "", "KeyboardInterrupt:"

    # Fit BOTH streams under a single overall cap
    stdout_fit, stderr_fit = _fit_streams_to_budget(
        stdout or "", stderr or "", MAX_TOOL_MSG_CHARS
    )

    print("STDOUT: ", stdout_fit)
    print("STDERR: ", stderr_fit)

    return f"STDOUT:\n{stdout_fit}\nSTDERR:\n{stderr_fit}"

search_tools

run_arxiv_search(prompt, query, runtime, max_results=3)

Search ArXiv for the first 'max_results' papers and summarize them in the context of the user prompt

Parameters:

Name	Type	Description	Default
prompt	str	string describing the information the agent is interested in from arxiv papers	required
query	str	1 and 8 word search query for the Arxiv search API to find papers relevant to the prompt	required
max_results	int	integer number of papers to return (defaults 3). Request fewer if searching for something very specific or a larger number if broadly searching for information. Do not exceeed 10.	3

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

@tool
def run_arxiv_search(
    prompt: str, query: str, runtime: ToolRuntime, max_results: int = 3
):
    """
    Search ArXiv for the first 'max_results' papers and summarize them in the context
    of the user prompt

    Arguments:
        prompt:
            string describing the information the agent is interested in from arxiv papers
        query:
            1 and 8 word search query for the Arxiv search API to find papers relevant to the prompt
        max_results:
            integer number of papers to return (defaults 3). Request fewer if searching for something
            very specific or a larger number if broadly searching for information. Do not exceeed 10.
    """
    try:
        agent = ArxivAgent(
            llm=runtime.state.get("model"),
            summarize=True,
            process_images=False,
            max_results=max_results,
            # rag_embedding=self.embedding,
            database_path=Path("./arxiv_downloaded"),
            summaries_path=Path("./arxiv_summaries"),
            download=True,
        )
        console.print(f"[bold cyan]Searching ArXiv for: [default]{query}")
        assert isinstance(query, str)

        arxiv_result = agent.invoke(
            arxiv_search_query=query,
            context=prompt,
        )
        console.print(
            Panel(
                f"{arxiv_result}",
                title=f"[bold cyan on black]ArXiv summary for {query}",
                border_style="cyan on black",
                style="cyan on black",
            )
        )
        return f"[ArXiv Agent Output]:\n {arxiv_result}"
    except Exception as e:
        return f"Unexpected error while running ArxivAgent: {e}"

run_osti_search(prompt, query, runtime, max_results=3)

Search OSTI.gov for the first 'max_results' papers and summarize them in the context of the user prompt

Parameters:

Name	Type	Description	Default
prompt	str	string describing the information the agent is interested in from arxiv papers	required
query	str	1 and 8 word search query for the OSTI.gov search API to find papers relevant to the prompt	required
max_results	int	integer number of papers to return (defaults 3). Request fewer if searching for something very specific or a larger number if broadly searching for information. Do not exceeed 10.	3

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

@tool
def run_osti_search(
    prompt: str,
    query: str,
    runtime: ToolRuntime,
    max_results: int = 3,
):
    """
    Search OSTI.gov for the first 'max_results' papers and summarize them in the context
    of the user prompt

    Arguments:
        prompt:
            string describing the information the agent is interested in from arxiv papers
        query:
            1 and 8 word search query for the OSTI.gov search API to find papers relevant to the prompt
        max_results:
            integer number of papers to return (defaults 3). Request fewer if searching for something
            very specific or a larger number if broadly searching for information. Do not exceeed 10.
    """
    max_results
    try:
        agent = OSTIAgent(
            llm=runtime.state.get("model"),
            summarize=True,
            process_images=False,
            max_results=max_results,
            # rag_embedding=self.embedding,
            database_path=Path("./osti_downloaded_papers"),
            summaries_path=Path("./osti_generated_summaries"),
            vectorstore_path=Path("./osti_vectorstores"),
            download=True,
        )
        console.print(f"[bold cyan]Searching OSTI.gov for: [default]{query}")
        assert isinstance(query, str)

        osti_result = agent.invoke(
            query=query,
            context=prompt,
        )
        console.print(
            Panel(
                f"[cyan on black]{osti_result}",
                title=f"[bold cyan on black]OSTI.gov summary for {query}",
                border_style="cyan on black",
                style="cyan on black",
            )
        )
        return f"[OSTI Agent Output]:\n {osti_result}"
    except Exception as e:
        return f"Unexpected error while running OSTIAgent: {e}"

run_web_search(prompt, query, runtime, max_results=3)

Search the internet for the first 'max_results' pages and summarize them in the context of the user prompt

Parameters:

Name	Type	Description	Default
prompt	str	string describing the information the agent is interested in from websites	required
query	str	1 and 8 word search query for the web search engines to find papers relevant to the prompt	required
max_results	int	integer number of pages to return (defaults 3). Request fewer if searching for something very specific or a larger number if broadly searching for information. Do not exceeed 10.	3

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

@tool
def run_web_search(
    prompt: str,
    query: str,
    runtime: ToolRuntime,
    max_results: int = 3,
):
    """
    Search the internet for the first 'max_results' pages and summarize them in the context
    of the user prompt

    Arguments:
        prompt:
            string describing the information the agent is interested in from websites
        query:
            1 and 8 word search query for the web search engines to find papers relevant to the prompt
        max_results:
            integer number of pages to return (defaults 3). Request fewer if searching for something
            very specific or a larger number if broadly searching for information. Do not exceeed 10.
    """
    try:
        agent = WebSearchAgent(
            llm=runtime.state.get("model"),
            summarize=True,
            process_images=False,
            max_results=max_results,
            # rag_embedding=self.embedding,
            database_path=Path("./web_downloads"),
            summaries_path=Path("./web_summaries"),
            download=True,
        )
        console.print(f"[bold cyan]Searching Web for: [default]{query}")
        assert isinstance(query, str)

        web_result = agent.invoke(
            query=query,
            context=prompt,
        )
        console.print(
            Panel(
                f"{web_result}",
                title=f"[bold cyan on black]Web summary for {query}",
                border_style="cyan on black",
                style="cyan on black",
            )
        )
        return f"[Web Search Agent Output]:\n {web_result}"
    except Exception as e:
        return f"Unexpected error while running WebSearchAgent: {e}"

write_code_tool

edit_code(old_code, new_code, filename, state)

Replace the first occurrence of old_code with new_code in filename.

Parameters:

Name	Type	Description	Default
old_code	str	Code fragment to search for.	required
new_code	str	Replacement fragment.	required
filename	str	Target file inside the workspace.	required

Returns:

Type	Description
str	Success / failure message.

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

@tool
def edit_code(
    old_code: str,
    new_code: str,
    filename: str,
    state: Annotated[dict, InjectedState],
) -> str:
    """Replace the **first** occurrence of *old_code* with *new_code* in *filename*.

    Args:
        old_code: Code fragment to search for.
        new_code: Replacement fragment.
        filename: Target file inside the workspace.

    Returns:
        Success / failure message.
    """
    workspace_dir = state["workspace"]
    console.print("[cyan]Editing file:[/cyan]", filename)

    code_file = os.path.join(workspace_dir, filename)
    try:
        content = read_text_file(code_file)
    except FileNotFoundError:
        console.print(
            "[bold bright_white on red] :heavy_multiplication_x: [/] "
            "[red]File not found:[/]",
            filename,
        )
        return f"Failed: {filename} not found."

    # Clean up markdown fences
    old_code_clean = _strip_fences(old_code)
    new_code_clean = _strip_fences(new_code)

    if old_code_clean not in content:
        console.print(
            "[yellow] ⚠️ 'old_code' not found in file'; no changes made.[/]"
        )
        return f"No changes made to {filename}: 'old_code' not found in file."

    updated = content.replace(old_code_clean, new_code_clean, 1)

    console.print(
        Panel(
            DiffRenderer(content, updated, filename),
            title="Diff Preview",
            border_style="cyan",
        )
    )

    try:
        with open(code_file, "w", encoding="utf-8") as f:
            f.write(updated)
    except Exception as exc:
        console.print(
            "[bold bright_white on red] :heavy_multiplication_x: [/] "
            "[red]Failed to write file:[/]",
            exc,
        )
        return f"Failed to edit {filename}."

    console.print(
        f"[bold bright_white on green] :heavy_check_mark: [/] "
        f"[green]File updated:[/] {code_file}"
    )
    file_list = state.get("code_files", [])
    if code_file not in file_list:
        file_list.append(filename)
    state["code_files"] = file_list

    return f"File {filename} updated successfully."

write_code(code, filename, tool_call_id, state)

Write source code to a file and update the agent’s workspace state.

Parameters:

Name	Type	Description	Default
code	str	The source code content to be written to disk.	required
filename	str	Name of the target file (including its extension).	required
tool_call_id	Annotated[str, InjectedToolCallId]	Identifier for this tool invocation.	required
state	Annotated[dict, InjectedState]	Agent state dict holding workspace path and file list.	required

Returns:

Name	Type	Description
Command	Command	Contains an updated state (including code_files) and
	Command	a ToolMessage acknowledging success or failure.

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

@tool
def write_code(
    code: str,
    filename: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
    state: Annotated[dict, InjectedState],
) -> Command:
    """Write source code to a file and update the agent’s workspace state.

    Args:
        code: The source code content to be written to disk.
        filename: Name of the target file (including its extension).
        tool_call_id: Identifier for this tool invocation.
        state: Agent state dict holding workspace path and file list.

    Returns:
        Command: Contains an updated state (including code_files) and
        a ToolMessage acknowledging success or failure.
    """
    # Determine the full path to the target file
    workspace_dir = state["workspace"]
    console.print("[cyan]Writing file:[/]", filename)

    # Clean up markdown fences on submitted code.
    code = _strip_fences(code)

    # Show syntax-highlighted preview before writing to file
    try:
        lexer_name = Syntax.guess_lexer(filename, code)
    except Exception:
        lexer_name = "text"

    console.print(
        Panel(
            Syntax(code, lexer_name, line_numbers=True),
            title="File Preview",
            border_style="cyan",
        )
    )

    # Write cleaned code to disk
    code_file = os.path.join(workspace_dir, filename)
    try:
        with open(code_file, "w", encoding="utf-8") as f:
            f.write(code)
    except Exception as exc:
        console.print(
            "[bold bright_white on red] :heavy_multiplication_x: [/] "
            "[red]Failed to write file:[/]",
            exc,
        )
        return f"Failed to write {filename}."

    console.print(
        f"[bold bright_white on green] :heavy_check_mark: [/] "
        f"[green]File written:[/] {code_file}"
    )

    # Append the file to the list in agent's state for later reference
    file_list = state.get("code_files", [])
    if filename not in file_list:
        file_list.append(filename)

    # Create a tool message to send back to acknowledge success.
    msg = ToolMessage(
        content=f"File {filename} written successfully.",
        tool_call_id=tool_call_id,
    )

    # Return updated code files list & the message
    return Command(
        update={
            "code_files": file_list,
            "messages": [msg],
        }
    )