Implement audit logging for OpenRouter compliance. Use when meeting regulatory requirements or security audits. Trigger with phrases like 'openrouter audit', 'openrouter compliance log', 'openrouter security log', 'audit trail'.
npxskills add jeremylongshore/claude-code-plugins-plus-skills--skill openrouter-audit-loggingLoading…
Every OpenRouter API call returns a generation ID and metadata that enables comprehensive audit logging. The generation endpoint (GET /api/v1/generation?id=) provides exact cost, token counts, provider used, and latency -- data that the initial response doesn't always include. This skill covers structured logging, cost tracking, PII redaction, and compliance-ready audit trails.
sk-or-v1-...) exported as OPENROUTER_API_KEY — see the openrouter-install-auth skill for setuprequests (pip install openai requests) — the audit wrapper fetches exact cost from the generation endpoint with requestssqlite3 module writes the audit table; the CLI runs the Audit Queries against sqlite3openrouter_audit.dbaudited_completion() from Core: Generation Metadata Retrieval — it hashes the prompt (SHA-256), times the call, and fetches exact cost via GET /api/v1/generation?id= after each request.init_audit_db() per Structured Log Storage, then persist every AuditEntry with write_audit() — INSERT OR IGNORE keeps retries from double-writing a generation_id.redact_pii() from PII Redaction Before Logging over any prompt preview before it touches a log: emails, phones, SSNs, card numbers, and sk-or-v1- keys are scrubbed, and raw prompts are never stored (hashes only).user_id.total_cost comes back missing, apply the fixes in Error Handling (fetch within 30 minutes; retry after 1-2 seconds).import os, json, time, hashlib, logging
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from typing import Optional
import requests
from openai import OpenAI
log = logging.getLogger("openrouter.audit")
@dataclass
class AuditEntry:
timestamp: str
generation_id: str
model_requested: str
model_used: str # Actual model served (may differ with fallbacks)
prompt_tokens: int
completion_tokens: int
total_cost: float
latency_ms: float
status: str # "success" | "error" | "timeout"
user_id: str
prompt_hash: str # SHA-256 of prompt (not raw content)
error_code: Optional[str] = None
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
default_headers={
"HTTP-Referer": "https://my-app.com",
"X-Title": "my-app",
},
)
def audited_completion(
messages: list[dict],
model: str = "anthropic/claude-3.5-sonnet",
user_id: str = "system",
**kwargs,
) -> tuple:
"""Make a completion request with full audit logging."""
prompt_text = json.dumps(messages)
prompt_hash = hashlib.sha256(prompt_text.encode()).hexdigest()[:16]
start = time.monotonic()
status = "success"
error_code = None
try:
response = client.chat.completions.create(
model=model, messages=messages, **kwargs
)
except Exception as e:
status = "error"
error_code = type(e).__name__
raise
finally:
latency = (time.monotonic() - start) * 1000
# Fetch exact cost from generation endpoint
gen_data = {}
try:
gen = requests.get(
f"https://openrouter.ai/api/v1/generation?id={response.id}",
headers={"Authorization": f"Bearer {os.environ['OPENROUTER_API_KEY']}"},
timeout=5,
).json()
gen_data = gen.get("data", {})
except Exception:
log.warning(f"Failed to fetch generation metadata for {response.id}")
entry = AuditEntry(
timestamp=datetime.now(timezone.utc).isoformat(),
generation_id=response.id,
model_requested=model,
model_used=response.model,
prompt_tokens=response.usage.prompt_tokens,
completion_tokens=response.usage.completion_tokens,
total_cost=float(gen_data.get("total_cost", 0)),
latency_ms=round(latency, 1),
status=status,
user_id=user_id,
prompt_hash=prompt_hash,
error_code=error_code,
)
log.info(json.dumps(asdict(entry)))
return response, entry
import sqlite3
def init_audit_db(db_path: str = "openrouter_audit.db"):
"""Create append-only audit table."""
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
generation_id TEXT UNIQUE NOT NULL,
model_requested TEXT NOT NULL,
model_used TEXT NOT NULL,
prompt_tokens INTEGER,
completion_tokens INTEGER,
total_cost REAL,
latency_ms REAL,
status TEXT NOT NULL,
user_id TEXT,
prompt_hash TEXT,
error_code TEXT
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_audit_user ON audit_log(user_id)")
conn.commit()
return conn
def write_audit(conn: sqlite3.Connection, entry: AuditEntry):
"""Write audit entry to SQLite (append-only)."""
conn.execute(
"""INSERT OR IGNORE INTO audit_log
(timestamp, generation_id, model_requested, model_used,
prompt_tokens, completion_tokens, total_cost, latency_ms,
status, user_id, prompt_hash, error_code)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(entry.timestamp, entry.generation_id, entry.model_requested,
entry.model_used, entry.prompt_tokens, entry.completion_tokens,
entry.total_cost, entry.latency_ms, entry.status, entry.user_id,
entry.prompt_hash, entry.error_code),
)
conn.commit()
import re
PII_PATTERNS = [
(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]'),
(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]'),
(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN]'),
(r'\bsk-or-v1-[a-zA-Z0-9]+\b', '[API_KEY]'),
(r'\b(?:\d{4}[- ]?){3}\d{4}\b', '[CARD]'),
]
def redact_pii(text: str) -> str:
"""Scrub PII from text before logging."""
for pattern, replacement in PII_PATTERNS:
text = re.sub(pattern, replacement, text)
return text
-- Daily cost by model
SELECT date(timestamp) as day, model_used,
COUNT(*) as requests, SUM(total_cost) as cost
FROM audit_log GROUP BY day, model_used ORDER BY day DESC, cost DESC;
-- Error rate by model (last 24h)
SELECT model_requested, COUNT(*) as total,
SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as errors,
ROUND(100.0 * SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) / COUNT(*), 1) as error_pct
FROM audit_log WHERE timestamp > datetime('now', '-1 day')
GROUP BY model_requested;
-- Top spenders
SELECT user_id, COUNT(*) as requests, SUM(total_cost) as total_cost
FROM audit_log GROUP BY user_id ORDER BY total_cost DESC LIMIT 10;
AuditEntry per request: timestamp, generation_id, model_requested vs model_used, prompt/completion token counts, exact total_cost, latency_ms, status, user_id, and a 16-char prompt_hashaudit_log table (openrouter_audit.db) indexed on timestamp and user_id, protected against duplicate writes by INSERT OR IGNOREuser_idWrap a call with the JSONL AuditLogger variant from the references and read back the entry it appends:
result = audited_completion("user-123", "What is machine learning?")
# [Audit] user=user-123 tokens=97 latency=450ms
The corresponding line in audit.jsonl:
{"timestamp": "2026-03-17T10:00:00Z", "user_id": "user-123",
"model": "openai/gpt-3.5-turbo", "prompt_hash": "a1b2c3d4e5f6g7h8",
"prompt_preview": "What is machine learning?", "prompt_tokens": 12,
"completion_tokens": 85, "total_tokens": 97, "status": "success",
"latency_ms": 450, "generation_id": "gen-abc123"}
More worked examples: references/examples.md.
| Error | Cause | Fix |
|---|---|---|
| Generation endpoint 404 | Generation ID not found or too old | Fetch within 30 minutes of request |
| Duplicate generation_id | Retry wrote same request twice | Use INSERT OR IGNORE |
Missing total_cost | Generation still processing | Retry fetch after 1-2 seconds |
| Auth 401 on generation fetch | Wrong API key for that generation | Use same key that made the request |
user_id field to enable per-user cost attribution and abuse detectiongeneration_id for fast correlation with OpenRouter dashboard
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Set up and use 1Password CLI (op). Use when installing the CLI, enabling desktop app integration, signing in (single or multi-account), or reading/injecting/running secrets via op.
CLI to manage emails via IMAP/SMTP. Use `himalaya` to list, read, write, reply, forward, search, and organize emails from the terminal. Supports multiple accounts and message composition with MML (MIME Meta Language).
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Set up and use 1Password CLI (op). Use when installing the CLI, enabling desktop app integration, signing in (single or multi-account), or reading/injecting/running secrets via op.
CLI to manage emails via IMAP/SMTP. Use `himalaya` to list, read, write, reply, forward, search, and organize emails from the terminal. Supports multiple accounts and message composition with MML (MIME Meta Language).