Vercel AI SDK

Quick start

Write your rules in a .rules file:

Load, compile, gate:

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

const { nets } = await loadRules("./safety.rules");

const gate = createPetriflowGate(nets, {
  isToolResultError: (_name, result) =>
    typeof result === "object" && result !== null && (result as any).success === false,
});

// Define your tools — PetriFlow wraps each one's execute method
const session = gate.wrapTools({
  backup: tool({
    description: "Create a backup of a file",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => ({ backedUp: `${path}.bak` }),
  }),
  delete: tool({
    description: "Delete a file (requires backup first)",
    inputSchema: z.object({ path: z.string() }),
    execute: async ({ path }) => ({ deleted: path }),
  }),
});

const { text } = await generateText({
  model: openai("gpt-4o"),
  tools: session.tools,
  system: session.systemPrompt(),
  prompt: "Clean up old data files",
});

loadRules reads a .rules file and builds a verified Petri net for each rule. createPetriflowGate wraps the nets into a stateless gate. wrapTools takes your Vercel AI SDK tool() definitions and returns a GateSession with instrumented tools and prompt helpers. Each wrapTools call creates fresh state. session.systemPrompt() tells the model which tools are gated and why, so it can plan around the constraints instead of hitting them blind.

How it works

wrapTools replaces each tool's execute with a wrapper that runs three phases:

// session = gate.wrapTools() instruments each tool's execute method:

// 1. Before execute → handleToolCall (gate check)
//    Blocked? → throw ToolCallBlockedError, tool never runs

// 2. Execute → original tool.execute(input, options)

// 3. After execute → handleToolResult
//    Success → deferred transitions advance
//    Failure (thrown error or isToolResultError) → marking unchanged

Tools without an execute method (schema-only) pass through unchanged. wrapTools returns a GateSession containing the wrapped tools, the system prompt, and status helpers. The tools preserve the same type T you pass in, so type inference works everywhere.

For details on the rule DSL (sequences, blocks, limits, tool mapping), see the rules engine docs.

Using with rules

The full pipeline: write rules in the DSL, compile them, create a gate, wrap your tools. Three imports, four steps.

import { compile } from "@petriflow/rules";
import { createPetriflowGate, ToolCallBlockedError } from "@petriflow/vercel-ai";
import { generateText } from "ai";

const { nets } = compile(`
  require lint before deploy
  require human-approval before deploy
  limit deploy to 2 per session
`);

const gate = createPetriflowGate(nets, {
  isToolResultError: (_name, result) =>
    typeof result === "object" && result !== null && (result as any).success === false,
  confirm: async (title, message) => {
    // your UI or CLI prompt here
    return await askUser(message);
  },
});

// myTools = your tool definitions (see Quick Start above)
const session = gate.wrapTools(myTools);

const { text } = await generateText({
  model: openai("gpt-4o"),
  tools: session.tools,
  system: session.systemPrompt(),
  prompt: "Run lint, then deploy to production",
});

Each rule compiles to an independent Petri net. The gate checks all nets on every tool call. A tool fires only if every net allows it. Rules compose by intersection, the same way they do in @petriflow/rules directly. See how rules compose.

Manual confirmation

require human-approval before X rules compile to manual transitions. The runtime calls your confirm callback and waits for a response. If the human approves, the tool executes. If they reject, it's blocked. If no confirm callback is provided, manual transitions are always blocked.

import { createPetriflowGate } from "@petriflow/vercel-ai";
import { compile } from "@petriflow/rules";
import * as readline from "readline";

const { nets } = compile("require human-approval before deploy");

// CLI: prompt on stdin
const gate = createPetriflowGate(nets, {
  isToolResultError: /* your classifier */ () => false,
  confirm: async (title, message) => {
    const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
    return new Promise((resolve) => {
      rl.question(`${message} (y/n) `, (answer) => {
        rl.close();
        resolve(answer.toLowerCase() === "y");
      });
    });
  },
});

// React: show a dialog
const gate2 = createPetriflowGate(nets, {
  isToolResultError: /* your classifier */ () => false,
  confirm: async (title, message) => {
    return await showApprovalDialog({ title, message });
  },
});

The callback receives two arguments: a title (e.g. "Approve: deploy") and a message (e.g. "Allow 'deploy' via transition 'approve' in net 'approve-before-deploy'?"). Return true to approve, false to reject.

Shadow mode

Set mode: "shadow" to observe gating decisions without blocking anything. Tools always execute. The onDecision callback fires on every call with what would have happened in enforce mode.

const gate = createPetriflowGate(nets, {
  isToolResultError: /* your classifier */ () => false,
  mode: "shadow",
  onDecision: (event, decision) => {
    if (decision?.block) {
      console.warn(`[AUDIT] ${event.toolName} would be blocked: ${decision.reason}`);
    }
  },
});

// Tools execute normally. Nothing is blocked.
// onDecision fires on every call, so you see what *would* happen.
// myTools = your tool definitions (see Quick Start above)
const { tools } = gate.wrapTools(myTools);

Use this for auditing an existing agent. Deploy shadow mode first, review the logs, then switch to "enforce" when you're confident the rules match your intent.

Registry mode

Pass a ComposeConfig instead of an array to enable dynamic net management. Register all nets upfront, activate a subset, and add or remove nets at runtime.

import { createPetriflowGate } from "@petriflow/vercel-ai";
import { compile } from "@petriflow/rules";

const { nets: safetyNets } = compile("block rm");
const { nets: deployNets } = compile(`
  require lint before deploy
  require human-approval before deploy
`);

// Register all nets, activate only safety at startup
const gate = createPetriflowGate({
  registry: {
    safety: safetyNets[0],
    deploy: deployNets[0],
    approval: deployNets[1],
  },
  active: ["safety"],
}, {
  isToolResultError: /* your classifier */ () => false,
});

// myTools = your tool definitions (see Quick Start above)
const session = gate.wrapTools(myTools);

// Later: enable deploy policy at runtime
session.addNet("deploy");
session.addNet("approval");
// { ok: true, message: "Activated 'deploy'" }

// Disable when no longer needed
session.removeNet("approval");
// { ok: true, message: "Deactivated 'approval' (state preserved)" }

addNet and removeNet return { ok: boolean, message: string }. Net state is preserved on deactivation and restored on reactivation. Inactive nets don't participate in gating decisions.

Use this when your agent's permissions change mid-session: escalating privileges after authentication, enabling deploy rules only during a release window, or disabling rate limits for admin users.

Replay

The AI SDK is stateless — generateText and streamText take the full message history each call. There's no built-in session persistence. Pass messages to wrapTools and the gate derives its state from the same history.

// The AI SDK is stateless — each generateText call takes the full
// message history. Gate state is derived from that same history.

const gate = createPetriflowGate(nets, {
  isToolResultError: (_name, result) =>
    typeof result === "object" && result !== null && (result as any).success === false,
});

async function handleRequest(messages: ModelMessage[]) {
  // Pass messages to initialize gate state from conversation history
  const session = gate.wrapTools(myTools, { messages });

  return generateText({
    model: openai("gpt-4o"),
    tools: session.tools,
    system: session.systemPrompt(),
    messages,
  });
}

For long-lived connections (SSE, WebSocket) where you hold the session in memory, you can omit messages — the wrapped tools track state automatically during execution.

Low-level replay

Use manager.replay() directly when you already know which tools completed, or when you need to pass input for toolMapper resolution.

// Framework-agnostic: replay from known tool names
manager.replay(["lint", "test"]);

// With full entries (for toolMapper resolution)
manager.replay([
  { toolName: "bash", input: { command: "cp file backup" }, isError: false },
  { toolName: "bash", input: { command: "rm file" }, isError: false },
]);

// Failed tools are skipped — deferred transitions don't advance
manager.replay([
  { toolName: "test", isError: true },   // skipped
  { toolName: "test", isError: false },   // fires
]);

Replay is idempotent — if a transition has already fired or can't fire, it's skipped silently. Failed tool results (isError: true) are skipped so deferred transitions don't advance on failure.

Security note: replay treats the message history as authoritative. If messages are round-tripped through the client, a caller could forge tool results to unlock gated tools. Use server-persisted message history for replay — the same best practice the AI SDK recommends for any stateless setup.

Error classification

By default, deferred transitions fire whenever a tool returns without throwing. But some tools signal errors through return values (e.g. { success: false, error: "sandbox crashed" }) rather than exceptions. Without classification, these "soft failures" advance the net as if they succeeded — inflating token counts and unlocking downstream transitions that shouldn't be available.

Use isToolResultError to teach the gate which return values are failures. It applies in both live execution and replay:

const gate = createPetriflowGate(nets, {
  // Classify non-throwing error results as failures.
  // Applied in both live execution and replay.
  isToolResultError: (toolName, result) => {
    if (typeof result === "object" && result !== null) {
      return (result as any).success === false;
    }
    return false;
  },
});

The callback always receives the raw tool return value — during replay, SDK output wrappers ({ type: "json", value: ... }) are stripped automatically. Built-in detection for Vercel AI error types (error-text, error-json, execution-denied) runs first as a fallback — the callback is only consulted when the built-in check does not already classify the result as an error. These types track the AI SDK's ToolResultOutput union. If the SDK adds new error variants in a future version, your isToolResultError callback is the primary safety net.

Error handling

When a tool is blocked in enforce mode, wrapTools throws a ToolCallBlockedError. The original tool never executes. Block reasons are constraint-stating messages (e.g. "deploy requires a successful call to test first") that tell the model what to do next, rather than exposing internal Petri net state.

import { ToolCallBlockedError } from "@petriflow/vercel-ai";

try {
  const session = gate.wrapTools(myTools);
  const result = await generateText({
    model: openai("gpt-4o"),
    tools: session.tools,
    prompt: "Delete all logs",
  });
} catch (e) {
  if (e instanceof ToolCallBlockedError) {
    console.log(e.toolName);   // "delete"
    console.log(e.toolCallId); // "call_abc123"
    console.log(e.reason);     // "delete requires a successful call to backup first."
    console.log(e.message);    // "Tool 'delete' blocked: delete requires ..."
  }
}

ToolCallBlockedError extends Error. Use instanceof to distinguish it from tool execution errors. The Vercel AI SDK catches tool errors and feeds them back to the model, so the agent sees the block reason and can adjust its plan.

API reference

createPetriflowGate(nets, opts)

Creates a gate instance. Two overloads: pass an array of nets for static composition, or a ComposeConfig for registry mode with dynamic activation.

// Static: pass compiled nets directly
function createPetriflowGate(
  nets: SkillNet<string>[],
  opts: GateOptions,
): PetriflowGate;

// Registry: dynamic activation/deactivation
function createPetriflowGate(
  config: ComposeConfig,
  opts: GateOptions,
): PetriflowGate;

GateOptions

type GateOptions = {
  mode?: "enforce" | "shadow";  // default: "enforce"
  confirm?: (title: string, message: string) => Promise<boolean>;
  onDecision?: (event: GateToolCall, decision: GateDecision) => void;
  transformBlockReason?: (toolName: string, reason: string) => string;
  isToolResultError: (toolName: string, result: unknown) => boolean;
};

• mode: "enforce" throws on blocked calls. "shadow" logs but never blocks. Default: "enforce".
• confirm: callback for require human-approval rules. If omitted, manual transitions are blocked.
• onDecision: fires after every gating decision. Receives the tool call event and the decision ({ block: true, reason } or undefined).
• transformBlockReason: optional hook to transform block reasons before they reach the model. Receives the tool name and the default constraint-stating message. Return a custom string to override it.
• isToolResultError: classifies a tool result as an error. Applied in both live execution and replay. Always receives the raw tool return value (SDK output wrappers are stripped during replay). When this returns true, deferred transitions do not fire and the net marking stays unchanged. Built-in detection for Vercel AI error types (error-text, error-json, execution-denied) always runs first; this callback is only consulted when the built-in check passes.

PetriflowGate

type PetriflowGate = {
  wrapTools: <T extends Record<string, any>>(
    tools: T,
    opts?: { messages?: { role: string; content: unknown }[] },
  ) => GateSession<T>;
};

type GateSession<T> = {
  tools: T;
  systemPrompt: () => string;
  formatStatus: () => string;
  addNet: (name: string) => { ok: boolean; message: string };
  removeNet: (name: string) => { ok: boolean; message: string };
  manager: GateManager;
};

• wrapTools(tools, opts?): instruments each tool's execute with gate checks. Returns a GateSession containing the wrapped tools, prompt helpers, and net management methods. Each call creates fresh state. Pass { messages } to initialize gate state from conversation history.

GateSession

• tools: the wrapped tools, same type as the input. Use these in generateText / streamText.
• systemPrompt(): returns a formatted prompt describing active nets, gated tools, and current state. Pass this to the model so it knows the constraints.
• formatStatus(): returns the current marking of all nets. Useful for debugging.
• addNet(name): registry mode only. Activates a registered net.
• removeNet(name): registry mode only. Deactivates a net. State is preserved for later reactivation.
• manager: the underlying GateManager instance. Useful for advanced introspection or custom integrations.

ReplayEntry

type ReplayEntry = {
  toolName: string;
  input?: Record<string, unknown>;
  isError: boolean;
};

• toolName: the tool that was called
• input: optional tool input, used for toolMapper resolution. replayFromMessages populates this automatically from the message history.
• isError: if true, the entry is skipped (deferred transitions don't advance on failure)

ToolCallBlockedError

class ToolCallBlockedError extends Error {
  readonly toolName: string;
  readonly toolCallId: string;
  readonly reason: string;
}

• toolName: the tool that was blocked
• toolCallId: the specific invocation ID
• reason: constraint-stating explanation from the gate (e.g. "deploy requires a successful call to test first")
• message: formatted as "Tool '<name>' blocked: <reason>"

RuleMetadata

Structured metadata attached to each SkillNet by the rules compiler. Used internally to generate constraint-stating block reasons. Re-exported for consumers who need to inspect rule structure.

type RuleMetadata =
  | { kind: "sequence"; prerequisite: string; dependent: string }
  | { kind: "approval"; tool: string }
  | { kind: "block"; tool: string }
  | { kind: "limit"; tool: string; limit: number; scope: "session" | string };

• sequence: "deploy requires a successful call to test first."
• approval: "deploy requires human approval."
• block: "rm is blocked and cannot be called."
• limit: "search has reached its limit of 3 calls per session."

vercelAiToolApprovalNet

A bundled Petri net for a common pattern: read-only tools are free, write tools are gated.

import { vercelAiToolApprovalNet } from "@petriflow/vercel-ai";
// or: import { vercelAiToolApprovalNet } from "@petriflow/vercel-ai/nets/tool-approval";

const gate = createPetriflowGate([vercelAiToolApprovalNet], {
  isToolResultError: /* your classifier */ () => false,
});
// Free tools: readData, fetchData (always allowed)
// Gated tools: writeData, sendEmail (require "ready" state)

Available from the main export or from @petriflow/vercel-ai/nets/tool-approval.