#!/usr/bin/env python3 """The three tools the Part 24 agent may use, and the guard rails around them. Purpose: one place where every tool the agent can reach is defined, together with the checks that make each one safe to hand to a token predictor. A tool here is a schema the model reads, a function the loop calls, and a set of refusals that happen before the function runs: a path that cannot leave the workspace, an executable allow-list, a wall-clock timeout with POSIX resource limits, and a truncation cap on everything returned. minimal-agent.py imports this module and knows nothing about any individual tool. Platform: all (pure Python). The command sandbox uses POSIX resource limits, so on Windows run it inside WSL2; without them the command tool refuses to run at all. Minimum memory: 8 GB on the machine serving the model; this module needs almost none Assumes: Python 3.9 or later and no third-party packages. A workspace directory that the agent is allowed to read, which should contain nothing you would mind an agent reading. Optionally a document index built by Part 10's ingest.py, which is queried with ordinary SQL keyword matching here rather than with embeddings, so no embedding server is needed. What this does NOT do, and the page says so too: the command sandbox is not a security boundary. The child runs as your user, on your filesystem, with your network. The limits here bound accidents and cheap denial of service - an infinite loop, a runaway allocation, a fork bomb, a disk-filling write - and the allow-list bounds what can be started at all. They do not contain a program written to do harm. An agent that must run untrusted code belongs in a container or a dedicated user with no credentials, which is what Part 25 builds. Usage: imported by minimal-agent.py and by mcp-server.py: from toolbox import Toolbox, ToolError run directly to exercise every tool and every guard rail without a model: python3 toolbox.py --workspace ./agent-workspace --self-test """ from __future__ import annotations import argparse import json import os import re import shutil import sqlite3 import subprocess import sys from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple try: import resource # POSIX only; the command tool refuses to run without it except ImportError: # pragma: no cover - platform check, not logic resource = None # type: ignore[assignment] # Commands that read and do not write. Adding to this list is a decision about what an # injected instruction is allowed to make happen, so the page asks you to say out loud # what each addition permits before you make it. DEFAULT_ALLOWED_COMMANDS = ("ls", "cat", "head", "tail", "wc", "grep", "find", "file") FINISH_TOOL = "finish" TEXT_SUFFIXES = {".md", ".txt", ".py", ".sh", ".json", ".toml", ".yaml", ".yml", ".cfg", ".ini"} class ToolError(Exception): """A refusal the model should see and be able to correct. Raised for anything the caller got wrong: a path outside the workspace, a command that is not allowed, a missing argument. The loop turns it into an ordinary tool result rather than a crash, which is what lets the model try something else. The MCP specification makes the same distinction between a protocol error and a tool execution error carrying `isError: true`, and for the same reason. """ JSON_TYPES = { "string": str, "integer": int, "number": (int, float), "boolean": bool, "array": list, "object": dict, } def schema_errors(arguments: Any, schema: Dict[str, Any]) -> List[str]: """Every way `arguments` fails `schema`, as short human-readable strings. Not a full JSON Schema implementation: it covers object type, required, additionalProperties, per-property type, enum and numeric bounds, which is everything the tool schemas in this course use. The agent sends the errors back to the model as an ordinary tool result, so a wrong call costs one turn rather than the run. """ errors: List[str] = [] if not isinstance(arguments, dict): return ["arguments are not a JSON object"] properties = schema.get("properties", {}) or {} for name in schema.get("required", []) or []: if name not in arguments: errors.append("missing required parameter %s" % name) if schema.get("additionalProperties") is False: for name in arguments: if name not in properties: errors.append("invented parameter %s" % name) for name, value in arguments.items(): spec = properties.get(name) if not isinstance(spec, dict): continue wanted = spec.get("type") expected = JSON_TYPES.get(wanted) if isinstance(wanted, str) else None if expected is not None: if wanted == "integer" and isinstance(value, float) and value.is_integer(): value = int(value) if isinstance(value, bool) and wanted != "boolean": errors.append("%s is a boolean, expected %s" % (name, wanted)) elif not isinstance(value, expected): errors.append("%s is %s, expected %s" % (name, type(value).__name__, wanted)) if "enum" in spec and value not in spec["enum"]: errors.append("%s=%r is not one of %s" % (name, value, spec["enum"])) if isinstance(value, (int, float)) and not isinstance(value, bool): if "minimum" in spec and value < spec["minimum"]: errors.append("%s is below the minimum" % name) if "maximum" in spec and value > spec["maximum"]: errors.append("%s is above the maximum" % name) return errors class Toolbox: """Every tool the agent can call, plus the schemas that describe them.""" def __init__(self, workspace: Path, index: Optional[Path] = None, allowed_commands: Sequence[str] = DEFAULT_ALLOWED_COMMANDS, timeout_s: int = 10, cpu_seconds: int = 10, memory_mb: int = 512, max_output_chars: int = 4000, max_file_bytes: int = 200_000) -> None: self.workspace = Path(workspace).expanduser().resolve() if not self.workspace.is_dir(): raise ToolError("workspace %s does not exist or is not a directory" % self.workspace) self.index = Path(index).expanduser().resolve() if index else None self.allowed_commands = tuple(allowed_commands) self.timeout_s = timeout_s self.cpu_seconds = cpu_seconds self.memory_mb = memory_mb self.max_output_chars = max_output_chars self.max_file_bytes = max_file_bytes # ---------------------------------------------------------------- schemas def schemas(self) -> List[Dict[str, Any]]: """The tool list exactly as it goes into the chat-completions request. Serialised once and reused every turn, in a fixed order: anything that varies at the top of a prompt destroys the prefix cache from that point on, and the tool list sits very near the top. """ return [ { "type": "function", "function": { "name": "read_file", "description": ( "Read one text file from the workspace and return its contents. " "The path must be relative to the workspace root and must not " "contain '..'. Large files are truncated."), "parameters": { "type": "object", "properties": { "path": { "type": "string", "description": "Path relative to the workspace root, e.g. 'notes/backup.md'.", }, }, "required": ["path"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "run_command", "description": ( "Run one allowed read-only command inside the workspace and return " "its output. Allowed commands: " + ", ".join(self.allowed_commands) + ". No shell is used, so pipes, redirection and globs are not " "interpreted; pass arguments separately."), "parameters": { "type": "object", "properties": { "command": { "type": "string", "description": "The executable to run.", "enum": list(self.allowed_commands), }, "args": { "type": "array", "items": {"type": "string"}, "description": "Arguments, one per element. Omit for none.", }, }, "required": ["command"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "search_documents", "description": ( "Search the document collection for passages matching keywords and " "return the best matches with their source. Use this when the answer " "might be written down somewhere rather than known in general."), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search terms, as words rather than a question.", }, "limit": { "type": "integer", "description": "How many passages to return, 1 to 10. Defaults to 5.", "minimum": 1, "maximum": 10, }, }, "required": ["query"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": FINISH_TOOL, "description": ( "Call this when the task is complete, with the final answer. " "Calling it ends the run, so call it exactly once and only when " "you have the answer."), "parameters": { "type": "object", "properties": { "answer": { "type": "string", "description": "The final answer, in a few sentences.", }, }, "required": ["answer"], "additionalProperties": False, }, }, }, ] # ------------------------------------------------------------- dispatch def call(self, name: str, arguments: Dict[str, Any]) -> str: """Run one validated call and return its result as text. Raises ToolError for anything refused. The loop is responsible for having already checked the arguments against the schema; these functions check the things a schema cannot express, which is all of the safety. """ if name == "read_file": return self.read_file(str(arguments["path"])) if name == "run_command": return self.run_command(str(arguments["command"]), [str(a) for a in arguments.get("args", [])]) if name == "search_documents": return self.search_documents(str(arguments["query"]), int(arguments.get("limit", 5))) raise ToolError("no tool named %r" % name) # ------------------------------------------------------------ the tools def resolve_in_workspace(self, path: str) -> Path: """The one path check, used by every tool that touches the filesystem. Rejects absolute paths and '..' by hand for a message the model can act on, then resolves symlinks and checks containment, which catches the cases the textual check misses: a symlink inside the workspace pointing outside it. """ if not path or not path.strip(): raise ToolError("path is empty") if path.strip() in (".", "./"): raise ToolError("path must name a file, not the workspace root") if os.path.isabs(path): raise ToolError("path must be relative to the workspace root") if ".." in Path(path).parts: raise ToolError("path must not contain '..'") candidate = (self.workspace / path).resolve() if candidate != self.workspace and self.workspace not in candidate.parents: raise ToolError("path resolves outside the workspace") return candidate def read_file(self, path: str) -> str: target = self.resolve_in_workspace(path) if not target.is_file(): raise ToolError("%s is not a file in the workspace" % path) data = target.read_bytes()[: self.max_file_bytes] text = data.decode("utf-8", "replace") if target.stat().st_size > self.max_file_bytes: text += "\n[truncated at %d bytes]" % self.max_file_bytes return self._cap(text) def run_command(self, command: str, args: Optional[List[str]] = None) -> str: args = list(args or []) if command not in self.allowed_commands: raise ToolError("%r is not an allowed command; allowed: %s" % (command, ", ".join(self.allowed_commands))) if resource is None: raise ToolError("resource limits are unavailable on this platform; " "run the agent inside WSL2 on Windows") executable = shutil.which(command) if executable is None: raise ToolError("%s is allowed but not installed on this machine" % command) for arg in args: if not arg.startswith("-") and (os.path.isabs(arg) or ".." in Path(arg).parts): raise ToolError("argument %r points outside the workspace" % arg) try: done = subprocess.run( # noqa: S603 - no shell, allow-listed executable [executable] + args, cwd=str(self.workspace), env={"PATH": os.environ.get("PATH", ""), "HOME": str(self.workspace), "LANG": "C.UTF-8"}, capture_output=True, text=True, timeout=self.timeout_s, check=False, preexec_fn=self._apply_limits, # noqa: PLW1509 - single-threaded caller ) except subprocess.TimeoutExpired: raise ToolError("command timed out after %d s" % self.timeout_s) from None except OSError as exc: raise ToolError("could not start the command: %s" % exc) from None parts = [] if done.stdout: parts.append(done.stdout.rstrip()) if done.stderr: parts.append("[stderr]\n" + done.stderr.rstrip()) parts.append("[exit code %d]" % done.returncode) return self._cap("\n".join(parts)) def _apply_limits(self) -> None: # pragma: no cover - runs in the child process """Bound the child before it executes anything. CPU time stops an infinite loop, address space stops a runaway allocation, file size stops a disk-filling write, and a process cap stops a fork bomb. The Python documentation is explicit that these are Unix-only and platform-dependent, so each one is applied only if the constant exists. """ limits = [ ("RLIMIT_CPU", self.cpu_seconds), ("RLIMIT_AS", self.memory_mb * 1024 * 1024), ("RLIMIT_FSIZE", 8 * 1024 * 1024), ("RLIMIT_NPROC", 64), ] for name, value in limits: constant = getattr(resource, name, None) if constant is not None: try: resource.setrlimit(constant, (value, value)) except (ValueError, OSError): pass os.setsid() def search_documents(self, query: str, limit: int = 5) -> str: limit = max(1, min(10, int(limit))) terms = [t for t in re.findall(r"[\w-]+", query.lower()) if len(t) > 1][:5] if not terms: raise ToolError("query has no searchable words") if self.index and self.index.is_file(): rows = self._search_index(terms, limit) if rows: return self._cap("\n\n".join(rows)) return "no passages matched %r in %s" % (query, self.index.name) rows = self._search_files(terms, limit) if rows: return self._cap("\n\n".join(rows)) return "no passages matched %r under the workspace" % query def _search_index(self, terms: List[str], limit: int) -> List[str]: """Keyword search over a Part 10 index, with no embedding server involved. Part 10's ingest.py stores every chunk's text in an ordinary `chunks` table beside the vector table, so a keyword query needs nothing but SQLite. It is a worse search than the embedding path; it is also always available, which for an agent tool is worth more than a better ranking you cannot always reach. """ where = " and ".join(["lower(text) like ?"] * len(terms)) params = ["%" + t + "%" for t in terms] + [limit] out: List[str] = [] try: with sqlite3.connect("file:%s?mode=ro" % self.index, uri=True) as db: cursor = db.execute( "select source, heading, text from chunks where %s limit ?" % where, params) for source, heading, text in cursor.fetchall(): label = source if not heading else "%s > %s" % (source, heading) out.append("[%s]\n%s" % (label, text.strip()[:800])) except sqlite3.Error as exc: raise ToolError("could not read the index: %s" % exc) from None return out def _search_files(self, terms: List[str], limit: int) -> List[str]: """The fallback: a plain-text scan of the workspace, paragraph by paragraph.""" out: List[str] = [] for path in sorted(self.workspace.rglob("*")): if len(out) >= limit: break if not path.is_file() or path.suffix.lower() not in TEXT_SUFFIXES: continue if path.stat().st_size > self.max_file_bytes: continue try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: continue for paragraph in re.split(r"\n\s*\n", text): lowered = paragraph.lower() if all(term in lowered for term in terms): relative = path.relative_to(self.workspace) out.append("[%s]\n%s" % (relative, paragraph.strip()[:800])) break return out def _cap(self, text: str) -> str: """Every tool result is truncated. An unbounded observation is an unbounded prompt.""" if len(text) <= self.max_output_chars: return text return text[: self.max_output_chars] + "\n[output truncated at %d characters]" % self.max_output_chars # -------------------------------------------------------------------------------------- # Exercising the guard rails without a model # -------------------------------------------------------------------------------------- def build_self_tests(box: "Toolbox") -> List[Tuple[str, str, Dict[str, Any], str]]: """Every guard rail, as a case with the outcome it should produce. The first case needs a file that really exists, so it is discovered rather than hard-coded: a self-test that fails because of its own fixture teaches nothing. """ sample = next((p.relative_to(box.workspace).as_posix() for p in sorted(box.workspace.rglob("*")) if p.is_file() and p.suffix.lower() in TEXT_SUFFIXES), None) cases: List[Tuple[str, str, Dict[str, Any], str]] = [] if sample: cases.append(("read a file in the workspace", "read_file", {"path": sample}, "allowed")) cases += [ ("read a file above the workspace", "read_file", {"path": "../secrets.txt"}, "refused"), ("read an absolute path", "read_file", {"path": "/etc/hostname"}, "refused"), ("read a directory as a file", "read_file", {"path": "."}, "refused"), ("list the workspace", "run_command", {"command": "ls", "args": ["-la"]}, "allowed"), ("run a command that is not allowed", "run_command", {"command": "curl", "args": []}, "refused"), ("pass an argument outside the workspace", "run_command", {"command": "cat", "args": ["../secrets.txt"]}, "refused"), ("search the documents", "search_documents", {"query": "backup window", "limit": 3}, "allowed"), ("search with no usable words", "search_documents", {"query": "a"}, "refused"), ] return cases def self_test(box: Toolbox) -> int: """Run every guard rail and report. Returns the number of surprises.""" surprises = 0 for label, name, arguments, expectation in build_self_tests(box): try: result = box.call(name, arguments) outcome = "allowed" detail = result.strip().splitlines()[0][:90] if result.strip() else "(empty)" except ToolError as exc: outcome = "refused" detail = str(exc)[:90] mark = "ok " if outcome == expectation else "SURPRISE" if outcome != expectation: surprises += 1 print("%-8s %-38s %-8s %s" % (mark, label, outcome, detail)) return surprises def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--workspace", required=True, help="the directory the agent may read") parser.add_argument("--index", default=None, help="a Part 10 document index, optional") parser.add_argument("--allow-command", action="append", default=[], help="add one executable to the allow-list; repeatable") parser.add_argument("--self-test", action="store_true", help="exercise every guard rail") parser.add_argument("--schemas", action="store_true", help="print the tool schemas as JSON") args = parser.parse_args() box = Toolbox( workspace=Path(args.workspace), index=Path(args.index) if args.index else None, allowed_commands=tuple(DEFAULT_ALLOWED_COMMANDS) + tuple(args.allow_command), ) if args.schemas: print(json.dumps(box.schemas(), indent=2)) return if args.self_test: surprises = self_test(box) print("\n%d surprise(s). Every 'refused' line above is a guard rail doing its job." % surprises) sys.exit(1 if surprises else 0) print(__doc__) if __name__ == "__main__": main()