Anthropic Document Pipeline for Klaviyo SMB Email Campaigns

Generate personalized marketing email content from product feeds and customer segments using Anthropic Claude, and push directly to Klaviyo for SMB campaign execution.

anthropic klaviyo email-marketing document-pipeline nextjs hybrid-rag structured-output budget-enforcement

The problem

Small business marketers spend hours manually creating email campaigns, repurposing product data and customer segments across multiple channels without consistency.

Built from

Intro

This tutorial walks you through building an email campaign automation pipeline that loads product catalogs, fetches customer segments from Klaviyo, drafts personalized email content with Anthropic Claude, repairs the LLM output with Zod-driven validation, and pushes the final campaigns back to Klaviyo. You’ll use REAA packages for budget enforcement, semantic caching, structured output repair, prompt versioning, and cost telemetry. By the end you’ll have a Next.js API that can generate and schedule segmented email campaigns with a single HTTP call.

Prerequisites

Node.js >= 22 and pnpm installed on your machine
A Klaviyo account with an API key that has campaign and segment read/write permissions
An Anthropic API key for Claude
A VoyageAI API key for embedding generation (used by the semantic cache)
(Optional) A Langfuse account for LLM cost telemetry tracing
Familiarity with Next.js App Router patterns and TypeScript

Step 1: Scaffold the project and install dependencies

Create a new Next.js project with pnpm. This sets up the App Router shell, TypeScript config, ESLint, and Vitest.

terminal

pnpm create next-app@latest anthropic-email-pipeline --typescript --eslint --app --src-dir --import-alias "@/*"

Example artifact

A complete, working implementation of this recipe — downloadable as a zip or browsable file by file. Generated by our build pipeline; tested with full coverage before publishing.

Download example (zip)Browse files

170 kB·92 tests·99.5% coverage·vitest passing

SHA-256ce8528e773718562f71f8affd07348ef7ae3040371b0fe17c3f73e8e5f718677

Book a conversation All solutions

Comments

Loading comments…

import { BudgetController } from "@reaatech/agent-budget-engine"; import { SpendStore } from "@reaatech/agent-budget-spend-tracker"; export interface BudgetServiceOptions { dailyLimit: number; scopeKey: string; } export interface BudgetCheckResult { allowed: boolean; remaining: number; } export class BudgetService { controller: BudgetController; scopeKey: string; constructor(options: BudgetServiceOptions) { this.scopeKey = options.scopeKey; const store = new SpendStore(); this.controller = new BudgetController({ spendTracker: store }); const ctrl = this.controller as never; const budgetPromise = (ctrl as { defineBudget: (args: Record<string, unknown>) => Promise<unknown> }).defineBudget({ scopeType: "user", scopeKey: this.scopeKey, limit: options.dailyLimit, policy: { softCap: 0.8, hardCap: 1.0 }, }); budgetPromise.catch(() => {}); } async checkCost(estimatedCost: number, modelId: string): Promise<BudgetCheckResult> { const ctrl = this.controller as never; const result = await (ctrl as { check: (args: Record<string, unknown>) => Promise<Record<string, unknown>> }).check({ scopeType: "user", scopeKey: this.scopeKey, estimatedCost, modelId, tools: [], }); return { allowed: Boolean(result.allowed), remaining: Number(result.remaining), }; } recordSpend( requestId: string, cost: number, inputTokens: number, outputTokens: number, modelId: string, ): void { const ctrl = this.controller as never; const promise = (ctrl as { record: (args: Record<string, unknown>) => Promise<unknown> }).record({ requestId, scopeType: "user", scopeKey: this.scopeKey, cost, inputTokens, outputTokens, modelId, provider: "anthropic", timestamp: new Date(), }); promise.catch(() => {}); } async getState(): Promise<{ spent: number; remaining: number; state: string }> { const ctrl = this.controller as never; const state = await (ctrl as { getState: (scopeType: string, scopeKey: string) => Promise<Record<string, unknown>> }).getState("user", this.scopeKey); return { spent: Number(state.spent ?? 0), remaining: Number(state.remaining ?? 0), state: typeof state.state === "string" ? state.state : "Unknown", }; } }

import { CacheEngine, InMemoryAdapter, type CacheResult } from "@reaatech/llm-cache"; import { VoyageAIClient } from "voyageai"; interface EmbeddingProvider { embed(inputs: string[]): Promise<number[][]>; } class VoyageEmbeddingProvider implements EmbeddingProvider { private client: VoyageAIClient; private model: string; constructor(apiKey: string, model?: string) { this.client = new VoyageAIClient({ apiKey }); this.model = model ?? "voyage-3"; } async embed(inputs: string[]): Promise<number[][]> { const response = await this.client.embed({ input: inputs, model: this.model, }); return (response.data?.map((d) => d.embedding) ?? []).filter((e): e is number[] => e !== undefined); } } export interface CacheServiceOptions { voyageApiKey: string; similarityThreshold?: number; upstashUrl?: string; upstashToken?: string; } export class CacheService { engine: CacheEngine; constructor(options: CacheServiceOptions) { const embedder = new VoyageEmbeddingProvider(options.voyageApiKey); this.engine = new CacheEngine({ storage: new InMemoryAdapter(), vectorStorage: new InMemoryAdapter(), embedder: embedder as never, config: { storage: { adapter: "memory" }, vectorStorage: { adapter: "memory" }, embedding: { provider: "openai" as const, model: "voyage-3", dimensions: 1024, batchSize: 100, maxRetries: 3, }, similarity: { threshold: options.similarityThreshold ?? 0.8, metric: "cosine", maxResults: 10, }, ttl: { default: 3600, factual: 7200, creative: 7200, analytical: 3600, sensitive: 600, byUseCase: {}, }, segmentation: { enabled: true, defaultUseCase: "general" }, cost: { enabled: false, currency: "USD" }, observability: { metrics: false, tracing: false, logging: "info" }, }, }); } async getCached( prompt: string, options: { model: string; modelVersion: string; useCase: string }, ): Promise<CacheResult> { return this.engine.get(prompt, options); } async setCached( prompt: string, response: unknown, options: { model: string; modelVersion: string; useCase: string }, metadata?: { queryType: "creative"; tokens: { prompt: number; completion: number } }, ) { await this.engine.set(prompt, response, options, metadata); } async invalidate(useCase: string) { await this.engine.invalidate({ useCase }); } }

import { type Document, type Chunk, type ChunkingConfig } from "@reaatech/hybrid-rag"; import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters"; import { Readability, isProbablyReaderable } from "@mozilla/readability"; import { JSDOM } from "jsdom"; export interface ProductFeedItem { id: string; title: string; description: string; price: number; category: string; imageUrl: string; sku: string; } export class ProductFeedLoader { async loadFromJson(url: string): Promise<ProductFeedItem[]> { const response = await fetch(url); if (!response.ok) { throw new Error(`Failed to fetch product feed: ${String(response.status)} ${response.statusText}`); } const data = (await response.json()) as ProductFeedItem[]; return data; } async loadFromWebPage(url: string): Promise<string> { const response = await fetch(url); if (!response.ok) { throw new Error(`Failed to fetch web page: ${String(response.status)} ${response.statusText}`); } const html = await response.text(); const dom = new JSDOM(html, { url }); const doc = dom.window.document; if (!isProbablyReaderable(doc)) { return ""; } const reader = new Readability(doc); const article = reader.parse(); return article ? (article.textContent ?? "") : ""; } toDocuments(items: ProductFeedItem[]): Document[] { return items.map((item) => ({ id: item.id, content: item.description, source: "product-feed", metadata: { title: item.title, price: item.price, category: item.category, sku: item.sku, }, })); } async chunkDocuments(docs: Document[], config: ChunkingConfig): Promise<Chunk[]> { const splitter = new RecursiveCharacterTextSplitter({ chunkSize: config.chunkSize, chunkOverlap: config.overlap, }); const chunks: Chunk[] = []; for (const doc of docs) { const texts = await splitter.splitText(doc.content); for (let idx = 0; idx < texts.length; idx++) { const text = texts[idx]; chunks.push({ id: `${doc.id}-chunk-${String(idx)}`, documentId: doc.id, content: text, tokenCount: text.length, position: idx, metadata: doc.metadata, index: idx, characterCount: text.length, startPosition: idx * 200, endPosition: idx * 200 + text.length, strategy: "recursive", } as Chunk); } } return chunks; } }

import styles from "./page.module.css"; export default function Home() { return ( <div className={styles.page}> <main className={styles.main}> <h1>Anthropic Document Pipeline for Klaviyo SMB Email Campaigns</h1> <p className={styles.description}> Generate personalized marketing email content from product feeds and customer segments using Anthropic Claude, and push directly to Klaviyo for SMB campaign execution. </p> <div className={styles.endpoints}> <h2>API Endpoints</h2> <ul> <li> <code>POST /api/generate</code> — Generate email content for a segment </li> <li> <code>POST /api/campaigns</code> — Run the full pipeline and create a campaign </li> <li> <code>POST /api/webhooks/klaviyo</code> — Receive Klaviyo event notifications </li> </ul> </div> <div className={styles.info}> <h2>How It Works</h2> <ol> <li>Load product feed from a JSON data source</li> <li>Retrieve active customer segments via Klaviyo API</li> <li>Pass products and segments to Anthropic Claude to draft email content</li> <li>Repair and validate output with structured-repair-core</li> <li>Create the campaign in Klaviyo</li> </ol> </div> <div className={styles.examples}> <h2>API Usage Examples</h2> <h3>Generate email content</h3> <pre>{`curl -X POST http://localhost:3000/api/generate \\\\ -H "Content-Type: application/json" \\\\ -d '{ "productFeedUrl": "https://example.com/feed.json", "segmentName": "VIP Customers", "campaignGoal": "Promote new arrivals", "tone": "professional" }'`}</pre> <h3>Create a campaign</h3> <pre>{`curl -X POST http://localhost:3000/api/campaigns \\\\ -H "Content-Type: application/json" \\\\ -d '{ "productFeedUrl": "https://example.com/feed.json", "listId": "YOUR_LIST_ID", "campaignGoal": "Weekly newsletter", "tone": "casual" }'`}</pre> <h3>Health check</h3> <pre>{`curl http://localhost:3000/api/generate`}</pre> </div> </main> </div> ); }

import { setupServer } from "msw/node"; import { http, HttpResponse } from "msw"; import { beforeAll, afterEach, afterAll } from "vitest"; export const server = setupServer( http.post("https://api.anthropic.com/v1/messages", () => HttpResponse.json({ id: "msg_test", type: "message", role: "assistant", model: "claude-sonnet-4-6", content: [{ type: "text", text: "mocked response" }], stop_reason: "end_turn", usage: { input_tokens: 10, output_tokens: 5 }, }), ), http.post("https://api.voyageai.com/v1/embeddings", () => HttpResponse.json({ data: [{ embedding: [0.1, 0.2, 0.3] }, { embedding: [0.4, 0.5, 0.6] }], model: "voyage-3", }), ), http.all("https://a.klaviyo.com/api/*", ({ request }) => { const url = new URL(request.url); if (url.pathname.includes("segments")) { return HttpResponse.json({ data: [ { id: "seg-1", attributes: { name: "VIP Customers" } }, { id: "seg-2", attributes: { name: "New Subscribers" } }, ], }); } if (url.pathname.includes("campaigns")) { if (request.method === "POST") { return HttpResponse.json({ data: { id: "camp-123" } }); } return HttpResponse.json({ data: [ { id: "camp-1", attributes: { name: "Test Campaign" } }, ], }); } if (url.pathname.includes("profiles")) { return HttpResponse.json({ data: [] }); } return HttpResponse.json({ data: [] }); }), http.get("https://example.com/*", ({ request }) => { const url = new URL(request.url); if (url.pathname.includes("feed.json")) { return HttpResponse.json([ { id: "p1", title: "Product 1", description: "Desc 1", price: 10, category: "Cat1", imageUrl: "https://example.com/img.jpg", sku: "SKU001" }, ]); } if (url.pathname.includes("not-found")) { return new Response(null, { status: 404, statusText: "Not Found" }); } if (url.pathname.includes("nonreaderable")) { return HttpResponse.text("<html><body><p>Hi</p></body></html>"); } return HttpResponse.text("<html><body><article><h1>Test Article That Is Long Enough For Reader View</h1><p>" + "This is a lengthy article about product descriptions. It contains enough text to pass the isProbablyReaderable check, which requires a minimum content length to determine if the page is suitable for parsing. ".repeat(5) + "</p></article></body></html>"); }), ); beforeAll(() => { server.listen({ onUnhandledRequest: "error" }); }); afterEach(() => { server.resetHandlers(); }); afterAll(() => { server.close(); });

import { describe, it, expect } from "vitest"; import { ContentRepair } from "../src/services/content-repair.js"; import { EmailContentSchema } from "../src/services/content-repair.js"; describe("ContentRepair", () => { const repair = new ContentRepair(); it("repairEmailContent passes valid JSON through", async () => { const valid = '{"subject":"Hello","previewText":"Preview","bodyHtml":"<p>Hello</p>","bodyText":"Hello"}'; const result = await repair.repairEmailContent(valid); expect(result.subject).toBe("Hello"); expect(result.previewText).toBe("Preview"); expect(result.bodyHtml).toBe("<p>Hello</p>"); expect(result.bodyText).toBe("Hello"); }); it("repairEmailContent repairs malformed JSON with fences", async () => { const malformed = '```json\n{"subject":"Test","previewText":"Preview","bodyHtml":"<p>Test</p>","bodyText":"Test"}\n```'; const result = await repair.repairEmailContent(malformed); expect(result.subject).toBe("Test"); }); it("repairEmailContent repairs trailing comma", async () => { const malformed = '{"subject":"Test","previewText":"Preview","bodyHtml":"<p>Test</p>","bodyText":"Test",}'; const result = await repair.repairEmailContent(malformed); expect(result.subject).toBe("Test"); }); it("repairEmailContent coerces string to number in fields", async () => { const coerced = '{"subject":"Test","previewText":"Preview","bodyHtml":"<p>Test</p>","bodyText":"Test"}'; const result = await repair.repairEmailContent(coerced); expect(typeof result.subject).toBe("string"); }); it("repairWithDiagnostics returns step metadata", () => { const result = repair.repairWithDiagnostics('{"subject":"S","previewText":"P","bodyHtml":"<p>B</p>","bodyText":"B"}'); expect(result.success).toBe(true); }); it("validateRaw returns true for valid data", () => { const valid = '{"subject":"S","previewText":"P","bodyHtml":"<p>B</p>","bodyText":"B"}'; expect(repair.validateRaw(JSON.parse(valid))).toBe(true); }); it("validateRaw returns false for invalid data", () => { expect(repair.validateRaw({ not: "valid" })).toBe(false); }); it("EmailContentSchema infers correct types", () => { const valid = { subject: "Test Subject", previewText: "Preview", bodyHtml: "<p>Test</p>", bodyText: "Test", }; const result = EmailContentSchema.safeParse(valid); expect(result.success).toBe(true); }); it("EmailContentSchema rejects missing required fields", () => { const result = EmailContentSchema.safeParse({ subject: "Only subject" }); expect(result.success).toBe(false); }); });

Anthropic Document Pipeline for Klaviyo SMB Email Campaigns

The problem

Built from

Intro

Prerequisites

Step 1: Scaffold the project and install dependencies

Example artifact

Comments

Intro

Prerequisites

Step 1: Scaffold the project and install dependencies

Step 2: Create the Klaviyo API client wrapper

Step 3: Build the Anthropic LLM client

Step 4: Add budget enforcement, caching, and telemetry services

Budget service

Cache service with VoyageAI embeddings

Telemetry service with Langfuse

Step 5: Load and chunk product feeds

Step 6: Generate email content with Claude

Step 7: Repair LLM output with structured-repair-core

Step 8: Create the segment and campaign service wrappers

Step 9: Build the email campaign pipeline

Step 10: Wire up the API routes

Step 11: Set up the public API exports

Step 12: Add tests and verify

Next steps