Build a File Safety Agent

What you'll build

A file management agent with 5 tools: listFiles, readFile, backup, delete, and rm. Two rules control the agent's behavior: backup must succeed before delete is allowed, and rm is permanently blocked. The other tools are free. No rules mention them, so they always work.

Prerequisites

bun add @petriflow/vercel-ai @petriflow/rules ai @ai-sdk/openai zod

You'll also need an API key for your model provider (e.g. OPENAI_API_KEY).

Write the rules

Create a file called safety.rules:

Two rules, two different mechanisms:

• require backup before delete: compiles to a Petri net where delete is only reachable after backup succeeds. The transition is deferred: the net only advances when the tool's execute completes without error. If backup fails, delete stays locked.
• block rm: compiles to a net where the rm transition is provably dead in all reachable states. No sequence of tool calls can ever unlock it.

Tools not mentioned in any rule (listFiles, readFile) are free tools. They pass through the gate without checks.

Define the tools

Define your tools using the Vercel AI SDK's tool() function. The execute functions delegate to a files service object that wraps node:fs. PetriFlow doesn't change how you write them.

import { tool } from "ai";
import { z } from "zod";
import { files } from "./tools";

const myTools = {
  listFiles: tool({
    description: "List files in a directory",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.list(path),
  }),
  readFile: tool({
    description: "Read a file's contents",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.read(path),
  }),
  backup: tool({
    description: "Create a backup of a file before modifying it",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.backup(path),
  }),
  delete: tool({
    description: "Delete a file (requires backup first)",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.remove(path),
  }),
  rm: tool({
    description: "Remove a file with force",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.forceRemove(path),
  }),
};

Each tool has a description (for the model), a parameters schema (validated by the SDK), and an execute function (your implementation). PetriFlow wraps execute with gate checks. The tool definitions themselves stay the same.

Wire it up

Load the rules, create a gate, wrap your tools, and pass everything to generateText. Three imports, four steps.

import { loadRules } from "@petriflow/rules";
import { createPetriflowGate } from "@petriflow/vercel-ai";
import { generateText, tool, stepCountIs } from "ai";
import { openai } from "@ai-sdk/openai";
import { z } from "zod";
import { files } from "./tools";

// 1. Load and verify rules
const { nets, verification } = await loadRules("./safety.rules");
console.log(verification);
// [
//   { name: "require-backup-before-delete", reachableStates: 3 },
//   { name: "block-rm",                     reachableStates: 2 },
// ]

// 2. Create the gate
const gate = createPetriflowGate(nets);

// 3. Define and wrap tools
const tools = gate.wrapTools({
  listFiles: tool({
    description: "List files in a directory",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.list(path),
  }),
  readFile: tool({
    description: "Read a file's contents",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.read(path),
  }),
  backup: tool({
    description: "Create a backup of a file before modifying it",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.backup(path),
  }),
  delete: tool({
    description: "Delete a file (requires backup first)",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.remove(path),
  }),
  rm: tool({
    description: "Remove a file with force",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => files.forceRemove(path),
  }),
});

// 4. Run the agent
const { text } = await generateText({
  model: openai("gpt-4o"),
  tools,
  system: gate.systemPrompt(),
  stopWhen: stepCountIs(10),
  prompt: "Clean up /tmp/project: list the files, delete temp.log, and rm old-backup.tar.gz",
});

loadRules reads the .rules file and compiles each rule into a verified Petri net. createPetriflowGate wraps them into a gate. wrapTools instruments every tool's execute method with gate checks. systemPrompt() tells the model which tools are gated and why.

See it in action

Here's what happens when the agent tries to clean up a directory:

Key observations:

• listFiles works immediately. It's a free tool, not mentioned in any rule.
• delete is blocked until backup succeeds. If backup threw an error, delete would stay locked.
• rm is permanently blocked. The model sees the block reason in the error and adjusts its plan.
• The backup/delete cycle is repeatable. After deleting, you need to backup again before the next delete.

Concepts

Next steps