System tools

Give an in-workflow agent shell access, file read/write, grep, and directory traversal — scoped to a workspace directory that the agent cannot escape.

system_tools() returns a list of @task functions that give an agent access to the host filesystem and shell. Every tool runs inside a workspace directory you supply; paths outside that directory are rejected.

Import and call

from flux.tasks.ai import agent, system_tools

tools = system_tools(workspace="/tmp/my-project", timeout=30)
assistant = await agent("...", model="...", tools=tools)

system_tools is the only public export from flux.tasks.ai.tools. It returns a flat list[task] with tools from all four groups: shell, files, search, and directory.

Parameters

def system_tools(
    workspace: str | Path,
    timeout: int = 30,
    blocklist: list[str] | None = None,
    max_output_chars: int = 100_000,
) -> list[task]: ...

workspace is resolved to an absolute path at call time. timeout is the Flux task timeout in seconds; it applies to the shell tool only (file and search tools have no timeout). max_output_chars caps each tool’s text output before it is returned to the LLM — output that exceeds the limit is truncated and the response includes a "truncated": true field with the original byte count. blocklist is a list of regex patterns appended to the shell blocklist (see below); pass an empty list to disable the defaults entirely.

Tool groups

shell

One tool: shell(command: str, stream: bool = False).

Runs command in a subprocess with cwd set to the workspace. Returns {"status": "ok", "exit_code": int, "stdout": str, "stderr": str}. If stream=True, stdout chunks are emitted as progress events during execution.

Security layer. Every command passes through two checks before a subprocess is started:

  1. Baseline checks (shell_security.py) — 12 patterns rejecting fork bombs, destructive disk commands (mkfs, dd if=/dev/zero), system control (shutdown, reboot, halt), protected file writes (.env, .ssh/*, credentials), path traversal, download-and-execute pipes (curl | bash), unicode control characters, IFS/null-byte injection, dangerous env var overrides (LD_PRELOAD, PATH=), privilege escalation (sudo, chmod 777), reverse shells (/dev/tcp/), and crypto mining tools.

  2. Blocklist patterns — the default blocklist in system_tools.py adds seven regex patterns: rm -rf /, mkfs, dd (any use), shutdown, reboot, : () { (fork bomb), and writes to /dev/sd*.

Either check returning an error string causes shell to return {"status": "error", "error": "<reason>"} without spawning a process.

files

Four tools:

Two tools:

directory

Two tools:

Workspace sandboxing

Every tool that takes a path argument calls resolve_path() before doing anything with it. That function resolves both the workspace and the requested path to absolute paths (following symlinks) then checks that the requested path is relative to the workspace root. If it is not — including cases where a symlink points outside the workspace — the tool returns an error.

def resolve_path(config: SystemToolsConfig, path: str) -> Path:
    workspace = config.workspace.resolve()
    if Path(path).is_absolute():
        resolved = Path(path).resolve()
    else:
        resolved = (workspace / path).resolve()
    if not resolved.is_relative_to(workspace):
        raise ValueError(f"path escapes workspace boundary: {path}")
    return resolved

Absolute paths are allowed if they resolve to somewhere inside the workspace. Relative paths are resolved relative to the workspace root. Anything that resolves outside — including ../ traversal and symlink escapes — is rejected.

Selecting a subset of tools

system_tools() returns a plain list. Filter it by func.__name__ to pass only the tools you want:

all_tools = system_tools(workspace=workspace)
read_only = [
    t for t in all_tools
    if t.func.__name__ in ("read_file", "find_files", "grep", "list_directory", "directory_tree", "file_info")
]

Filtering works well for agents that should inspect a codebase but not modify it.

Example: file and search tools scoped to a workspace

The following workflow creates a code-review agent that can read files and search for patterns but cannot run shell commands or write to disk.

import tempfile
from pathlib import Path
from flux import ExecutionContext, workflow
from flux.tasks.ai import agent, system_tools


@workflow
async def review_agent(ctx: ExecutionContext[dict]):
    data = ctx.input or {}
    instruction = data.get("instruction", "Review the code and report findings.")
    workspace = data.get("workspace", tempfile.mkdtemp(prefix="flux_review_"))

    all_tools = system_tools(workspace=workspace)
    tools = [
        t for t in all_tools
        if t.func.__name__ in (
            "read_file", "file_info",
            "find_files", "grep",
            "list_directory", "directory_tree",
        )
    ]

    reviewer = await agent(
        "You are a code reviewer. Read and search the codebase, then report "
        "bugs, anti-patterns, and missing tests. Do not modify any files.",
        model="ollama/qwen2.5-coder:14b",
        name="reviewer",
        tools=tools,
        max_tool_calls=30,
    )

    return {"report": await reviewer(instruction), "workspace": workspace}

Run it inline:

ws = Path(tempfile.mkdtemp())
(ws / "app.py").write_text("def divide(a, b):\n    return a / b\n")

ctx = review_agent.run({"instruction": "Find unhandled exceptions.", "workspace": str(ws)})
print(ctx.output["report"])

Agent harness YAML

When defining an agent in YAML for the agent harness, the tool resolver accepts named groups. These map directly to the four system_tools sub-groups:

tools:
  - system_tools:           # all four groups together
      workspace: .
      timeout: 60
      max_output_chars: 200000

  # or individual groups:
  - files:
      workspace: /data/project
  - search:
      workspace: /data/project

Bare strings (e.g. - shell) use default timeout and max_output_chars. Dict form lets you override them per group.