Production Patterns

Before you go to production

A concise checklist for teams deploying Ravi in cloud, CI, or headless environments. Each item links to the relevant section on this page or in the related guides.

#	Check	Why it matters
1	Token injection configured — set RAVI_ACCESS_TOKEN as an env var or mount ~/.ravi/auth.json read-only	Never hardcode tokens in source; CI/Docker can’t complete a device-code flow at runtime
2	Parallel isolation verified — agents use RAVI_CONFIG_DIR or per-project .ravi/config.json, not ravi identity use	ravi identity use writes shared state — concurrent agents race and corrupt each other’s context
3	Token refresh handled — processes running longer than ~1 hour call ravi auth refresh or POST to /api/auth/token/refresh/ before expiry	Expired tokens produce silent 401 failures; long-running agents need proactive refresh
4	OTP polling loops in place — no bare sleep 5 anywhere in the codebase	Fixed sleeps fail silently when SMS is slow and waste time when it’s fast; use a retry loop with timeout
5	Ephemeral identities cleaned up — task-scoped identities are deleted after use	Identities accumulate indefinitely if never cleaned up; use dev- / test- prefix for non-production names so bulk cleanup is trivial
6	Subprocess errors are handled — ravi CLI calls check return codes	Network errors and token expiry surface as non-zero exit codes; silently returning empty data masks failures
7	Identity names are deterministic — use worker-${task_id} or agent-${env}-${service}, not random UUIDs	You need to look up identities by name later (for config files, logging, cleanup); random names make this hard

---

Production Patterns

The CLI quickstart is a good introduction, but production agent systems call Ravi programmatically — provisioning identities at runtime, isolating parallel workers, polling for OTPs with proper retry logic, and cleaning up after task completion.

This page covers the patterns that matter at scale.

---

Programmatic identity creation

Python (subprocess)

The fastest path from zero to a working identity in Python wraps the CLI directly. This works anywhere the ravi binary is on the PATH:

import subprocess
import json

def create_identity(name: str) -> dict:
    result = subprocess.run(
        ["ravi", "identity", "create", "--name", name, "--json"],
        capture_output=True,
        text=True,
        check=True,
    )
    return json.loads(result.stdout)

def get_email(identity_name: str) -> str:
    result = subprocess.run(
        ["ravi", "get", "email", "--identity", identity_name, "--json"],
        capture_output=True,
        text=True,
        check=True,
    )
    return json.loads(result.stdout)["email"]

def get_phone(identity_name: str) -> str:
    result = subprocess.run(
        ["ravi", "get", "phone", "--identity", identity_name, "--json"],
        capture_output=True,
        text=True,
        check=True,
    )
    return json.loads(result.stdout)["phone"]

# Usage: provision an identity per task
identity = create_identity(f"worker-{task_id}")
email = get_email(f"worker-{task_id}")
phone = get_phone(f"worker-{task_id}")

TypeScript (subprocess)

import { execSync } from "child_process";

function createIdentity(name: string): { inbox: string; phone: string } {
  const output = execSync(
    `ravi identity create --name "${name}" --json`,
    { encoding: "utf8" }
  );
  return JSON.parse(output);
}

function getEmail(identityName: string): string {
  const output = execSync(
    `ravi get email --identity "${identityName}" --json`,
    { encoding: "utf8" }
  );
  return JSON.parse(output).email;
}

// Usage
const identity = createIdentity(`worker-${taskId}`);
const { inbox: email, phone } = identity;

Note: The --identity flag passes the identity name directly to the command without modifying global state. This is the safe approach for programmatic use — avoid ravi identity use in automated scripts (see Parallel Agent Isolation below).

---

Parallel agent isolation

ravi identity use <name> writes to ~/.ravi/config.json — a single shared file. If two agent processes call ravi identity use concurrently, they race and corrupt each other’s context.

Do not use ravi identity use in parallel agent systems.

Instead, use one of two approaches:

Option 1: Per-process config directory

Give each agent process its own config directory via the RAVI_CONFIG_DIR environment variable:

import os
import subprocess
import tempfile
import json

def run_agent_with_isolated_identity(identity_name: str, task: callable):
    # Each agent gets its own scratch config dir
    with tempfile.TemporaryDirectory() as config_dir:
        env = {**os.environ, "RAVI_CONFIG_DIR": config_dir}

        # Write identity config into the isolated dir
        config = {"identity": identity_name}
        with open(f"{config_dir}/config.json", "w") as f:
            json.dump(config, f)

        # All ravi commands in this env use identity_name
        result = subprocess.run(
            ["ravi", "inbox", "email", "--unread", "--json"],
            capture_output=True, text=True, env=env
        )
        return json.loads(result.stdout)

import { execSync } from "child_process";
import { mkdtempSync, writeFileSync, rmSync } from "fs";
import { tmpdir } from "os";
import path from "path";

function withIsolatedIdentity<T>(
  identityName: string,
  fn: (env: NodeJS.ProcessEnv) => T
): T {
  const configDir = mkdtempSync(path.join(tmpdir(), "ravi-"));
  try {
    writeFileSync(
      path.join(configDir, "config.json"),
      JSON.stringify({ identity: identityName })
    );
    const env = { ...process.env, RAVI_CONFIG_DIR: configDir };
    return fn(env);
  } finally {
    rmSync(configDir, { recursive: true, force: true });
  }
}

// Usage
const emails = withIsolatedIdentity("worker-42", (env) => {
  const output = execSync("ravi inbox email --unread --json", {
    env,
    encoding: "utf8",
  });
  return JSON.parse(output);
});

Option 2: Per-project .ravi/config.json

For agents that each operate in their own project directory, place a .ravi/config.json in each project root. Ravi walks up the directory tree and uses the first config it finds, so each project automatically uses its own identity:

# Set up a workspace for worker-42
mkdir -p ~/workspaces/worker-42/.ravi
echo '{"identity": "worker-42"}' > ~/workspaces/worker-42/.ravi/config.json

# All ravi commands run from this directory use worker-42
cd ~/workspaces/worker-42
ravi inbox email --unread --json

See Multi-Agent Setup for a full walkthrough.

---

Retry polling patterns

Never use sleep 5 as a polling strategy in production. Fixed-delay sleeps fail silently when services are slow, and waste time when they’re fast. Use a retry loop with a timeout.

Polling for an OTP (Python)

import subprocess
import json
import time

def poll_for_otp(identity_name: str, max_attempts: int = 30, delay: float = 2.0) -> str | None:
    """Poll the SMS inbox for an OTP code. Returns the code or None on timeout."""
    env_with_identity = {"RAVI_CONFIG_DIR": f"/tmp/ravi-{identity_name}"}  # isolated config

    for attempt in range(max_attempts):
        result = subprocess.run(
            ["ravi", "inbox", "sms", "--unread", "--json"],
            capture_output=True, text=True
        )
        messages = json.loads(result.stdout)
        if messages:
            body = messages[0].get("preview", "")
            # Extract 4-8 digit OTP
            import re
            match = re.search(r"\b\d{4,8}\b", body)
            if match:
                return match.group(0)
        time.sleep(delay)

    return None  # timeout

otp = poll_for_otp("worker-42")
if otp is None:
    raise TimeoutError("OTP not received within 60 seconds")

Polling for an email verification link (TypeScript)

import { execSync } from "child_process";

async function pollForVerificationLink(
  env: NodeJS.ProcessEnv,
  maxAttempts = 20,
  delayMs = 3000
): Promise<string | null> {
  for (let i = 0; i < maxAttempts; i++) {
    const output = execSync("ravi inbox email --unread --json", {
      env,
      encoding: "utf8",
    });
    const threads = JSON.parse(output) as Array<{ preview: string }>;

    for (const thread of threads) {
      const match = thread.preview.match(/https?:\/\/\S+/);
      if (match) return match[0];
    }

    await new Promise((resolve) => setTimeout(resolve, delayMs));
  }

  return null; // timeout after ~60s
}

---

Identity lifecycle management

For task-scoped agents (spin up, do work, tear down), manage identities explicitly:

import subprocess
import json
from contextlib import contextmanager

@contextmanager
def ephemeral_identity(name: str):
    """Create an identity for the duration of a task, then clean up."""
    # Provision
    result = subprocess.run(
        ["ravi", "identity", "create", "--name", name, "--json"],
        capture_output=True, text=True, check=True
    )
    identity = json.loads(result.stdout)
    try:
        yield identity
    finally:
        # Archive or delete when done
        subprocess.run(
            ["ravi", "identity", "delete", name],
            capture_output=True
        )

# Usage
with ephemeral_identity(f"signup-worker-{job_id}") as identity:
    email = identity["inbox"]
    # ... do the work ...
    # identity is deleted on context exit

Naming convention for cleanup: Prefix development and test identities with dev- or test- so you can identify and bulk-delete them:

# List all test identities
ravi identity list --json | jq '.[] | select(.name | startswith("dev-"))'

# Delete a specific identity
ravi identity delete dev-worker-42

---

Token refresh in long-running processes

Ravi auth tokens expire. Agent processes that run for hours need to refresh their token before it expires. Check the token age and refresh proactively:

import subprocess
import json
import os
from datetime import datetime, timezone

def ensure_authenticated() -> bool:
    """Re-authenticate if the token is close to expiry. Returns True if ready."""
    auth_path = os.path.expanduser("~/.ravi/auth.json")
    if not os.path.exists(auth_path):
        return False

    with open(auth_path) as f:
        auth = json.load(f)

    expires_at = datetime.fromisoformat(auth.get("expires_at", "2000-01-01T00:00:00+00:00"))
    now = datetime.now(timezone.utc)

    # Refresh if less than 30 minutes remaining
    if (expires_at - now).total_seconds() < 1800:
        result = subprocess.run(
            ["ravi", "auth", "refresh"],
            capture_output=True, text=True
        )
        return result.returncode == 0

    return True

# In a long-running loop
import time
while True:
    ensure_authenticated()
    # ... agent work ...
    time.sleep(300)  # Check every 5 minutes

Headless note: If running in CI or Docker, complete ravi auth login once interactively on a machine with a browser, then copy ~/.ravi/auth.json to the target environment. The token will refresh automatically from that point. See Authentication for the full headless setup flow.

---

What to read next

• Multi-Agent Setup — per-project identity isolation with .ravi/config.json
• Human Approval via Email — block agent execution pending a human yes/no
• Troubleshooting — common errors: race conditions, OTP timeouts, token expiry