What you'll build

A deployment agent with 5 tools: lint, test, deploy, checkStatus, and rollback. Four rules enforce a full CI/CD pipeline: lint before test, test before deploy, human approval before deploy, and a maximum of 2 deploys per session. checkStatus and rollback are free tools.

Prerequisites

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

You'll also need an API key for your model provider. This tutorial assumes you've read Tutorial 1: File Safety.

Write the rules

Create a file called pipeline.rules:

# pipeline.rules require lint before test require test before deploy require human-approval before deploy limit deploy to 2 per session

Four rules, three different mechanisms:

  • require lint before test + require test before deploy: chained sequencing. Deploy is two steps away from being reachable. The agent must complete lint, then test, in order.
  • require human-approval before deploy: compiles to a manual transition. The gate calls your confirm callback and waits. If the human approves, the transition fires and deploy unlocks. If they reject (or no callback is provided), deploy stays blocked.
  • limit deploy to 2 per session: places 2 tokens in a budget pool. Each successful deploy consumes one. After 2 deploys, the budget is empty and deploy is permanently blocked for the rest of the session.

Each rule compiles to an independent Petri net. All nets must agree for a tool call to proceed. This is AND composition.

Define the tools

Define your tools with tool() from the Vercel AI SDK, delegating to a pipeline service object that runs real shell commands and manages deploy state. checkStatus and rollback are free tools. No rules mention them.

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

const myTools = {
  lint: tool({
    description: "Run the linter on the codebase",
    inputSchema: z.object({}),
    execute: async () => pipeline.lint(),
  }),
  test: tool({
    description: "Run the test suite",
    inputSchema: z.object({}),
    execute: async () => pipeline.test(),
  }),
  deploy: tool({
    description: "Deploy to an environment",
    inputSchema: z.object({
      environment: z.enum(["production", "staging"]),
    }),
    execute: async ({ environment }) => pipeline.deploy(environment),
  }),
  checkStatus: tool({
    description: "Check the current deployment status",
    inputSchema: z.object({
      environment: z.enum(["production", "staging"]),
    }),
    execute: async ({ environment }) => pipeline.checkStatus(environment),
  }),
  rollback: tool({
    description: "Rollback to the previous deployment",
    inputSchema: z.object({
      environment: z.enum(["production", "staging"]),
    }),
    execute: async ({ environment }) => pipeline.rollback(environment),
  }),
};

Wire it up

The key difference from Tutorial 1 is the confirm callback. This is called whenever a require human-approval rule triggers.

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 { pipeline } from "./tools";

// 1. Load and verify rules
const { nets, verification } = await loadRules("./pipeline.rules");

// 2. Create the gate with a confirm callback for human-approval rules
const gate = createPetriflowGate(nets, {
  confirm: async (title, message) => {
    // In production: show a dialog, send a Slack message, etc.
    // Return true to approve, false to reject.
    console.log(`APPROVAL: ${message}`);
    return true;
  },
});

// 3. Define and wrap tools
const tools = gate.wrapTools({
  lint: tool({
    description: "Run the linter on the codebase",
    inputSchema: z.object({}),
    execute: async () => pipeline.lint(),
  }),
  test: tool({
    description: "Run the test suite",
    inputSchema: z.object({}),
    execute: async () => pipeline.test(),
  }),
  deploy: tool({
    description: "Deploy to an environment",
    inputSchema: z.object({
      environment: z.enum(["production", "staging"]),
    }),
    execute: async ({ environment }) => pipeline.deploy(environment),
  }),
  checkStatus: tool({
    description: "Check the current deployment status",
    inputSchema: z.object({
      environment: z.enum(["production", "staging"]),
    }),
    execute: async ({ environment }) => pipeline.checkStatus(environment),
  }),
  rollback: tool({
    description: "Rollback to the previous deployment",
    inputSchema: z.object({
      environment: z.enum(["production", "staging"]),
    }),
    execute: async ({ environment }) => pipeline.rollback(environment),
  }),
});

// 4. Run the agent
const { text } = await generateText({
  model: openai("gpt-4o"),
  tools,
  system: gate.systemPrompt(),
  stopWhen: stepCountIs(15),
  prompt: "Deploy the latest version to production, then deploy to staging.",
});

The confirm callback receives 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. If no callback is provided, manual transitions are always blocked.

See it in action

The agent tries to deploy to production, then to staging:

// deploy to production, then staging
Agent calls deploy({ environment: "production" })
BLOCKED lint hasn't run, test hasn't run, human hasn't approved
Agent calls lint()
OK linter passes, test is now unlocked
Agent calls test()
OK tests pass, deploy is one gate away
PetriFlow calls confirm: "Allow 'deploy'?"
Human approves → manual gate opens
Agent calls deploy({ environment: "production" })
ALLOWED all gates satisfied. Budget: 1/2 remaining
Agent calls lint() then test()
OK pipeline resets for second deploy
PetriFlow calls confirm: "Allow 'deploy'?"
Human approves → manual gate opens
Agent calls deploy({ environment: "staging" })
ALLOWED Budget: 0/2 remaining
A third deploy attempt
BLOCKED rate limit reached, budget exhausted

Key observations:

  • The chained rules create a pipeline: lint → test → approval → deploy. You can't skip steps.
  • The pipeline resets after each deploy. The agent must lint and test again for the next deploy.
  • Human approval fires every time deploy becomes reachable, not just once.
  • After 2 deploys, the rate limit kicks in permanently for this session.
  • checkStatus and rollback work at any point. They're free tools.

Concepts

Chained sequencingMultiple require ... before ... rules compose into a pipeline. Each step must complete in order.
Human approval gatesrequire human-approval before X creates a manual transition. The confirm callback controls the gate.
Rate limitslimit X to N per session places N tokens in a budget. Each call consumes one. Budget exhaustion is permanent.
Pipeline resetSequencing rules cycle. After each deploy, the pipeline resets and the agent must lint and test again.
AND compositionAll rules (all nets) must agree. Deploy requires passing lint, test, approval, and having budget remaining.

Next steps

Tutorial 3: Discord BotDot notation tool mapping and action-dispatch tools.
Manual confirmation docsCLI prompts, React dialogs, and other confirm patterns.
Rules Engine referenceFull DSL syntax and composition model.