Agents & tools

The seven core tools, the permission harness, sandboxing, and the agent loop.

How do I give an agent a set of file tools?

grapharc.tools ships seven tools: read_file, write_file, edit_file, list_dir, glob, grep, run_command. register_core_tools puts them in a ToolRegistry bound to one workspace directory, which must already exist. Registering a tool and authorising it are two separate decisions in two different objects: include/exclude is the coarse one (a tool never registered can never be called, however the policy is edited later) and the PermissionPolicy is the fine one. Excluding run_command from a toolset with no business shelling out is a stronger statement than denying it, because there is nothing left to deny. visible_tools() returns sorted order while CORE_TOOL_NAMES is build order; core_tools() always returns CORE_TOOL_NAMES order so two callers requesting the same set get byte-identical schema lists.

python
registry = ToolRegistry()
register_core_tools(registry, workspace, exclude=["run_command"])

policy = PermissionPolicy(rules=[PermissionRule(action=Decision.ALLOW, pattern="*")])
harness = Harness(registry, policy, executor=LocalExecutor(), workspace=str(workspace))

print([spec.name for spec in harness.visible_tools()])
print(harness.call("read_file", {"path": "calc.py"}))
console
('read_file', 'write_file', 'edit_file', 'list_dir', 'glob', 'grep', 'run_command')
['edit_file', 'glob', 'grep', 'list_dir', 'read_file', 'write_file']
def add(a, b):
    return a - b

How do I stop a tool reading outside its workspace?

You do not have to: every core tool routes every path argument through Workspace.resolve before doing anything else. Absolute paths, .. segments and symlinks are not three special cases with three rules โ€” the path is resolved first and then judged by one rule: the thing the OS would actually open has to be inside the root. That is why subdir/../calc.py is fine, while a blocklist that refuses it on spelling and lets a symlink walk past is the shape of every path-traversal CVE. Containment is compared component-wise, so a sibling directory named <workspace>-evil is outside, not inside.

python
def attempt(path: str) -> None:
    try:
        print(f"{path!r:<20} -> {read_file(path)!r}")
    except WorkspaceEscape as exc:
        print(f"{path!r:<20} -> {str(exc).replace(str(workspace), &#x27;<ws>')}")

attempt("calc.py")
attempt("subdir/../calc.py")  # `..` is legal โ€” only where it lands is judged
attempt("../../etc/passwd")
attempt("/etc/passwd")
attempt("innocent.txt")  # a symlink pointing out of the workspace
console
'calc.py'            -> 'def add(a, b):\n    return a - b\n'
'subdir/../calc.py'  -> 'def add(a, b):\n    return a - b\n'
'innocent.txt'       -> path 'innocent.txt' resolves to /etc/passwd, which is outside the workspace <ws>. Core tools only reach paths inside the workspace; use a path relative to its root.

How do I add a tool of my own?

A ToolSpec is a name, a description, a callable, and two flags; the schema the model sees is derived from the callable's signature, so annotate it. ToolSpec carries no parameter schema of its own โ€” annotations become JSON types, defaulted parameters drop out of required, and a **kwargs tool is declared open with additionalProperties. Anything unannotated or unrecognised becomes "string" rather than being dropped, because a parameter the model cannot see is a parameter it cannot fill. The description is the only thing telling the model when to reach for the tool and is not decoration โ€” compare the core run_command description, which spends several sentences on argv being a list, there being no shell, and grep being cheaper for reading files.

python
def find_owner(service: str, region: str = "us-east-1", page: int = 1) -> str:
    """Look up the on-call owner for a service."""
    return f"{service}@{region} p{page}"


spec = ToolSpec(
    name="find_owner",
    description=find_owner.__doc__ or "",
    fn=find_owner,
    needs_network=True,      # the sandbox blocks sockets unless a tool declares this
    timeout_seconds=10.0,    # a hang past this kills the tool's whole process group
)
print(json.dumps(tool_schema(spec), indent=2))

How do I run the agent loop without calling a real model?

grapharc.testing.ScriptedChatModel scripts text only, and BaseChatModel leaves bind_tools unimplemented, so it cannot drive a tool loop; the two missing pieces are ~15 lines, saved as cookbook_model.py next to the other snippets. ToolScriptedModel adds bind_tools (which returns self so the loop keeps talking to the same script) and a _generate that attaches tool_call_script[i] to scripted response i, where an empty entry is a plain text turn โ€” that is how the loop learns the task is done. With that, an AgentNode is a model, a harness and a loop: observe state -> model (visible tools bound) -> tool request -> harness.call: permission -> approval -> hooks -> executor -> result (or denial, or violation) back to the model -> repeat until a recorded StopReason. Nothing routes around the harness: the node never touches ToolSpec.fn. A refusal is data, not an exception โ€” denied tools, sandbox violations and crashing tools all come back as tool results and run() does not raise; only BudgetExceeded, the run's hard ceiling, is re-raised.

python
class ToolScriptedModel(ScriptedChatModel):
    tool_call_script: list[list[dict[str, Any]]] = Field(default_factory=list)
    bound: list[list[dict[str, Any]]] = Field(default_factory=list)

    def bind_tools(self, tools, **kwargs):
        self.bound.append(list(tools))
        return self  # the loop must keep talking to this object's script

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        index = min(self._cursor, max(len(self.responses) - 1, 0))
        result = super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
        calls = self.tool_call_script[index] if index < len(self.tool_call_script) else None
        if not calls:
            return result
        base = result.generations[0].message
        return ChatResult(generations=[ChatGeneration(message=AIMessage(
            content=base.content,
            usage_metadata=base.usage_metadata,
            tool_calls=[{"type": "tool_call", **call} for call in calls],
        ))])
python
model = ToolScriptedModel(
    responses=["let me look", "fixing it", "add() now adds."],
    tool_call_script=[
        [{"name": "read_file", "args": {"path": "calc.py"}, "id": "1"}],
        [{"name": "edit_file",
          "args": {"path": "calc.py", "old_string": "a - b", "new_string": "a + b"},
          "id": "2"}],
        [],  # no tool call: the loop reads this turn as the answer
    ],
)
agent = AgentNode(model, harness, max_iterations=6)
result = agent.run("calc.add subtracts. Fix it.")

How do I find out why the loop stopped?

AgentResult.termination_reason is always set โ€” the loop cannot fall out of the bottom โ€” and note says why this reason and not just which. StopReason members are target_met, max_iterations, no_progress, budget_exhausted, human_stopped, error. Only target_met fills output: a run stopped mid-work puts the model's last words in partial_output, so a downstream node reading answer without termination_reason is never handed a plausible-looking non-answer. Progress is measured in observations, not requests โ€” re-sending byte-identical arguments is how you poll a job or re-run a suite after an edit, so a repeat that returns something new counts as progress and only a repeat returning the same result counts as a stall. max_stalled_turns defaults to 2 and max_iterations to 8.

python
# 1. The model keeps re-reading the same file and learning nothing new.
stalled = build(max_iterations=10).run("read it forever")
# 2. The turn cap, with the stall guard raised out of the way.
capped = build(max_iterations=3, max_stalled_turns=99).run("read it forever")
# 3. The meter's ceiling, checked before every turn.
ctx = RunContext(run_id="demo", graph="cookbook", meter=BudgetMeter(Budget(max_tokens=200)))
broke = build(max_iterations=99, max_stalled_turns=99).run("read it forever", ctx)
console
no_progress | 2 consecutive turns produced no new tool result
  answer: ''  partial: 'still working'
max_iterations | iteration cap reached (3/3)
budget_exhausted | max_tokens reached (335/200)

How do I decide which tool an agent may call?

A PermissionPolicy is a list of rules over fnmatch patterns, evaluated by tier โ€” every deny rule, then every ask, then every allow โ€” not in the order you wrote them. Three consequences: rule order in the list is irrelevant (you can append a deny to an existing policy and it takes effect over everything already there); the default is deny, so a tool nobody thought about does not run; and you cannot punch a hole in a deny. Patterns match the tool name only, never its arguments.

python
show("a broad deny beats a narrow allow โ€” there is no exception list:",
     PermissionPolicy(rules=[
         PermissionRule(action=Decision.DENY, pattern="*_file"),
         PermissionRule(action=Decision.ALLOW, pattern="read_file"),
     ]), names)
console
a broad deny beats a narrow allow โ€” there is no exception list:
   read_file        deny
   write_file       deny
   run_command      deny
   deploy_to_prod   deny

default when nothing matches: deny

How do I make sure a denied tool is never even offered to the model?

You do not have to do anything: Harness.visible_tools() filters by policy and AgentNode binds exactly that set, so a denied tool's schema is never described to the model โ€” and if it asks anyway, the call is still refused. Two enforcement points, and you need both: filtering schemas is the cheap one (no tokens spent describing tools it may not use), and re-checking inside Harness.call is the one that actually holds, because schemas are advisory, a model can hallucinate a tool name, and a second caller may reach the harness without going through visible_tools(). refused_by is a recorded field โ€” "policy" for a permission denial, "sandbox" for a sandbox violation โ€” so an audit is a lookup, not a string match. A denial is not a failure of the run; the run in the snippet still ends target_met, because killing the run on a denial would teach every caller to widen their policy until nothing is denied.

python
registry = ToolRegistry()
register_core_tools(registry, workspace)  # all seven are registered
policy = PermissionPolicy(rules=[
    PermissionRule(action=Decision.DENY, pattern="run_command"),
    PermissionRule(action=Decision.DENY, pattern="write_file"),
    PermissionRule(action=Decision.ALLOW, pattern="*"),
])
harness = Harness(registry, policy, executor=LocalExecutor(), workspace=str(workspace))
print("offered   :", [s["function"]["name"] for s in tool_schemas(harness)])
console
registered: ['read_file', 'write_file', 'edit_file', 'list_dir', 'glob', 'grep', 'run_command']
offered   : ['edit_file', 'glob', 'grep', 'list_dir', 'read_file']
direct    : PermissionDenied: tool 'write_file' denied by policy

stopped   : target_met
  denied   refused_by=policy   PERMISSION_DENIED: tool 'write_file' denied by policy

How do I put a human in front of a tool?

Mark the tool ASK and give the harness an approval callback that receives the tool name and arguments and returns a bool. ASK with no approval= is DENY โ€” there is no "proceed unless someone objects" mode, and an absent channel means no approval. Unlike the policy, the callback does see the arguments, so argument-level rules live here: a size cap, a path allowlist, a check that the diff does not touch main. The refusal is a PermissionDenied, which AgentNode turns into an ordinary PERMISSION_DENIED tool result, so the model can revise and try again.

python
def approve_small_writes(tool: str, args: dict[str, Any]) -> bool:
    """The gate. Return True to let the call through, False to refuse it."""
    asked.append((tool, args))
    return len(args.get("content", "")) < 100

gated = Harness(registry, policy, executor=LocalExecutor(),
                workspace=str(workspace), approval=approve_small_writes)
console
approved  : overwrote calc.py (3 bytes)
refused   : tool 'write_file' requires approval and none was granted
no channel: tool 'write_file' requires approval and none was granted
allowed   : 'ok\n'
asked     : 2 times total

How do I block or rewrite a call deterministically?

Pre-hooks run after the permission check and before the executor; they can deny a call or rewrite its arguments, and post-hooks transform the result. Hooks are code: they fire every time, whatever the model was told in a system prompt, which is the difference between a rule and a request. A pre-hook is where argument-level policy belongs, since the permission layer only ever sees a tool name. Returning None from a pre-hook means "no opinion".

python
def no_secrets(tool: str, args: dict[str, Any]) -> HookDecision | None:
    """Deny any read of a dotfile. Returning None means 'no opinion'."""
    if tool == "read_file" and Path(args.get("path", "")).name.startswith("."):
        return HookDecision(action=HookAction.DENY, reason="dotfiles are off limits")
    return None


def bound_reads(tool: str, args: dict[str, Any]) -> HookDecision | None:
    """Rewrite, rather than refuse: cap every read at 50 lines."""
    if tool == "read_file" and not args.get("limit"):
        return HookDecision(action=HookAction.REWRITE, args={**args, "limit": 50})
    return None


def redact(tool: str, args: dict[str, Any], result: Any) -> Any:
    """A post-hook sees the result and may transform it."""
    return result.replace("sk-real-key", "sk-***") if isinstance(result, str) else result
console
denied  : tool 'read_file' blocked by hook: dotfiles are off limits
rewritten: 'def add(a, b):\n    return a - b\n\n[lines 1-2 of 2]'
redacted : '1 match(es) in . (2 file(s) searched)\n.env:1: OPENROUTER_API_KEY=sk-***'

What does the default executor actually confine?

If you do not pass executor=, Harness builds a SandboxedExecutor over workspace=: it forks a child per call, scrubs its environment, and installs a CPython audit hook. There are two grants, deliberately different sizes โ€” reads reach the workspace and the interpreter's runtime paths (stdlib, site-packages, /usr/lib) because a tool has to be able to import; mutations reach the workspace and nothing else. That split exists because a writable site-packages lets a tool drop a .pth file, which executes arbitrary Python on every later start of that interpreter โ€” an escape in another process, outliving the run. The rest follows from one rule: anything that leaves this interpreter leaves the hook โ€” subprocesses are refused (including _posixsubprocess.fork_exec, not just Popen), as are ctypes, sqlite extension loading, and importing a compiled extension from outside the runtime paths. Network is off unless the ToolSpec declared needs_network, and then it is the whole network.

python
harness = Harness(
    registry,
    PermissionPolicy(rules=[PermissionRule(action=Decision.ALLOW, pattern="*")]),
    workspace=str(workspace),  # no executor= -> SandboxedExecutor over this workspace
)
console
write_here         ok       'scratch'
read_stdlib        ok       'json'
read_etc           REFUSED  tool 'read_etc' touched '/etc/passwd' outside its workspace (open)
write_sitepackages REFUSED  tool 'write_sitepackages' tried to modify '<site-packages>/evil.pth' outside its workspace (open)
shell_out          REFUSED  tool 'shell_out' tried to spawn a process (subprocess.Popen)
phone_home         REFUSED  tool 'phone_home' attempted network access (socket.getaddrinfo) without declaring needs_network
go_native          REFUSED  tool 'go_native' tried to reach native code through ctypes (ctypes.dlsym)

What does the sandbox not confine?

This matters more than the previous recipe: SandboxedExecutor is an in-process audit-hook confinement, not a kernel boundary, and the holes are known, documented and still open. Metadata reads are not confined (CPython raises no audit event for os.stat, so size, mtime and existence anywhere on the host stay readable); raw file descriptors are not (os.read/os.write/os.dup/mmap carry no path, and fork hands the child every descriptor the parent had); the forked heap is not (os.environ is scrubbed to an allowlist, but a secret a parent module already holds in a global stays reachable through sys.modules). The practical reading: a strong guard against a confused tool and a decent one against a careless model โ€” against a tool you would not run on your laptop, it is not a boundary. Use a container.

python
run("open()", lambda: outside.read_text())        # confined: content reads are checked
run("os.stat", lambda: os.stat(outside).st_size)  # not confined: no audit event exists
run("raw fd", lambda: os.read(leaked_fd, 64))     # not confined: an fd carries no path
run("global", lambda: API_KEY)                    # not confined: fork copies the heap
run("environ", lambda: os.environ.get("MY_API_KEY"))  # confined: the env is scrubbed
console
open()         REFUSED  tool 'open()' touched '<tmp>/outside.txt' outside its workspace (open)
os.stat        17
raw fd         b'a secret on disk\n'
global         'sk-a-secret-this-module-already-holds'
environ        None

Why do my tools say "outside its workspace" when the path is right?

Because Harness(registry, policy) without workspace= builds a SandboxedExecutor over a fresh temp directory, which is not the directory your tools were built for โ€” two confinements, two roots, and the sandbox wins. The tools confine paths and the executor confines paths independently and on purpose: LocalExecutor confines nothing, and run_command leaves the interpreter the audit hook lives in, so neither layer can be the only one. The cost is that you have to point both at the same directory โ€” pass the same path to register_core_tools and to Harness(..., workspace=...), or pass an executor you built yourself.

python
harness = Harness(  # note the missing workspace= argument
    registry,
    PermissionPolicy(rules=[PermissionRule(action=Decision.ALLOW, pattern="*")]),
)
print("same directory?  :", harness.executor.workspace == str(workspace))
console
same directory?  : False
result           : tool 'read_file' touched '<tools-ws>/calc.py' outside its workspace (open)

Why does run_command fail under the default executor?

Because the audit hook refuses to spawn a process and run_command spawns one โ€” a design decision, not a bug, with no flag to turn it off. Note which refusal you get: not the spawn, but the /dev/null open that stdin=DEVNULL performs a moment earlier โ€” same outcome, one audit event sooner. So run_command needs LocalExecutor (no confinement at all) or ContainerExecutor (a real one); under LocalExecutor the child is an ordinary process with your privileges reaching the whole filesystem, and what limits it is the permission policy deciding whether it may run at all. What run_command does give you is discipline around the call: argv is a list and shell=False, always; a pipeline is spelled ["bash", "-lc", "..."], explicit and visible in the tool-call record; the deadline kills the whole process group; and the environment is an allowlist so your provider key is not handed to whatever was run.

python
try:  # the default executor โ€” SandboxedExecutor โ€” refuses every subprocess
    harness_with(None).call("run_command", {"argv": ["echo", "hi"]})
except SandboxViolation as exc:
    print("sandbox:", exc)

print("local  :", harness_with(LocalExecutor()).call("run_command", {"argv": ["ls"]}))
console
sandbox: tool 'run_command' tried to modify '/dev/null' outside its workspace (open); the runtime paths are readable but never writable โ€” code left in them runs in later interpreters, where no hook is watching
local  : exit code 0 after 0.00s
stdout:
calc.py

How do I get a real boundary?

ContainerExecutor runs each call in a fresh throwaway container, so the confinement is the kernel's โ€” namespaces, cgroups, dropped capabilities โ€” rather than CPython's, behind the same run(spec, args) interface so callers never branch on which executor they hold. The constraint you hit in the first five minutes: a container is a different filesystem and interpreter, so spec.fn cannot cross; what crosses is an import path module:qualname derived from spec.fn, resolved by importing that module inside the container, which means the tool's code must already be installed in the image. So it is not a drop-in replacement for SandboxedExecutor on the core toolset โ€” either the derived path resolves in your image or you name it yourself with entrypoints={"my_tool": "mypkg.tools:my_tool"}. What the container gets: one bind mount (the workspace, read-write, at /workspace, also the cwd), --network none unless the tool declared needs_network, all capabilities dropped, no-new-privileges, a non-root uid, a read-only rootfs, and memory and pid limits โ€” plus the image's filesystem, which is why the image is part of the security decision.

python
executor = ContainerExecutor(str(workspace), image="python:3.12-slim")
version = ToolSpec(name="version", description="", fn=platform.python_version)
listing = ToolSpec(name="listing", description="", fn=os.listdir)
print("entrypoints      :", executor.entrypoint_for(version), executor.entrypoint_for(listing))
print("python in image  :", executor.run(version, {}))
print("workspace inside :", executor.run(listing, {"path": "/workspace"}))
print("host -> container:", executor.container_path(workspace / "calc.py"))
console
runtime reachable: True
entrypoints      : platform:python_version posix:listdir
python here      : 3.14.6
python in image  : 3.12.13
workspace inside : ['calc.py']
host -> container: /workspace/calc.py
local function   : tool 'my_tool' is defined in '__main__', which names a different file inside the container than it does here

How do I put an agent inside a graph?

AgentNode is callable as a node, and agent.writes reports exactly the state fields it writes, so the graph's write-permission check stays declarative. The state field names are configurable (task_field, output_field, reason_field, record_field) and writes is derived from whichever you chose, so the two cannot drift apart; for a prompt richer than one state field pass prompt_fn=lambda state: ..., and observe() raises AgentConfigError if it finds neither. An agent turn is metered like any other node: one iteration charged per model call against the RunContext's meter, and the message is named when tokens are charged so the runtime's usage callback recognises a re-report rather than double-counting. Set record_field to land the whole AgentResult โ€” every tool call, every refusal โ€” in state.

python
agent = AgentNode(model, harness)  # reads state.task; writes answer + termination_reason
print("agent.writes:", sorted(agent.writes))


def judge(state: State) -> dict:
    """A downstream node reads the reason, not just the answer."""
    return {"verdict": "reviewed" if state.termination_reason == "target_met" else "incomplete"}


g = GraphARC(State, name="review")
g.add_node("agent", agent, writes=agent.writes)
g.add_node("judge", judge, writes={"verdict"})
g.add_edge(START, "agent")
g.add_edge("agent", "judge")
g.add_edge("judge", END)

out = g.compile().invoke({"task": "what is wrong with calc.py?"})
console
agent.writes: ['answer', 'termination_reason']
answer      : add() subtracts where it should add.
reason      : target_met
verdict     : reviewed

How do I see all of this working end to end?

grapharc/examples/agent_fixit.py is the worked example: a project whose add() subtracts, a test suite that catches it, and an agent that must find the bug, fix it, and prove the fix by running the tests. delete_file is registered and denied, so the shortest path to a green suite is closed off in code. The point of the whole example is the last line of output: the agent said the suite was green, and run_pytest runs afterwards, outside the loop and says so independently โ€” an agent's claim about its own work is not evidence. build_harness uses LocalExecutor and says why in its docstring: run_tests shells out to pytest, which SandboxedExecutor refuses by design, so the example leans on the permission layer instead.

python
workspace = make_workspace()          # calc.py with `a - b`, and a test that wants `a + b`
harness = build_harness(workspace)    # read/write/run_tests, and a denied delete_file
result = AgentNode(model, harness, system_prompt=SYSTEM_PROMPT, max_iterations=12).run(TASK)
print("independent test run:", "PASS" if run_pytest(workspace).returncode == 0 else "FAIL")
bash
export OPENROUTER_API_KEY=sk-or-...
uv run python -m grapharc.examples.agent_fixit --model openrouter/anthropic/claude-haiku-4.5

How do I run an agent from the shell?

grapharc agent takes the task from you, the tools from grapharc.tools, and the permissions from flags: --allow, --deny and --ask are repeatable globs; --executor picks sandbox (default) or local; --max-turns, --max-tokens and --max-seconds bound the run; --json prints one document instead of prose. The mock/ backend resolves to ScriptedChatModel, which has no bind_tools, making it a way to see the wiring without spending anything โ€” tools are loaded, the policy applied, and the command then refuses to pretend a text-only model can drive a tool loop. The JSON payload answers the three questions a run is graded on: what it was allowed to do (policy, tools_visible), what it did (tool_calls), and why it stopped (termination_reason, note โ€” or error). A trace lands at <workspace>/trace.jsonl either way; grapharc trace, grapharc metrics and grapharc viz read it.

bash
grapharc agent --model mock/none --workspace /tmp/grapharc-cookbook-ws --run-id demo \
  --deny 'run_command' --deny 'write_file' --ask 'edit_file' --json "list the files"
console
{
  "ok": false,
  "command": "agent",
  "error": "model ScriptedChatModel does not implement bind_tools, so it cannot drive a tool loop (5 tools are visible to it); use a tool-calling backend such as openrouter/*, openai/* or ollama/*",
  "executor": "sandbox",
  "tools_from": "grapharc.tools.register_core_tools",
  "policy": {"allow": ["*"], "ask": ["edit_file"], "deny": ["run_command", "write_file"]},
  "tools_visible": ["edit_file", "glob", "grep", "list_dir", "read_file"]
}