"""Adapterly logger — drop-in usage logger for OpenAI and Anthropic. Installation ------------ Just drop this file next to your app code and call `instrument()` once at startup, before you make any LLM calls: from adapterly_logger import instrument instrument() # ...your existing OpenAI / Anthropic code, unchanged... That's it. Every completion call now appends a row to `~/.adapterly/logs/YYYY-MM-DD.jsonl`. When you're ready for an analysis, concatenate those files and upload to https://adapterly.ai/tools/cost-analyzer. cat ~/.adapterly/logs/*.jsonl > my-usage.jsonl What gets logged ---------------- { "timestamp": "2026-08-17T14:33:22.104Z", "model": "gpt-4o-2024-08-06", "input_tokens": 1204, "output_tokens": 287, "cached_tokens": 0, "has_images": false, "prompt_snippet": "You are a helpful assistant. Answer the follo..." } Only the first ~400 chars of the first user message is logged as a snippet (for workload classification). Full prompts are never captured. Response content is never captured. Privacy ------- - All logging is local — nothing is sent anywhere until you upload manually. - Prompt snippets are truncated to 400 chars. - Set env var ADAPTERLY_LOGGER_DISABLE=1 to disable at runtime. - Set env var ADAPTERLY_LOGGER_NO_PROMPTS=1 to skip snippets entirely (numeric analysis will still work; qualitative analysis will not). Compatibility ------------- - openai>=1.0 (sync + async client, .chat.completions.create) - anthropic>=0.20 (sync + async client, .messages.create) - Streaming calls are NOT logged (chunks arrive without a final usage record). If instrumentation fails silently for a client, your app is not affected — the logger swallows its own errors to guarantee it never breaks production. """ from __future__ import annotations import json import logging import os import threading from datetime import datetime, timezone from pathlib import Path _LOG_DIR = Path(os.environ.get("ADAPTERLY_LOG_DIR", str(Path.home() / ".adapterly" / "logs"))) _SNIPPET_LEN = 400 _LOCK = threading.Lock() _INSTRUMENTED = False log = logging.getLogger("adapterly_logger") # -------------------------------------------------------------------------- # Public API # -------------------------------------------------------------------------- def instrument() -> None: """Monkey-patch OpenAI + Anthropic clients so every call is logged. Safe to call multiple times — subsequent calls are no-ops. Safe to call even if the SDK is not installed — missing SDKs are skipped. """ global _INSTRUMENTED if _INSTRUMENTED: return if os.environ.get("ADAPTERLY_LOGGER_DISABLE") == "1": log.info("adapterly_logger disabled by env") _INSTRUMENTED = True return _LOG_DIR.mkdir(parents=True, exist_ok=True) _try_instrument_openai() _try_instrument_anthropic() _INSTRUMENTED = True # -------------------------------------------------------------------------- # OpenAI patch # -------------------------------------------------------------------------- def _try_instrument_openai() -> None: try: from openai.resources.chat.completions import Completions, AsyncCompletions except ImportError: log.debug("openai SDK not installed — skipping") return _wrap_sync(Completions, "create", _record_openai) _wrap_async(AsyncCompletions, "create", _record_openai) log.info("adapterly_logger: instrumented openai.chat.completions") def _record_openai(kwargs: dict, response) -> None: """Extract fields from an OpenAI chat.completions call+response.""" try: # Skip streaming — no final usage record if kwargs.get("stream"): return model = kwargs.get("model", "unknown") messages = kwargs.get("messages") or [] usage = getattr(response, "usage", None) input_tokens = int(getattr(usage, "prompt_tokens", 0) or 0) output_tokens = int(getattr(usage, "completion_tokens", 0) or 0) cached = 0 pt_details = getattr(usage, "prompt_tokens_details", None) if pt_details is not None: cached = int(getattr(pt_details, "cached_tokens", 0) or 0) _write({ "timestamp": _now_iso(), "model": model, "input_tokens": input_tokens, "output_tokens": output_tokens, "cached_tokens": cached, "has_images": _messages_have_images(messages), "prompt_snippet": _extract_snippet(messages), }) except Exception: log.debug("openai record failed", exc_info=True) # -------------------------------------------------------------------------- # Anthropic patch # -------------------------------------------------------------------------- def _try_instrument_anthropic() -> None: try: from anthropic.resources.messages import Messages, AsyncMessages except ImportError: log.debug("anthropic SDK not installed — skipping") return _wrap_sync(Messages, "create", _record_anthropic) _wrap_async(AsyncMessages, "create", _record_anthropic) log.info("adapterly_logger: instrumented anthropic.messages") def _record_anthropic(kwargs: dict, response) -> None: try: if kwargs.get("stream"): return model = kwargs.get("model", "unknown") messages = kwargs.get("messages") or [] usage = getattr(response, "usage", None) input_tokens = int(getattr(usage, "input_tokens", 0) or 0) output_tokens = int(getattr(usage, "output_tokens", 0) or 0) cached = int(getattr(usage, "cache_read_input_tokens", 0) or 0) _write({ "timestamp": _now_iso(), "model": model, "input_tokens": input_tokens, "output_tokens": output_tokens, "cached_tokens": cached, "has_images": _messages_have_images(messages), "prompt_snippet": _extract_snippet(messages), }) except Exception: log.debug("anthropic record failed", exc_info=True) # -------------------------------------------------------------------------- # Wrapping helpers — preserve original signature and return value verbatim # -------------------------------------------------------------------------- def _wrap_sync(cls, method_name: str, recorder) -> None: original = getattr(cls, method_name) def wrapper(self, *args, **kwargs): response = original(self, *args, **kwargs) recorder(kwargs, response) return response wrapper.__wrapped__ = original setattr(cls, method_name, wrapper) def _wrap_async(cls, method_name: str, recorder) -> None: original = getattr(cls, method_name) async def wrapper(self, *args, **kwargs): response = await original(self, *args, **kwargs) recorder(kwargs, response) return response wrapper.__wrapped__ = original setattr(cls, method_name, wrapper) # -------------------------------------------------------------------------- # Message/prompt extraction (OpenAI and Anthropic use the same structure) # -------------------------------------------------------------------------- def _messages_have_images(messages) -> bool: for m in messages: content = m.get("content") if isinstance(m, dict) else None if isinstance(content, list): for part in content: if isinstance(part, dict) and part.get("type") in ("image", "image_url"): return True return False def _extract_snippet(messages) -> str | None: if os.environ.get("ADAPTERLY_LOGGER_NO_PROMPTS") == "1": return None for m in messages: if not isinstance(m, dict): continue if m.get("role") not in ("system", "user"): continue content = m.get("content") text = _content_to_text(content) if text: return text[:_SNIPPET_LEN] return None def _content_to_text(content) -> str: if isinstance(content, str): return content if isinstance(content, list): parts = [] for p in content: if isinstance(p, dict) and p.get("type") == "text": parts.append(str(p.get("text", ""))) return " ".join(parts) return "" # -------------------------------------------------------------------------- # File writer # -------------------------------------------------------------------------- def _write(record: dict) -> None: day = datetime.now(timezone.utc).strftime("%Y-%m-%d") path = _LOG_DIR / f"{day}.jsonl" line = json.dumps(record, ensure_ascii=False) with _LOCK: with path.open("a", encoding="utf-8") as f: f.write(line + "\n") def _now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") # Auto-instrument if imported as a module and env flag set — makes it easy to # enable via `PYTHONSTARTUP` or a wrapper script without touching app code. if os.environ.get("ADAPTERLY_LOGGER_AUTO") == "1": instrument()