Small businesses using Mixpanel struggle to extract deeper insights beyond dashboards—they need custom queries but lack data science skills and fear runaway cloud costs from AI-generated code.
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.
This tutorial builds a self-service Mixpanel analyst — a Next.js app where small business owners type natural-language questions like “What were my top events this week?” and get Python-powered insights back. Cohere’s LLM generates the analysis code, e2b executes it in a secure sandbox, and REAA packages enforce per-user budgets, cache similar queries, repair malformed LLM output, and track every penny of cost.
You’ll end up with a working API endpoint (POST /api/analyze) and a dashboard UI that anyone can open in their browser.
Prerequisites
Node.js >= 22 and pnpm >= 10 installed
API keys for Cohere, E2B, Mixpanel (OAuth token + project ID), OpenAI (for the cache embedder), and Langfuse (for cost tracing)
Basic familiarity with TypeScript, Next.js App Router, and REST APIs
Step 1: Scaffold the project and install dependencies
Create your Next.js project and install all dependencies. The dependencies fall into four groups:
Expected output:pnpm install completes without errors, and ls node_modules/@reaatech/ shows all REAA packages.
Step 2: Configure environment variables
Copy the example env file and fill in your API keys. Create .env.local (Next.js auto-loads it):
terminal
cp .env.example .env.local
The .env.example should contain every variable the app reads. Create it with:
env
# Env vars used by cohere-code-sandbox-for-mixpanel-small-business-analytics.# Keep placeholders only — never commit real values.NODE_ENV=development# Cohere API key for LLM code generationCOHERE_API_KEY=<your-cohere-api-key># E2B API key for sandboxed Python executionE2B_API_KEY=<your-e2b-api-key># Mixpanel OAuth token and project ID for fetching analytics eventsMIXPANEL_OAUTH_TOKEN=<your-mixpanel-oauth-token>MIXPANEL_PROJECT_ID=<your-mixpanel-project-id># OpenAI API key for llm-cache embedding providerOPENAI_API_KEY=<your-openai-api-key># Langfuse credentials for observability tracingLANGFUSE_PUBLIC_KEY=<your-langfuse-public-key>LANGFUSE_SECRET_KEY=<your-langfuse-secret-key># Budget configurationDEFAULT_USER_BUDGET_LIMIT=10.0BUDGET_SOFT_CAP=0.8BUDGET_HARD_CAP=1.0# Cache configurationCACHE_SIMILARITY_THRESHOLD=0.8CACHE_DEFAULT_TTL=3600# Cohere model override (defaults to command-a-03-2025)COHERE_MODEL=command-a-03-2025
Expected output: The file .env.local exists with your real API keys filled in where the <...> placeholders were.
Step 3: Define types, Zod schemas, and constants
Create src/lib/types.ts with all shared types, validation schemas, and the BudgetScope enum re-exported from @reaatech/agent-budget-types. The AnalyzeRequestSchema validates incoming API requests, AnalysisResultSchema defines the shape that sandbox output must conform to, and supporting types describe Mixpanel events and spend records.
Now create src/lib/constants.ts. This holds pricing, model IDs, budget defaults, and cache configuration — all with env-var fallbacks so the app is configurable without code changes.
Expected output:pnpm typecheck reports zero errors after both files are created.
Step 4: Build the in-memory spend store
@reaatech/agent-budget-engine expects a SpendStore interface for tracking spend. Extend the base SpendStore class from @reaatech/agent-budget-spend-tracker with a lightweight in-memory implementation backed by a Map.
Create src/lib/spend-store.ts:
ts
import { SpendStore } from '@reaatech/agent-budget-spend-tracker';import type { SpendRecord } from "./types.js";export class InMemorySpendStore extends SpendStore { private internalStore = new Map<string, SpendRecord[]>(); getSpent(scopeType: string, scopeKey: string): Promise<number> { const key = `${scopeType}:${scopeKey}`; const entries = this.internalStore.get(key); if (!entries || entries.length === 0) return Promise.resolve(0); return Promise.resolve(entries.reduce((sum, e) => sum + e.cost, 0)); } record(entry: SpendRecord): number { const key = `${entry.scopeType}:${entry.scopeKey}`; const existing = this.internalStore.get(key) ?? []; existing.push(entry); this.internalStore.set(key, existing); return 0; } reset(scopeType: string, scopeKey: string): void { this.internalStore.delete(`${scopeType}:${scopeKey}`); }}let instance: InMemorySpendStore | undefined;export function createSpendStore(): InMemorySpendStore { if (!instance) { instance = new InMemorySpendStore(); } return instance;}
The factory createSpendStore() returns a singleton so all budget checks share the same in-memory state. Extending SpendStore ensures the class satisfies the interface that BudgetController expects.
Step 5: Wire up budget enforcement
The budget service wraps BudgetController from @reaatech/agent-budget-engine. It uses the BudgetScope.User enum from @reaatech/agent-budget-types (re-exported through types.ts) to scope per-user budgets with soft/hard cap policies, checks remaining quota before each query, records actual spend after execution, and subscribes to lifecycle events for logging.
The BudgetScope.User enum value comes from @reaatech/agent-budget-types and is re-exported through src/lib/types.ts for convenience. Every scope reference in this file uses it rather than raw strings.
Expected output:pnpm typecheck reports zero errors.
Step 6: Add cost telemetry
The telemetry service uses @reaatech/llm-cost-telemetry to create structured CostSpan objects with unique IDs, timestamps, and cost calculations. It validates each span with CostSpanSchema before logging it to both the console and Langfuse.
The cache service wraps CacheEngine from @reaatech/llm-cache with in-memory storage and an OpenAI-powered semantic embedder. It provides two operations — getCachedResponse and storeResponse — segmented under the "mixpanel-analytics" use case so different recipes never collide.
When Cohere generates Python code, the sandbox’s output may be noisy — JSON wrapped in markdown fences, type mismatches, or truncated. @reaatech/structured-repair-core handles all of that. Create src/services/repair-service.ts:
This service sends Mixpanel event data plus the user’s question to Cohere’s chat API and extracts Python code from the markdown-fenced response. It wraps the call with p-retry for resilience.
Create src/services/cohere-service.ts:
ts
import { CohereClientV2, CohereError, CohereTimeoutError } from "cohere-ai";import pRetry from "p-retry";import { COHERE_MODEL } from "../lib/constants.js";import type { MixpanelEvent } from "../lib/types.js";const cohere = new CohereClientV2({});function buildPrompt(mixpanelData: MixpanelEvent[], userQuery: string): string { return `You are a Python data analyst for Mixpanel. Analyze the following event data and the user's query.Event data:${JSON.stringify(mixpanelData, null, 2)}User query: ${userQuery}Write Python code using pandas and matplotlib to analyze this data and generate insights.Output ONLY the Python code in a markdown code block with the \`\`\`python fence.`;}function extractCodeFromMarkdown(text: string): string { const match = text.match(/```python\n([\s\S]*?)```/); return match ? match[1].trim() : text.trim();}export const cohereService = { async generatePythonCode( mixpanelData: MixpanelEvent[], userQuery: string, ): Promise<{ code: string; inputTokens: number; outputTokens: number }> { const prompt = buildPrompt(mixpanelData, userQuery); const execute = async () => { const response = await cohere.chat({ model: COHERE_MODEL, messages: [{ role: "user", content: prompt }], }); const firstContent = response.message.content?.[0]; const responseText = firstContent?.type === "text" ? firstContent.text : ""; const code = extractCodeFromMarkdown(responseText); return { code, inputTokens: response.usage?.tokens?.inputTokens ?? 0, outputTokens: response.usage?.tokens?.outputTokens ?? 0, }; }; return pRetry(execute, { retries: 3, onFailedAttempt: (ctx) => { if (ctx.error instanceof CohereError) { console.error("Cohere API error", ctx.error.message); } else if (ctx.error instanceof CohereTimeoutError) { console.error("Cohere timeout", ctx.error.message); } else { console.error("Cohere API call failed", ctx.error); } }, }); },};
The constructor new CohereClientV2({}) auto-detects COHERE_API_KEY from the environment — no explicit token parameter needed.
Step 10: Add the e2b sandbox for safe Python execution
The sandbox service runs generated Python code inside an E2B cloud sandbox. It limits concurrent executions to 3 via p-limit, enforces a 30-second timeout per run, and always kills the sandbox in a finally block to prevent resource leaks.
Create src/services/sandbox-service.ts:
ts
import Sandbox from "e2b";import pLimit from "p-limit";import type { SandboxResult } from "../lib/types.js";const sandboxLimit = pLimit(3);export const sandboxService = { executePython(pythonCode: string): Promise<SandboxResult> { return sandboxLimit(async () => { const sandbox = await Sandbox.create(); const ac = new AbortController(); const timeout = setTimeout(() => { ac.abort(); }, 30_000); try { const result = await sandbox.commands.run("python3 -c " + JSON.stringify(pythonCode), { signal: ac.signal, }); return { stdout: result.stdout, stderr: result.stderr, exitCode: result.exitCode, }; } catch (e) { if (e instanceof Error && e.name === "AbortError") { throw new Error("Sandbox execution timed out after 30 seconds"); } if (e instanceof Error && e.message.includes("E2B")) { throw new Error("E2B authentication failed. Check your API key."); } throw e; } finally { clearTimeout(timeout); await sandbox.kill(); } }); },};
Step 11: Build the Mixpanel and Langfuse clients
The Mixpanel client fetches recent events from Mixpanel’s REST API using Bearer token auth. It validates the response envelope, retries on transient failures via p-retry, and aborts retry immediately on 401/403.
Create src/lib/mixpanel-client.ts:
ts
import { type MixpanelEvent, MixpanelError } from "../lib/types.js";import { MIXPANEL_BASE_URL } from "../lib/constants.js";import pRetry from "p-retry";export async function fetchEvents(options?: { fromDate?: string; toDate?: string; event?: string; limit?: number;}): Promise<MixpanelEvent[]> { const projectId = process.env.MIXPANEL_PROJECT_ID; const token = process.env.MIXPANEL_OAUTH_TOKEN ?? ""; const params = new URLSearchParams(); if (projectId) params.set("project_id", projectId); if (options?.fromDate) params.set("from_date", options.fromDate); if (options?.toDate) params.set("to_date", options.toDate); if (options?.event) params.set("event", options.event); if (options?.limit !== undefined) params.set("limit", String(options.limit)); const performFetch = async (): Promise<MixpanelEvent[]> => { const response = await fetch( `${MIXPANEL_BASE_URL}/events?${params.toString()}`, { headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json", }, }, ); if (!response.ok) { const errorBody: unknown = await response.json().catch(() => null); throw new MixpanelError( `Mixpanel API returned ${String(response.status)}`, response.status, errorBody, ); } const parsed: unknown = await response.json(); const resp = parsed as { data: unknown }; if ( typeof parsed !== "object" || parsed === null || !("data" in resp) || !Array.isArray(resp.data) ) { throw new MixpanelError("Invalid response shape", 200, parsed); } return resp.data as MixpanelEvent[]; }; return pRetry(performFetch, { retries: 3, onFailedAttempt: (ctx) => { if ( ctx.error instanceof MixpanelError && [401, 403].includes(ctx.error.status) ) { throw ctx.error; } }, });}
Now create the Langfuse client for observability tracing. It wraps Langfuse’s trace() and span() methods with a clean { end() } interface so the orchestrator can close spans with a single call.
Expected output:pnpm typecheck --noEmit reports zero errors across all files so far.
Step 12: Write the analysis orchestrator
This is the heart of the application. The analyzerService.analyze() function orchestrates every step of the pipeline in order:
Validate input with AnalyzeRequestSchema
Define the user’s budget (auto-creates if new)
Create a Langfuse trace
Check budget — if exhausted, return a capped response immediately
Check cache — if a semantically similar query was answered, return it
Fetch Mixpanel events
Generate Python code via Cohere
Execute the code in the e2b sandbox
Repair the sandbox output with structured-repair-core
Cache the result for future queries
Record the spend against the user’s budget
Record a cost telemetry span
Return the full AnalyzeResponse
Create src/services/analyzer-service.ts:
ts
import { budgetService } from "./budget-service.js";import { cacheService, buildPromptHash } from "./cache-service.js";import { cohereService } from "./cohere-service.js";import { sandboxService } from "./sandbox-service.js";import { repairService } from "./repair-service.js";import { telemetryService } from "./telemetry-service.js";import { fetchEvents } from "../lib/mixpanel-client.js";import { createTrace, createSpan } from "../lib/langfuse-client.js";import { AnalyzeRequestSchema, type AnalyzeRequest, type AnalyzeResponse,} from "../lib/types.js";
The key pattern: cache store is fire-and-forget (not awaited), and every span is closed in a finally block so tracing stays accurate even when a step fails.
Step 13: Expose the API route
The route handler at app/api/analyze/route.ts accepts POST requests with JSON { query, userId }, validates them, calls the analyzer, and returns the result. It uses NextRequest and NextResponse.json() — never bare Request/Response.
Expected output: Navigating to http://localhost:3000/api/analyze (after pnpm dev) shows a 405 — the route only accepts POST/OPTIONS.
Step 14: Build the dashboard UI
Replace the default app/page.tsx with a dashboard that lets users type questions, submit them, and see results with cost metadata. This is a client component using "use client".
For the API route handler, create tests/analyze-route.test.ts. The test mocks both the analyzer service and NextResponse so the route handler’s NextResponse.json() calls resolve correctly in the test environment:
The full test suite includes additional coverage files in tests/ for every service module, types, constants, and the route handler. Run the full suite:
terminal
pnpm vitest run --coverage --reporter=json --outputFile=vitest-report.json
Expected output: All tests pass with zero failures, and the coverage report shows metrics meeting the 90% thresholds configured in vitest.config.ts.
Step 16: Run the quality gate
Before deploying, run all three checks to confirm the project is clean:
terminal
pnpm typecheckpnpm lintpnpm vitest run --coverage --reporter=json --outputFile=vitest-report.json
Expected output:
pnpm typecheck exits 0 with no TypeScript errors
pnpm lint exits 0 with no ESLint violations
pnpm vitest run shows numFailedTests: 0 and coverage lines/branches/functions/statements all >= 90%
To see the app in action, start the dev server:
terminal
pnpm dev
Then open http://localhost:3000 in your browser. Type a question like “What are my top 5 events?” and click Analyze. The full pipeline runs: budget check, cache lookup, Mixpanel data fetch, Cohere code generation, sandbox execution, output repair, and cost recording.
Next steps
Add a query history view — persist past analyses to a database so users can revisit previous results
Extend budget policies — add auto-downgrade rules that switch to a cheaper Cohere model when the soft cap is breached
Upgrade to persistent cache storage — swap InMemoryAdapter for Redis or DynamoDB adapters from @reaatech/llm-cache-adapters-*
import
{ COHERE_MODEL }
from
"../lib/constants.js"
;
import { generateId } from "@reaatech/llm-cost-telemetry";
