Human-in-the-Loop Approvals

npm add @rivet-dev/workflows rivetkit@3.0.0-preview.1 hono @hono/node-server

SLACK_BOT_TOKEN=xoxb-...
SLACK_SIGNING_SECRET=...
APPROVER_SLACK_ID=U...
AGENT_SLACK_CHANNEL=C...
ORDERS_API_URL=https://api.example.com

// Tools that change something important. The approval workflow runs them only after a person approves.
// Each tool gets the approval request id, so your API can ignore a repeated call for the same request.
const tools = {
	refund_order: async (
		args: { orderId: string; amountCents: number },
		requestId: string,
	) => {
		const apiUrl = process.env.ORDERS_API_URL ?? "https://api.example.com";
		const res = await fetch(`${apiUrl}/orders/${args.orderId}/refunds`, {
			method: "POST",
			headers: {
				"content-type": "application/json",
				"idempotency-key": requestId,
			},
			body: JSON.stringify({ amountCents: args.amountCents }),
		});
		if (!res.ok) throw new Error(`refund failed with status ${res.status}`);
		return (await res.json()) as { refundId: string };
	},
};

type Tools = typeof tools;

// A tool name with matching arguments, such as { tool: "refund_order", args: { orderId, amountCents } }.
export type ToolCall = {
	[K in keyof Tools]: { tool: K; args: Parameters<Tools[K]>[0] };
}[keyof Tools];

export async function runTool(call: ToolCall, requestId: string) {
	return await tools[call.tool](call.args, requestId);
}

// Wait for a decision message, or give up after the timeout. Both are durable.
const [decision] = await ctx.queue.nextBatch("wait-decision", {
	names: ["decision"],
	timeout: DECISION_TIMEOUT_MS,
});

// No decision means the timeout fired first.
let outcome: Outcome = {
	status: decision ? "rejected" : "expired",
	decidedBy: decision?.body.decidedBy ?? null,
	result: null,
	error: null,
};
if (decision?.body.approved) {
	// The tool runs only here, after a person approved it.
	const run = await ctx.tryStep({
		name: "run-tool",
		maxRetries: 5,
		retryBackoffBase: 1000,
		run: async () => await runTool(request, requestId),
	});
	outcome = run.ok
		? { ...outcome, status: "approved", result: run.value }
		: { ...outcome, status: "failed", error: run.failure.error.message };
}

await ctx.step("record-outcome", async (step) => {
	step.state.status = outcome.status;
	step.state.outcome = outcome;
	step.broadcast("finished", outcome);
});
await ctx.step("notify-requester", async () => {
	await postMessage(
		request.requestedBy,
		`Your request to run \`${request.tool}\` ended as ${outcome.status}.`,
	);
});

import { serve } from "@hono/node-server";
import { setup } from "@rivet-dev/workflows";
import { Hono } from "hono";
import { createClient } from "rivetkit/client";
import { approval } from "./approval";
import { respond, type SlackButtonClick, verifySlackRequest } from "./slack";

export const registry = setup({ use: { approval } });
const client = createClient<typeof registry>();

registry.start();

const app = new Hono();

// Slack calls this when the approver clicks Approve or Reject.
app.post("/approvals", async (c) => {
	const rawBody = await c.req.text();
	const signed = verifySlackRequest(
		rawBody,
		c.req.header("x-slack-request-timestamp"),
		c.req.header("x-slack-signature"),
	);
	if (!signed) return c.text("invalid signature", 401);

	const click = JSON.parse(
		new URLSearchParams(rawBody).get("payload") ?? "{}",
	) as SlackButtonClick;
	const [button] = click.actions;
	const handle = client.approval.get([button.value]);

	// The pending request is readable while the workflow waits.
	const { status, request } = await handle.getStatus();
	if (status !== "pending") {
		await respond(click.response_url, {
			response_type: "ephemeral",
			replace_original: false,
			text: `This request is already ${status}.`,
		});
		return c.body(null, 200);
	}
	if (click.user.id !== request?.approver) {
		await respond(click.response_url, {
			response_type: "ephemeral",
			replace_original: false,
			text: "Only the assigned approver can answer.",
		});
		return c.body(null, 200);
	}

	const approved = button.action_id === "approve";
	await handle.send("decision", { approved, decidedBy: click.user.id });
	// Replace the buttons so nobody clicks twice.
	await respond(click.response_url, {
		replace_original: true,
		text: `${approved ? "Approved" : "Rejected"} by <@${click.user.id}>.`,
	});
	return c.body(null, 200);
});

serve({ fetch: app.fetch, port: Number(process.env.PORT ?? 3000) });

import { randomUUID } from "node:crypto";
import { createClient } from "rivetkit/client";
import type { ApprovalRequest, Outcome } from "./approval";
import type { registry } from "./server";

const client = createClient<typeof registry>();

// Starts an approval workflow for a new request and resolves when it finishes.
export async function requestApproval(
	request: ApprovalRequest,
): Promise<Outcome> {
	const handle = client.approval.getOrCreate([randomUUID()]);
	const conn = handle.connect();
	try {
		const finished = new Promise<Outcome>((resolve) => {
			conn.once("finished", (outcome) => resolve(outcome as Outcome));
		});
		await handle.send("request", request);
		return await finished;
	} finally {
		await conn.dispose();
	}
}

// An agent tool that asks a person before it refunds an order.
export const refundOrderTool = {
	name: "refund_order",
	description:
		"Refund part or all of an order. A person approves every refund before it happens.",
	parameters: {
		type: "object",
		properties: {
			orderId: { type: "string" },
			amountCents: { type: "number" },
			reason: { type: "string" },
		},
		required: ["orderId", "amountCents", "reason"],
	},
	execute: async (args: {
		orderId: string;
		amountCents: number;
		reason: string;
	}) => {
		const outcome = await requestApproval({
			tool: "refund_order",
			args: { orderId: args.orderId, amountCents: args.amountCents },
			summary: `Refund $${(args.amountCents / 100).toFixed(2)} on order ${args.orderId}. Reason: ${args.reason}`,
			requestedBy: process.env.AGENT_SLACK_CHANNEL ?? "C0AGENT",
			approver: process.env.APPROVER_SLACK_ID ?? "U0APPROVER",
		});
		if (outcome.status === "approved")
			return `Refund done: ${JSON.stringify(outcome.result)}`;
		return `No refund was made. The request ended as ${outcome.status}.`;
	},
};

npx tsx server.ts

import { refundOrderTool } from "./agent-tool";

// Call the tool the way an agent would. This waits until someone approves or rejects the refund.
const answer = await refundOrderTool.execute({
	orderId: "order-1042",
	amountCents: 2500,
	reason: "The package arrived damaged.",
});
console.log(answer);

npx tsx client.ts

import { createClient } from "rivetkit/client";
import type { registry } from "./server";

// Answer a request without Slack: npx tsx decide.ts <request-id> approve|reject
const [requestId, answer] = process.argv.slice(2);
const client = createClient<typeof registry>();
const handle = client.approval.get([requestId]);

const { request } = await handle.getStatus();
await handle.send("decision", {
	approved: answer === "approve",
	decidedBy: request?.approver ?? "unknown",
});
console.log(`sent ${answer} for request ${requestId}`);

npx tsx decide.ts <request-id> approve