Core

Design Patterns

Patterns for agent keys, subagents, agent-to-agent messages, workflows, schedules, background jobs, and shared credentials.

Each agent is an Actor, so agent apps use the same building blocks as the rest of Rivet: keys pick the agent, and agents call other Actors.

One Agent per Key

The key decides which conversation a prompt continues. Pick it from what the agent should remember:

Per user: a personal assistant remembers everything a user has asked, across every chat.
Per thread: each Slack thread, support ticket, or chat tab is its own conversation.
Per task: each job, such as fixing one GitHub issue, starts fresh in its own sandbox. When it’s done, an action that calls c.destroy() deletes the agent and its sandbox, like finish below.

import { createRegistry } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import { setup } from "rivetkit";

// The chat agents have no tools. The issue fixer gets read, write, edit, and bash in a sandbox.
const noTools = createRegistry();
const codingTools = createRegistry();
codingTools.install(CodingTools);

const assistant = pi({ model: "anthropic/claude-opus-5-5", registry: noTools });

const slackThread = pi({ model: "anthropic/claude-haiku-4-5", registry: noTools });

const issueFixer = pi({
	model: "anthropic/claude-opus-5-5",
	registry: codingTools,
	sandbox: e2bProvider(),
	actions: {
		finish: (c) => c.destroy(),
	},
});

export const registry = setup({ use: { assistant, slackThread, issueFixer } });

registry.start();

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

const client = createClient<typeof registry>();

export async function chat(userId: string, message: string) {
	const result = await client.assistant.getOrCreate([userId]).prompt(message);
	return result.status === "done" ? result.text : undefined;
}

export async function onSlackMessage(channel: string, threadTs: string, text: string) {
	const result = await client.slackThread.getOrCreate([channel, threadTs]).prompt(text);
	return result.status === "done" ? result.text : undefined;
}

export async function onIssueLabeled(repo: string, issue: { number: number; title: string; body: string }) {
	const fixer = client.issueFixer.getOrCreate([repo, String(issue.number)]);
	try {
		const result = await fixer.prompt(
			`Clone https://github.com/${repo}, fix issue #${issue.number}, and open a pull request.\n\n# ${issue.title}\n\n${issue.body}`,
		);
		return result.status === "done" ? result.text : undefined;
	} finally {
		await fixer.finish();
	}
}

Keys are arrays, so a channel id or repository name from a webhook can’t break the key’s structure. See Actor Keys.

The prompt action uses the agent’s root conversation, so by default the key is the conversation. An agent can also hold more conversations and forks, which lets related conversations share one Actor, one sandbox, and its documents.

Lead Agent with Subagents

A lead agent hands a focused task to a specialist with its own tools and waits for its answer. The specialist starts with a fresh context, so only the answer comes back.

import { createRegistry, defineExtension } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { setup } from "rivetkit";
import { askSpecialist, engineering, orders } from "./specialists";

const extensions = createRegistry();
extensions.install(defineExtension({ name: "support", tools: [askSpecialist] }));

const support = pi({ model: "anthropic/claude-opus-5-5", registry: extensions });

export const registry = setup({ use: { support, orders, engineering } });

registry.start();

import { Type } from "@earendil-works/pi-ai";
import { createRegistry, defineExtension, defineTool, ROOT_CONVERSATION_ID } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import type { Registry } from "rivetkit";
import { createClient } from "rivetkit/client";
import { getOrder } from "./get-order";

const orderTools = createRegistry();
orderTools.install(defineExtension({ name: "orders", tools: [getOrder] }));

export const orders = pi({ model: "openai/gpt-5.4-mini", registry: orderTools });

const codingTools = createRegistry();
codingTools.install(CodingTools);

export const engineering = pi({ model: "anthropic/claude-opus-5-5", registry: codingTools, sandbox: e2bProvider() });

export const askSpecialist = defineTool({
	name: "ask_specialist",
	description:
		"Hand one focused question to a specialist and get its answer. Use orders for order status, refunds, and shipping. Use engineering for bugs that need someone to read or run the code.",
	parameters: Type.Object({
		specialist: Type.Union([Type.Literal("orders"), Type.Literal("engineering")]),
		question: Type.String({ description: "Everything the specialist needs. It can't see this conversation." }),
	}),
	// A rerun reaches the same specialist Actor, through the memo below.
	replay: "safe",
	execute: async ({ specialist, question }, api, context) => {
		// The memo is saved with this tool call, so a rerun gets the same id.
		const id = await api.memo("specialist-id", crypto.randomUUID(), context);
		const agent = client[specialist].getOrCreate([id]);
		context.abortSignal?.addEventListener("abort", () => void agent.conversation.abort(ROOT_CONVERSATION_ID));

		const result = await agent.prompt(question, { requestId: id });
		if (result.status === "unanswered") {
			return { content: [{ type: "text", text: `The specialist did not answer: ${result.reason}` }], isError: true };
		}
		const answer = result.text || "The specialist finished without an answer.";
		return { content: [{ type: "text", text: answer }], details: { specialist } };
	},
});

const client = createClient<Registry<{ orders: typeof orders; engineering: typeof engineering }>>();

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

const client = createClient<typeof registry>();
const support = client.support.getOrCreate(["ticket-4821"]);

const result = await support.prompt(
	"Order 1042 shows as delivered but never arrived, and the tracking page throws a 500 error. Repo: https://github.com/acme/storefront",
);
console.log(result.status === "done" ? result.text : `Unanswered: ${result.reason}`);

The get_order tool in get-order.ts is the one from Custom Tools.

See Subagents.

Agents Messaging Each Other

When an agent hands work off and doesn’t need the answer, it sends a message instead of waiting. The receiving agent schedules its own prompt, so the message survives the sender going away.

import { Type } from "@earendil-works/pi-ai";
import { createRegistry, defineExtension, defineTool } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const requestReview = defineTool({
	name: "request_review",
	description: "Send a finished branch to the repository's reviewer. Don't wait for the review.",
	parameters: Type.Object({ repo: Type.String(), branch: Type.String(), summary: Type.String() }),
	execute: async ({ repo, branch, summary }) => {
		await client.reviewer.getOrCreate([repo]).requestReview(`Review the ${branch} branch of https://github.com/${repo}: ${summary}`);
		return { content: [{ type: "text", text: `Sent ${branch} for review.` }] };
	},
});

const extensions = createRegistry();
extensions.install(CodingTools);
extensions.install(defineExtension({ name: "review", tools: [requestReview] }));

export const coder = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	sandbox: e2bProvider(),
});

const client = createClient<typeof registry>();

import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { createRegistry } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import { setup } from "rivetkit";
import { coder } from "./coder";

const extensions = createRegistry();
extensions.install(CodingTools);

const reviewer = pi({
	model: "openai/gpt-5.5",
	registry: extensions,
	sandbox: e2bProvider(),
	actions: {
		// Saves the request and returns without waiting for the review.
		requestReview: async (c, request: string) => {
			const conversation = await c.pi.root(BACKGROUND_CONTEXT);
			await conversation.submit({ type: "input", content: request }, BACKGROUND_CONTEXT);
		},
	},
});

export const registry = setup({ use: { coder, reviewer } });

registry.start();

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

const client = createClient<typeof registry>();
const coder = client.coder.getOrCreate(["acme/app", "fix-flaky-checkout-test"]);

const result = await coder.prompt(
	"Clone https://github.com/acme/app, fix the flaky checkout test on a new fix-flaky-checkout-test branch, push it, and request a review.",
);
console.log(result.status === "done" ? result.text : `Unanswered: ${result.reason}`);

See Agent-to-Agent Messages.

Agent as a Workflow Step

When the work has fixed steps, waits, or retries, let a workflow drive the agent. Each step prompts the agent and waits for its answer.

npm add @rivet-dev/workflows

import { createRegistry } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import { type Registry, setup, workflow } from "@rivet-dev/workflows";

const extensions = createRegistry();
extensions.install(CodingTools);

const agent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	sandbox: e2bProvider(),
});

type Agents = Registry<{ agent: typeof agent }>;

const flakyTest = workflow({
	state: { report: null as string | null },
	run: async (ctx) => {
		await ctx.step({
			name: "first-run",
			timeout: 10 * 60_000,
			run: async (step) => {
				const fixer = step.client<Agents>().agent.getOrCreate([step.actorId]);
				// A retried attempt sends the same requestId, so it waits for the first attempt's run.
				await fixer.prompt("Clone https://github.com/acme/app, run its test suite, and note any failing tests.", {
					requestId: `${step.actorId}:first-run`,
				});
			},
		});

		await ctx.sleep("wait-before-rerun", 10 * 60_000);

		await ctx.step({
			name: "second-run",
			timeout: 10 * 60_000,
			run: async (step) => {
				const fixer = step.client<Agents>().agent.getOrCreate([step.actorId]);
				const result = await fixer.prompt("Run the suite again. Which failures happened both times?", {
					requestId: `${step.actorId}:second-run`,
				});
				step.state.report = result.status === "done" ? (result.text ?? null) : `Unanswered: ${result.reason}`;
			},
		});
	},
	actions: {
		getReport: (c) => c.state.report,
	},
});

export const registry = setup({ use: { agent, flakyTest } });

registry.start();

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

const client = createClient<typeof registry>();

// Creating the workflow starts it. The key names this run.
const run = client.flakyTest.getOrCreate(["acme/app", "2026-10-01"]);

let report = await run.getReport();
while (report === null) {
	await new Promise((resolve) => setTimeout(resolve, 60_000));
	report = await run.getReport();
}
console.log(report);

See Workflows.

Scheduled Agents

An agent can wake itself on a schedule and sleep between runs.

import { createRegistry } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import { setup } from "rivetkit";

const extensions = createRegistry();
extensions.install(CodingTools);

const agent = pi({
	model: "anthropic/claude-sonnet-5-5",
	registry: extensions,
	sandbox: e2bProvider(),
	onCreate: async (c) => {
		const [repo] = c.key;
		await c.cron.set({
			name: "morning-triage",
			expression: "0 9 * * *",
			action: "prompt",
			args: [`Clone https://github.com/${repo}, run its test suite, and summarize any failures.`],
		});
	},
});

export const registry = setup({ use: { agent } });

registry.start();

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

const client = createClient<typeof registry>();
const agent = client.agent.getOrCreate(["acme/app", "morning-triage"]);

// The first call creates the agent, and onCreate sets its cron.
const root = await agent.harness.root();

// Any time after a run, read the latest summary: the last message of the conversation.
const { messages } = await agent.conversation.context(root.id);
const last = messages.at(-1);
let summary = "";
if (last?.role === "assistant") {
	for (const block of last.content) if (block.type === "text") summary += block.text;
}
console.log(summary || "No run yet. The first one starts at 9:00 UTC.");

See Schedules.

Background Agents That Must Finish

When nobody is watching a run, such as an agent started by a webhook, a crash must not lose the work or repeat a side effect. Every model call and tool call of a pi() run is saved before the next step starts, so the run continues where it stopped. Pass a requestId, so a redelivered webhook returns the first submission instead of starting a second run.

import { Type } from "@earendil-works/pi-ai";
import { createRegistry, defineExtension, defineTool } from "@earendil-works/pi-durable";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { pi } from "@rivet-dev/pi";
import { e2bProvider } from "@rivet-dev/sandbox-adapter/e2b";
import { setup } from "rivetkit";

const openPullRequest = defineTool({
	name: "open_pull_request",
	description: "Open a pull request from a pushed branch.",
	parameters: Type.Object({ repo: Type.String(), branch: Type.String(), title: Type.String() }),
	// No replay: if a crash cuts this call off, the model learns it never completed
	// and can check for an existing pull request before opening another.
	execute: async ({ repo, branch, title }, _api, context) => {
		const response = await fetch(`https://api.github.com/repos/${repo}/pulls`, {
			method: "POST",
			headers: { authorization: `Bearer ${process.env.GITHUB_TOKEN}`, accept: "application/vnd.github+json" },
			body: JSON.stringify({ head: branch, base: "main", title }),
			signal: context.abortSignal,
		});
		if (!response.ok) throw new Error(`GitHub returned ${response.status}.`);
		const pull = (await response.json()) as { html_url: string };
		return { content: [{ type: "text", text: pull.html_url }] };
	},
});

const extensions = createRegistry();
extensions.install(CodingTools);
extensions.install(defineExtension({ name: "github", tools: [openPullRequest] }));

const fixer = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	sandbox: e2bProvider(),
});

export const registry = setup({ use: { fixer } });

registry.start();

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

const client = createClient<typeof registry>();

// Called from a GitHub webhook. GitHub redelivers a webhook that times out,
// so the delivery id makes a redelivery return the first submission.
export async function onIssueLabeled(repo: string, issue: number, title: string, deliveryId: string) {
	const fixer = client.fixer.getOrCreate([repo, String(issue)]);
	const root = await fixer.harness.root();
	await fixer.conversation.submit(root.id, {
		type: "input",
		content: `Fix issue #${issue} in ${repo}: ${title}. Push a branch and open a pull request.`,
		requestId: deliveryId,
	});
}

The conversation.submit action returns once the input is saved, so the webhook answers right away while the agent works.
The open_pull_request tool has no replay, so a call cut off by a crash isn’t repeated. The model is told the call was interrupted. Mark only tools that are safe to run twice as replay: "safe".
When you ship new code, the Actor lets the run finish, for up to sleepGracePeriod, before it restarts on the new version. Anything still running then carries on from its last saved step.

See Architecture.

Credentials Shared per Tenant

Start agent keys with the tenant id, and pick the credentials Actor from it. Every agent in a tenant shares one set of model logins, and never sees another tenant’s.

import { createRegistry } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { type Registry, setup } from "rivetkit";
import { credentials } from "./credentials";

const agent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: createRegistry(),
	// Agent keys start with the tenant id, so every agent in a tenant reads the same credentials Actor.
	credentials: (c) => {
		const [tenantId] = c.key;
		const client = c.client<Registry<{ credentials: typeof credentials }>>();
		return client.credentials.getOrCreate([tenantId]);
	},
});

export const registry = setup({ use: { credentials, agent } });

registry.start();

import { type Credential, InMemoryCredentialStore } from "@earendil-works/pi-ai";
import { ModelRuntime } from "@earendil-works/pi-coding-agent";
import type { PiProviderCredential } from "@rivet-dev/pi";
import { actor } from "rivetkit";

export const credentials = actor({
	state: { saved: {} as Record<string, Credential> },
	actions: {
		save: (c, provider: string, credential: Credential) => {
			c.state.saved[provider] = credential;
		},
		list: (c) => Object.entries(c.state.saved).map(([providerId, { type }]) => ({ providerId, type })),
		read: (c, provider: string) => withoutRefreshToken(c.state.saved[provider]),
		refresh: async (c, provider: string) => {
			const store = new InMemoryCredentialStore();
			await store.modify(provider, async () => c.state.saved[provider]);
			const runtime = await ModelRuntime.create({ credentials: store, modelsPath: null });
			await runtime.getAuth(provider, { minOAuthValidityMs: 10 * 60_000 });
			const refreshed = await store.read(provider);
			if (refreshed) c.state.saved[provider] = refreshed;
			return withoutRefreshToken(refreshed);
		},
	},
});

function withoutRefreshToken(credential: Credential | undefined): PiProviderCredential | undefined {
	if (credential?.type !== "oauth") return credential;
	const { refresh: _refresh, ...rest } = credential;
	return rest;
}

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

const client = createClient<typeof registry>();

// Called from your settings page when a tenant admin saves the team's Anthropic key.
export async function saveTeamKey(tenantId: string, anthropicKey: string) {
	await client.credentials.getOrCreate([tenantId]).save("anthropic", { type: "api_key", key: anthropicKey });
}

// Every agent keyed under the tenant uses that key, and never sees another tenant's.
export async function ask(tenantId: string, userId: string, text: string) {
	const result = await client.agent.getOrCreate([tenantId, userId]).prompt(text);
	return result.status === "done" ? result.text : undefined;
}

See User Subscriptions.

Anti-Patterns

One Agent for Every User

With a single key for everyone, every user shares one conversation and its history. A prompt from one user waits in the queue while another user’s prompt runs.

import { createClient } from "rivetkit/client";
import type { registry } from "./agent-per-key/server";

const client = createClient<typeof registry>();

export async function answer(userId: string, message: string) {
	const result = await client.assistant.getOrCreate(["support"]).prompt(`${userId}: ${message}`);
	return result.status === "done" ? result.text : undefined;
}

Solution: Key the agent by user, thread, or task.

A New Key per Request

A new key creates a new agent with an empty conversation, so the agent forgets the conversation after every message and leaves an Actor behind.

import { createClient } from "rivetkit/client";
import type { registry } from "./agent-per-key/server";

const client = createClient<typeof registry>();

export async function answer(message: string) {
	const result = await client.assistant.getOrCreate([crypto.randomUUID()]).prompt(message);
	return result.status === "done" ? result.text : undefined;
}

Solution: Reuse the key of the conversation the message belongs to.

Webhooks Without a `requestId`

Webhook providers deliver the same event again when your handler is slow or fails. Without a requestId, each delivery starts its own run, so the agent does the work twice and a tool with side effects can run twice.

Solution: Pass the delivery id as the requestId of every submission. See Background Agents That Must Finish.

Where to Go Next

Goal	Read
Give agents model access	LLM API Keys
Give the agent your own APIs	Custom Tools
Let the agent run commands and edit files	Sandboxes, then Built-in Tools
Run agents on your users’ own subscriptions	User Subscriptions
Make runs finish after a crash, with no one to retry them	Architecture
Wait for a person before a risky action	Human in the Loop
Share a result through a link	Scoped Access with JWTs
Talk to users in Slack, Linear, GitHub, or Discord	Slack and the other connectors
See what happens while an agent answers a prompt	Session Lifecycle
Build a chat UI	React SDK
Secure and deploy agents	Security Model, then Deploy