DriftRail SDK Documentation

pip install driftrail

# For async support
pip install driftrail[async]

from driftrail import DriftRail

client = DriftRail(
    api_key="dr_live_...",
    app_id="my-app"
)

# Log an LLM interaction
response = client.ingest(
    model="gpt-4o",
    provider="openai",
    input={"prompt": "What is the capital of France?"},
    output={"text": "The capital of France is Paris."},
    metadata={
        "latency_ms": 420,
        "tokens_in": 25,
        "tokens_out": 12,
        "temperature": 0.7
    }
)

print(f"Event ID: {response.event_id}")

from driftrail import DriftRail

client = DriftRail(api_key="dr_live_...", app_id="my-app")

# Get response from your LLM
llm_response = your_llm_call(user_prompt)

# Guard it before returning to user
result = client.guard(
    output=llm_response,
    input=user_prompt,
    mode="strict"  # or "permissive"
)

if result.allowed:
    return result.output  # May be redacted if PII was found
else:
    print(f"Blocked: {[t.reason for t in result.triggered]}")
    return "Sorry, I can't help with that."

import asyncio
from driftrail import DriftRailAsync

async def main():
    async with DriftRailAsync(api_key="dr_live_...", app_id="my-app") as client:
        response = await client.ingest(
            model="claude-sonnet-4",
            provider="anthropic",
            input={"prompt": "Hello!"},
            output={"text": "Hi there!"}
        )
        print(f"Event ID: {response.event_id}")

asyncio.run(main())

# Won't block your main thread
client.ingest_async(
    model="gpt-4o",
    provider="openai",
    input={"prompt": "..."},
    output={"text": "..."}
)

client.ingest(
    model="gpt-4o",
    provider="openai",
    input={
        "prompt": "What's our refund policy?",
        "retrieved_sources": [
            {"id": "doc-123", "content": "Refunds available within 30 days..."},
            {"id": "doc-456", "content": "Contact support for refund requests..."}
        ]
    },
    output={"text": "According to our policy, refunds are available within 30 days..."}
)

client = DriftRail(
    api_key="dr_live_...",
    app_id="my-app",
    fail_open=True,         # Default: errors logged but don't raise
    guard_mode="fail_open"  # Default: if guard API unavailable, allow content
)

# Even if DriftRail is down, this won't raise
response = client.ingest(...)
if not response.success:
    print(f"Warning: {response.error}")

from driftrail import DriftRailEnterprise

client = DriftRailEnterprise(api_key="dr_live_...", app_id="my-app")

# === Incident Management ===
stats = client.get_incident_stats()
incidents = client.list_incidents(status=["open"], severity=["critical"])
client.create_incident(title="High risk spike", severity="high", incident_type="risk_spike")
client.update_incident_status(incident_id, status="investigating")

# === Compliance & Reporting ===
compliance = client.get_compliance_status()
score = client.get_compliance_score()
reports = client.get_compliance_reports()
client.generate_compliance_report(framework="hipaa", format="pdf", include_ai_analysis=True)
client.create_custom_framework(name="Internal Policy", controls=[...])

# === Executive Dashboard ===
metrics = client.get_executive_metrics(period="7d")
targets = client.get_kpi_targets()
client.update_kpi_targets({"max_high_risk_percent": 5.0})
client.export_executive_metrics(period="30d", format="xlsx")

# === Model Analytics ===
summary = client.get_model_analytics_summary()
logs = client.get_historical_logs(model="gpt-4o", limit=100)
switches = client.get_model_switches()
client.record_model_switch(app_id="my-app", new_model="claude-4", new_provider="anthropic")
benchmarks = client.get_model_benchmarks()
client.calculate_model_benchmark(model="gpt-4o", days=7)

# === Drift Detection V3 ===
drift_score = client.get_drift_score()
heatmap = client.get_drift_heatmap(days=30)
thresholds = client.get_drift_thresholds()
client.update_drift_thresholds({"risk_score": {"warning": 15, "critical": 25}})
predictions = client.get_drift_predictions()
correlations = client.get_drift_deployment_correlations()
client.record_deployment(app_id="my-app", version="v2.1.0", deployment_type="release")
seasonality = client.get_seasonality_patterns()
distribution = client.get_distribution_analysis()
statistics = client.get_baseline_statistics()

# === Notification Channels ===
channels = client.get_notification_channels()
client.create_notification_channel(
    channel_type="slack",
    name="Alerts Channel",
    config={"webhook_url": "https://hooks.slack.com/..."},
    severity_filter=["critical", "warning"]
)

# === Drift Segments ===
segments = client.get_drift_segments()
client.create_drift_segment(name="Production GPT-4", filter_criteria={"model": "gpt-4o"})

# === Distributed Tracing ===
trace = client.start_trace(app_id="my-app", name="chat-completion", user_id="user-123")
span = client.start_span(trace_id=trace["trace_id"], name="llm-call", span_type="llm", model="gpt-4o")
client.end_span(span["span_id"], status="completed", tokens_in=100, tokens_out=50)
client.end_trace(trace["trace_id"])
traces = client.list_traces(status="completed", limit=50)

# === Prompt Management ===
prompt = client.create_prompt(name="Customer Support", content="You are a helpful assistant...")
version = client.create_prompt_version(prompt["prompt_id"], content="Updated prompt...", commit_message="v2")
client.deploy_prompt_version(version["version_id"], environment="production")
deployed = client.get_deployed_prompt(prompt["prompt_id"], environment="production")

# === Evaluation Framework ===
dataset = client.create_dataset(name="QA Test Set", schema_type="qa")
client.add_dataset_items(dataset["dataset_id"], items=[{"input": {...}, "expected_output": {...}}])
run = client.create_eval_run(dataset["dataset_id"], evaluators=[{"type": "exact_match"}])
results = client.get_eval_run(run["run_id"])

# === Semantic Caching ===
settings = client.get_cache_settings()
client.update_cache_settings({"is_enabled": True, "similarity_threshold": 0.95})
stats = client.get_cache_stats()
lookup = client.cache_lookup(input="What is the capital of France?")
client.cache_store(input="...", output="...", model="gpt-4o")

# === Agent Simulations ===
sim = client.create_simulation(name="Support Bot Test", scenario="User asks for refund")
run = client.run_simulation(sim["simulation_id"])
client.add_simulation_turn(run["run_id"], turn_number=1, role="user", content="I want a refund")
stats = client.get_simulation_stats()

# === Integrations ===
integrations = client.get_integrations()
client.create_integration(type="slack", webhook_url="https://...", events=["high_risk", "incident"])
client.test_integration(webhook_url="https://...", type="slack")

# === Benchmarks ===
industries = client.get_industries()
report = client.get_benchmark_report(industry="healthcare")
client.set_tenant_industry(industry="fintech")

# === Guardrails ===
guardrails = client.get_guardrails()
client.create_guardrail(name="PII Blocker", rule_type="pii", action="block")
stats = client.get_guardrail_stats()

# === Retention Policies ===
policies = client.get_retention_policies()
client.create_retention_policy(name="90 Day Retention", data_type="events", retention_days=90)

# === Audit Logs ===
logs = client.get_audit_logs(action="event.ingested", limit=100)

# === Events & Classifications ===
events = client.get_events(min_risk_score=70, limit=50)
event = client.get_event(event_id)
live = client.get_live_events()
classifications = client.get_classifications(min_score=0.8)

# === Custom Detections ===
detections = client.get_custom_detections()
client.create_custom_detection(name="Competitor Mention", detection_type="keyword", config={"keywords": [...]})

# === Webhooks ===
webhooks = client.get_webhooks()
client.create_webhook(url="https://...", events=["high_risk", "drift_alert"])

# === Stats ===
stats = client.get_stats(period="7d")

npm install @drift_rail/sdk
# or
yarn add @drift_rail/sdk
# or
pnpm add @drift_rail/sdk

// Won't block your code
client.ingestAsync({
  model: 'gpt-4',
  provider: 'openai',
  input: { prompt: '...' },
  output: { text: '...' }
});

// Convenience method for chat completions
await client.logChatCompletion({
  model: 'gpt-4',
  messages: [
    { role: 'system', content: 'You are a helpful assistant.' },
    { role: 'user', content: 'Hello!' }
  ],
  response: 'Hi there! How can I help you today?',
  latencyMs: 420,
  tokensIn: 25,
  tokensOut: 12
});

import type { 
  IngestParams, 
  IngestResponse, 
  Provider,
  InputPayload,
  OutputPayload 
} from '@drift_rail/sdk';

import { initAIObservability, logInference } from '@drift_rail/sdk';

initAIObservability({
  apiKey: 'dr_live_...',
  appId: 'my-app',
  environment: 'prod',
  batchSize: 10,          // Send events in batches
  flushIntervalMs: 5000,  // Or every 5 seconds
  debug: false
});

logInference({
  model: 'gpt-4',
  provider: 'openai',
  input: { prompt: 'User query here' },
  output: { text: 'AI response here' },
  metadata: {
    latencyMs: 350,
    temperature: 0.7
  }
});

client.ingest(
    model="gpt-5",
    provider="openai",
    input={
        "prompt": "What does our refund policy say?",
        "retrievedSources": [  # camelCase in API payload
            {
                "id": "doc-123",
                "content": "Refunds available within 30 days...",
                "type": "document",
                "url": "https://docs.example.com/refunds",
                "metadata": {"score": 0.95, "chunk_index": 2}
            },
            {
                "id": "doc-456", 
                "content": "Contact support for refund requests..."
            }
        ]
    },
    output={"text": "According to our policy, refunds are available within 30 days..."}
)

# Python SDK also accepts snake_case for convenience:
input={"retrieved_sources": [...]}

await client.ingest({
  model: 'gpt-5',
  provider: 'openai',
  input: {
    prompt: 'What does our refund policy say?',
    retrievedSources: [
      { id: 'doc-123', content: 'Refunds available within 30 days...' },
      { id: 'doc-456', content: 'Contact support for refund requests...' }
    ]
  },
  output: { text: 'According to our policy, refunds are available within 30 days...' }
});

from driftrail import DriftRail

client = DriftRail(api_key="dr_live_...", app_id="my-app")

# Get response from your LLM
llm_response = your_llm_call(user_prompt)

# Guard it BEFORE returning to user
result = client.guard(
    output=llm_response,
    input=user_prompt,
    mode="strict"
)

if result.allowed:
    return result.output  # May be redacted
else:
    print(f"Blocked: {[t.reason for t in result.triggered]}")
    return "Sorry, I can't help with that."

from driftrail import DriftRail, GuardBlockedError

# Fail-open (default): If DriftRail is unavailable, content is allowed
client = DriftRail(api_key="...", app_id="...", guard_mode="fail_open")

# Fail-closed: If DriftRail is unavailable, raises exception
client = DriftRail(api_key="...", app_id="...", guard_mode="fail_closed")

try:
    result = client.guard(output=llm_response)
except GuardBlockedError as e:
    # Content was blocked
    print(f"Blocked: {e.result.triggered}")

result = client.guard(output="...")

result.allowed      # bool - True if content can be shown to user
result.action       # "allow" | "block" | "redact" | "warn"
result.output       # Original or redacted content
result.triggered    # List of triggered guardrails/classifications
result.classification  # AI classification details (risk_score, pii, toxicity, etc.)
result.latency_ms   # Processing time
result.fallback     # True if classification failed (fail-open)

from fastapi import FastAPI, HTTPException
from driftrail import DriftRail

app = FastAPI()
driftrail = DriftRail(api_key="...", app_id="my-api")

@app.post("/api/chat")
async def chat(prompt: str):
    # Get LLM response
    llm_response = await call_llm(prompt)
    
    # Guard before returning
    guard = driftrail.guard(output=llm_response, input=prompt)
    
    if not guard.allowed:
        raise HTTPException(
            status_code=400,
            detail={"error": "Content blocked", "reasons": [t.reason for t in guard.triggered]}
        )
    
    return {"response": guard.output}

import { DriftRail } from '@drift_rail/sdk';

const client = new DriftRail({ apiKey: 'dr_live_...', appId: 'my-app' });

// Get response from your LLM
const llmResponse = await yourLLMCall(userPrompt);

// Guard it BEFORE returning to user
const result = await client.guard({
  output: llmResponse,
  input: userPrompt,
  mode: 'strict'
});

if (result.allowed) {
  return result.output; // May be redacted
} else {
  console.log('Blocked:', result.triggered.map(t => t.reason));
  return "Sorry, I can't help with that.";
}

import express from 'express';
import { DriftRail } from '@drift_rail/sdk';

const app = express();
const driftrail = new DriftRail({ apiKey: '...', appId: 'my-app' });

app.post('/api/chat', async (req, res) => {
  const { prompt } = req.body;
  
  // Get LLM response
  const llmResponse = await callLLM(prompt);
  
  // Guard before returning
  const guard = await driftrail.guard({
    output: llmResponse,
    input: prompt
  });
  
  if (!guard.allowed) {
    return res.status(400).json({ 
      error: 'Content blocked',
      reasons: guard.triggered.map(t => t.reason)
    });
  }
  
  res.json({ response: guard.output });
});

curl -X POST https://api.driftrail.com/api/guard \
  -H "Authorization: Bearer dr_live_..." \
  -H "X-App-Id: my-app" \
  -H "Content-Type: application/json" \
  -d '{
    "output": "The LLM response to check",
    "input": "Optional user prompt for context",
    "mode": "strict",
    "timeout_ms": 100
  }'

{
  "allowed": true,
  "action": "redact",
  "output": "Contact me at j***@***.com for details",
  "triggered": [
    {
      "type": "classification",
      "name": "PII Redaction",
      "reason": "Redacted email"
    }
  ],
  "classification": {
    "risk_score": 25,
    "pii": { "detected": true, "types": ["email"] },
    "toxicity": { "detected": false, "severity": "none" },
    "prompt_injection": { "detected": false, "risk": "none" }
  },
  "latency_ms": 42,
  "fallback": false
}

POST /api/ingest HTTP/1.1
Content-Type: application/json
Authorization: Bearer dr_live_...
X-App-Id: my-application

{
  "model": "gpt-5",
  "provider": "openai",
  "input": {
    "prompt": "What is the capital of France?",
    "messages": [
      { "role": "system", "content": "You are a helpful assistant." },
      { "role": "user", "content": "What is the capital of France?" }
    ],
    "retrievedSources": [
      { "id": "doc-123", "content": "France is a country..." }
    ]
  },
  "output": {
    "text": "The capital of France is Paris.",
    "toolCalls": []
  },
  "metadata": {
    "latencyMs": 420,
    "tokensIn": 25,
    "tokensOut": 12,
    "temperature": 0.7
  }
}

{
  "success": true,
  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "job_id": "job_a1b2c3d4..."
}

curl -X GET "https://api.driftrail.com/api/events?app_id=my-app&limit=50" \
  -H "Authorization: Bearer dr_live_..."

{
  "events": [
    {
      "event_id": "550e8400-e29b-41d4-a716-446655440000",
      "timestamp": "2026-01-15T10:30:00Z",
      "model": "gpt-5",
      "provider": "openai",
      "app_id": "my-app",
      "environment": "prod",
      "latency_ms": 420,
      "tokens_in": 25,
      "tokens_out": 12
    }
  ],
  "total": 1250,
  "limit": 50,
  "offset": 0
}

curl -X GET "https://api.driftrail.com/api/classifications/distribution?app_id=my-app" \
  -H "Authorization: Bearer dr_live_..."

{
  "low": 850,
  "medium": 280,
  "high": 95,
  "critical": 15,
  "total": 1240
}

curl -X GET "https://api.driftrail.com/api/classifications/high-risk?threshold=80&limit=20" \
  -H "Authorization: Bearer dr_live_..."

{
  "classifications": [
    {
      "event_id": "550e8400-e29b-41d4-a716-446655440000",
      "risk_score": 92,
      "risk_level": "critical",
      "detected_issues": ["pii_exposure", "hallucination"],
      "classified_at": "2026-01-15T10:30:05Z"
    }
  ]
}

curl -X POST "https://api.driftrail.com/api-playground" \
  -H "Authorization: Bearer dr_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "user", "content": "What is the capital of France?"}
    ],
    "model": "gemini-flash-lite-latest",
    "runDetections": true
  }'

{
  "content": "The capital of France is Paris.",
  "model": "gemini-2.5-flash-lite",
  "provider": "google",
  "detections": [
    {
      "type": "hallucination",
      "risk": "low",
      "confidence": 0.95,
      "details": "Response is factually accurate"
    }
  ],
  "latencyMs": 450,
  "tokensUsed": 25,
  "usage": {
    "current": 16,
    "limit": 500,
    "remaining": 484
  }
}

// Vercel, Netlify, Lambda, Cloudflare Workers
export async function POST(req: Request) {
  const { prompt } = await req.json();
  const response = await callLLM(prompt);
  
  // MUST await - ensures request completes before function terminates
  await client.ingest({
    model: 'gpt-4o',
    provider: 'openai',
    input: { prompt },
    output: { text: response }
  });
  
  return Response.json({ response });
}

// DON'T do this in serverless!
export async function POST(req: Request) {
  const response = await callLLM(prompt);
  
  client.ingestAsync({...}); // Race condition! Function terminates first
  
  return Response.json({ response });
  // HTTP request to DriftRail is killed here
}

// Express, Fastify, Koa, etc. - process stays alive
app.post('/chat', async (req, res) => {
  const response = await getLLMResponse(req.body);
  
  // Safe: Express process continues running after response
  client.ingestAsync({
    model: 'gpt-4o',
    provider: 'openai',
    input: { prompt: req.body.message },
    output: { text: response }
  });
  
  res.json({ response });
});

const client = new DriftRail({
  apiKey: process.env.DRIFTRAIL_API_KEY!,
  appId: 'my-app',
  timeout: 5000,    // 5s max - don't let logging block too long
  failOpen: true    // Default: errors logged but don't crash your app
});

async function handleStream(controller: ReadableStreamDefaultController) {
  let fullResponse = '';
  const startTime = Date.now();
  
  for await (const chunk of llmStream) {
    controller.enqueue(chunk);
    fullResponse += chunk;
  }
  
  // Log BEFORE closing the stream
  await client.ingest({
    model: 'gpt-4o',
    provider: 'openai',
    input: { prompt: userMessage },
    output: { text: fullResponse },
    metadata: { latencyMs: Date.now() - startTime }
  });
  
  controller.close(); // Close AFTER logging
}

// DON'T do this!
controller.close(); // Stream ends here
await client.ingest({...}); // Too late - may not execute

// app/api/chat/route.ts
import { DriftRail } from '@drift_rail/sdk';

const client = new DriftRail({ 
  apiKey: process.env.DRIFTRAIL_KEY!, 
  appId: 'my-app' 
});

export async function POST(req: Request) {
  const { messages } = await req.json();
  const startTime = Date.now();
  
  const stream = new ReadableStream({
    async start(controller) {
      let fullResponse = '';
      
      const llmStream = await openai.chat.completions.create({
        model: 'gpt-4o',
        messages,
        stream: true
      });
      
      for await (const chunk of llmStream) {
        const text = chunk.choices[0]?.delta?.content || '';
        fullResponse += text;
        controller.enqueue(new TextEncoder().encode(text));
      }
      
      // IMPORTANT: await in serverless + log before close
      await client.ingest({
        model: 'gpt-4o',
        provider: 'openai',
        input: { 
          prompt: messages[messages.length - 1].content,
          messages 
        },
        output: { text: fullResponse },
        metadata: { latencyMs: Date.now() - startTime }
      });
      
      controller.close();
    }
  });
  
  return new Response(stream, {
    headers: { 'Content-Type': 'text/event-stream' }
  });
}

{
  "event_type": "alert.created",
  "timestamp": "2025-01-03T10:30:00.000Z",
  "tenant_id": "tenant_abc123",
  "data": {
    "alert_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "alert_type": "risk_increase",
    "severity": "critical",
    "app_id": "my-chatbot",
    "model": "gpt-4",
    "current_value": 85.5,
    "baseline_value": 25.0,
    "deviation_percent": 242.0,
    "details": {
      "affected_events": 150,
      "time_window": "1h"
    }
  }
}

import crypto from 'crypto';

function verifyWebhook(signature: string, body: string, secret: string): boolean {
  const [timestampPart, signaturePart] = signature.split(',');
  const timestamp = timestampPart.split('=')[1];
  const receivedSig = signaturePart.split('=')[1];
  
  const expectedSig = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${body}`)
    .digest('hex');
  
  return crypto.timingSafeEqual(
    Buffer.from(receivedSig, 'hex'),
    Buffer.from(expectedSig, 'hex')
  );
}

# Register a Slack integration
curl -X POST "https://api.driftrail.com/api/integrations" \
  -H "Authorization: Bearer dr_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "type": "slack",
    "webhook_url": "https://hooks.slack.com/services/T.../B.../xxx",
    "channel_name": "#ai-alerts",
    "events": ["alert.created", "alert.critical"]
  }'

# Send a test notification
curl -X POST "https://api.driftrail.com/api/integrations/test" \
  -H "Authorization: Bearer dr_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "webhook_url": "https://hooks.slack.com/services/T.../B.../xxx",
    "type": "slack"
  }'