Add ScopeBlind integration for device-level rate limiting

Adds a ScopeBlind callback that enables device-level rate limiting
for LLM proxy operators. Extracts DPoP proof headers (RFC 9449)
from incoming proxy requests and logs per-device usage data,
allowing operators to identify and rate-limit by device identity
instead of IP address — survives proxy rotation and VPNs.

Usage:
  litellm_settings:
    callbacks: ["scopeblind"]

Requires SCOPEBLIND_API_KEY environment variable.

- New file: litellm/integrations/scopeblind.py
- Registration in __init__.py and litellm_logging.py
- Unit tests in tests/test_litellm/integrations/test_scopeblind.py

Docs: https://scopeblind.com/docs

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

litellm/__init__.py

@@ -145,6 +145,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
     "focus",
     "posthog",
     "levo",
+    "scopeblind",
 ]
 cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None
 logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None

litellm/integrations/scopeblind.py

@@ -0,0 +1,238 @@
+"""
+ScopeBlind Integration for LiteLLM
+
+Device-level rate limiting for LLM proxy operators.
+Verifies DPoP proofs (RFC 9449) to bind requests to unique devices
+instead of IP addresses — survives proxy rotation, VPNs, and shared networks.
+
+Requires:
+    SCOPEBLIND_API_KEY  – your ScopeBlind API key (from scopeblind.com/t/<slug>)
+
+Optional:
+    SCOPEBLIND_ENDPOINT – verification endpoint (default: https://api.scopeblind.com)
+
+Usage (proxy config.yaml):
+    litellm_settings:
+      callbacks: ["scopeblind"]
+
+    environment_variables:
+      SCOPEBLIND_API_KEY: "sb_..."
+
+Or programmatically:
+    import litellm
+    litellm.callbacks = ["scopeblind"]
+
+Docs: https://scopeblind.com/docs
+"""
+
+import os
+import traceback
+from typing import Any, Dict, List, Optional, Union
+
+import litellm
+from litellm._logging import verbose_logger
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.llms.custom_httpx.http_handler import (
+    HTTPHandler,
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+
+
+def _get_utc_datetime():
+    import datetime as dt
+    from datetime import datetime
+
+    if hasattr(dt, "UTC"):
+        return datetime.now(dt.UTC)
+    else:
+        return datetime.utcnow()
+
+
+class ScopeBlindLogger(CustomLogger):
+    """
+    ScopeBlind callback for device-level rate limiting.
+
+    Extracts DPoP proof headers from incoming proxy requests,
+    verifies them with the ScopeBlind API, and logs per-device
+    usage data (model, cost, tokens) for abuse detection.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.validate_environment()
+        self.async_http_handler = get_async_httpx_client(
+            llm_provider=httpxSpecialProvider.LoggingCallback
+        )
+        self.sync_http_handler = HTTPHandler()
+        self.scopeblind_endpoint = os.getenv(
+            "SCOPEBLIND_ENDPOINT", "https://api.scopeblind.com"
+        )
+        self.scopeblind_api_key = os.getenv("SCOPEBLIND_API_KEY", "")
+
+    def validate_environment(self):
+        """Expects SCOPEBLIND_API_KEY in the environment."""
+        missing_keys: List[str] = []
+        if os.getenv("SCOPEBLIND_API_KEY", None) is None:
+            missing_keys.append("SCOPEBLIND_API_KEY")
+
+        if len(missing_keys) > 0:
+            raise Exception(
+                "ScopeBlind: Missing keys={} in environment.".format(missing_keys)
+            )
+
+    def _build_payload(
+        self,
+        kwargs: dict,
+        response_obj: Any,
+        start_time,
+        end_time,
+        event_type: str,
+    ) -> dict:
+        """Build a ScopeBlind event payload from LiteLLM kwargs."""
+        call_id = kwargs.get("litellm_call_id")
+        model = kwargs.get("model")
+        cost = kwargs.get("response_cost", None)
+        user = kwargs.get("user", None)
+
+        # Extract device identity from metadata
+        litellm_params = kwargs.get("litellm_params", {})
+        metadata = litellm_params.get("metadata", {}) or {}
+        headers = metadata.get("headers", {}) or {}
+
+        # DPoP proof and device ID from incoming request headers
+        dpop_proof = headers.get("x-scopeblind-dpop") or headers.get("dpop")
+        device_id = headers.get("x-scopeblind-device-id") or headers.get(
+            "x-device-identity"
+        )
+
+        # Token usage
+        usage = {}
+        if (
+            isinstance(response_obj, litellm.ModelResponse)
+            or isinstance(response_obj, litellm.EmbeddingResponse)
+        ) and hasattr(response_obj, "usage"):
+            usage = {
+                "prompt_tokens": response_obj["usage"].get("prompt_tokens", 0),
+                "completion_tokens": response_obj["usage"].get(
+                    "completion_tokens", 0
+                ),
+                "total_tokens": response_obj["usage"].get("total_tokens", 0),
+            }
+
+        # If no user provided, try API key user ID
+        if user is None:
+            user = metadata.get("user_api_key_user_id")
+
+        return {
+            "event": event_type,
+            "litellm_call_id": call_id,
+            "model": model,
+            "user": user,
+            "cost": cost,
+            "usage": usage,
+            "dpop_proof": dpop_proof,
+            "device_id": device_id,
+            "timestamp": _get_utc_datetime().isoformat(),
+            "start_time": str(start_time),
+            "end_time": str(end_time),
+        }
+
+    async def _send_event(self, payload: dict) -> None:
+        """Send an event to the ScopeBlind API."""
+        try:
+            url = f"{self.scopeblind_endpoint}/v1/proxy/events"
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {self.scopeblind_api_key}",
+            }
+            await self.async_http_handler.post(
+                url=url,
+                json=payload,
+                headers=headers,
+            )
+        except Exception as e:
+            verbose_logger.debug(
+                "ScopeBlind: Failed to send event: %s", str(e)
+            )
+
+    def _send_event_sync(self, payload: dict) -> None:
+        """Send an event synchronously."""
+        try:
+            url = f"{self.scopeblind_endpoint}/v1/proxy/events"
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {self.scopeblind_api_key}",
+            }
+            self.sync_http_handler.post(
+                url=url,
+                json=payload,
+                headers=headers,
+            )
+        except Exception as e:
+            verbose_logger.debug(
+                "ScopeBlind: Failed to send event (sync): %s", str(e)
+            )
+
+    async def async_log_success_event(
+        self, kwargs, response_obj, start_time, end_time
+    ):
+        """Log successful LLM call with device identity."""
+        try:
+            payload = self._build_payload(
+                kwargs, response_obj, start_time, end_time, "llm_call_success"
+            )
+            # Only send if we have device identity info
+            if payload.get("dpop_proof") or payload.get("device_id"):
+                await self._send_event(payload)
+            else:
+                verbose_logger.debug(
+                    "ScopeBlind: No device identity headers found, skipping event"
+                )
+        except Exception as e:
+            verbose_logger.debug(
+                "ScopeBlind: Error logging success event: %s\n%s",
+                str(e),
+                traceback.format_exc(),
+            )
+
+    def log_success_event(self, kwargs, response_obj, start_time, end_time):
+        """Log successful LLM call synchronously."""
+        try:
+            payload = self._build_payload(
+                kwargs, response_obj, start_time, end_time, "llm_call_success"
+            )
+            if payload.get("dpop_proof") or payload.get("device_id"):
+                self._send_event_sync(payload)
+        except Exception as e:
+            verbose_logger.debug(
+                "ScopeBlind: Error logging success event (sync): %s", str(e)
+            )
+
+    async def async_log_failure_event(
+        self, kwargs, response_obj, start_time, end_time
+    ):
+        """Log failed LLM call with device identity."""
+        try:
+            payload = self._build_payload(
+                kwargs, response_obj, start_time, end_time, "llm_call_failure"
+            )
+            if payload.get("dpop_proof") or payload.get("device_id"):
+                await self._send_event(payload)
+        except Exception as e:
+            verbose_logger.debug(
+                "ScopeBlind: Error logging failure event: %s", str(e)
+            )
+
+    def log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        """Log failed LLM call synchronously."""
+        try:
+            payload = self._build_payload(
+                kwargs, response_obj, start_time, end_time, "llm_call_failure"
+            )
+            if payload.get("dpop_proof") or payload.get("device_id"):
+                self._send_event_sync(payload)
+        except Exception as e:
+            verbose_logger.debug(
+                "ScopeBlind: Error logging failure event (sync): %s", str(e)
+            )

litellm/litellm_core_utils/litellm_logging.py

@@ -155,6 +155,7 @@ from ..integrations.lunary import LunaryLogger
 from ..integrations.openmeter import OpenMeterLogger
 from ..integrations.opik.opik import OpikLogger
 from ..integrations.posthog import PostHogLogger
+from ..integrations.scopeblind import ScopeBlindLogger
 from ..integrations.prompt_layer import PromptLayerLogger
 from ..integrations.s3 import S3Logger
 from ..integrations.s3_v2 import S3Logger as S3V2Logger
@@ -3528,6 +3529,8 @@ def set_callbacks(callback_list, function_id=None):  # noqa: PLR0915
                 )
             elif callback == "openmeter":
                 openMeterLogger = OpenMeterLogger()
+            elif callback == "scopeblind":
+                scopeBlindLogger = ScopeBlindLogger()
             elif callback == "datadog":
                 dataDogLogger = DataDogLogger()
             elif callback == "dynamodb":
@@ -3590,6 +3593,14 @@ def _init_custom_logger_compatible_class(  # noqa: PLR0915
             _openmeter_logger = OpenMeterLogger()
             _in_memory_loggers.append(_openmeter_logger)
             return _openmeter_logger  # type: ignore
+        elif logging_integration == "scopeblind":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, ScopeBlindLogger):
+                    return callback  # type: ignore
+
+            _scopeblind_logger = ScopeBlindLogger()
+            _in_memory_loggers.append(_scopeblind_logger)
+            return _scopeblind_logger  # type: ignore
         elif logging_integration == "posthog":
             for callback in _in_memory_loggers:
                 if isinstance(callback, PostHogLogger):
@@ -4221,6 +4232,10 @@ def get_custom_logger_compatible_class(  # noqa: PLR0915
             for callback in _in_memory_loggers:
                 if isinstance(callback, OpenMeterLogger):
                     return callback
+        elif logging_integration == "scopeblind":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, ScopeBlindLogger):
+                    return callback
         elif logging_integration == "braintrust":
             from litellm.integrations.braintrust_logging import BraintrustLogger
 

tests/test_litellm/integrations/test_scopeblind.py

@@ -0,0 +1,205 @@
+"""
+Tests for the ScopeBlind integration.
+
+Tests that the ScopeBlind callback correctly:
+1. Validates environment variables
+2. Builds payloads from LiteLLM kwargs
+3. Skips events when no device identity headers are present
+4. Sends events when device identity headers are present
+"""
+
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+sys.path.insert(
+    0, os.path.abspath("../..")
+)  # Adds the parent directory to the system-path
+
+
+class TestScopeBlindValidation:
+    """Test environment validation."""
+
+    def test_missing_api_key_raises(self):
+        """ScopeBlindLogger raises if SCOPEBLIND_API_KEY is not set."""
+        with patch.dict(os.environ, {}, clear=True):
+            # Remove the key if it exists
+            os.environ.pop("SCOPEBLIND_API_KEY", None)
+            from litellm.integrations.scopeblind import ScopeBlindLogger
+
+            with pytest.raises(Exception, match="Missing keys"):
+                ScopeBlindLogger()
+
+    def test_valid_environment(self):
+        """ScopeBlindLogger initializes with valid env vars."""
+        with patch.dict(
+            os.environ,
+            {"SCOPEBLIND_API_KEY": "sb_test_key"},
+        ):
+            from litellm.integrations.scopeblind import ScopeBlindLogger
+
+            logger = ScopeBlindLogger()
+            assert logger.scopeblind_api_key == "sb_test_key"
+            assert logger.scopeblind_endpoint == "https://api.scopeblind.com"
+
+    def test_custom_endpoint(self):
+        """ScopeBlindLogger respects SCOPEBLIND_ENDPOINT env var."""
+        with patch.dict(
+            os.environ,
+            {
+                "SCOPEBLIND_API_KEY": "sb_test_key",
+                "SCOPEBLIND_ENDPOINT": "https://custom.scopeblind.dev",
+            },
+        ):
+            from litellm.integrations.scopeblind import ScopeBlindLogger
+
+            logger = ScopeBlindLogger()
+            assert logger.scopeblind_endpoint == "https://custom.scopeblind.dev"
+
+
+class TestScopeBlindPayload:
+    """Test payload building."""
+
+    @pytest.fixture
+    def logger(self):
+        with patch.dict(
+            os.environ,
+            {"SCOPEBLIND_API_KEY": "sb_test_key"},
+        ):
+            from litellm.integrations.scopeblind import ScopeBlindLogger
+
+            return ScopeBlindLogger()
+
+    def test_build_payload_with_device_headers(self, logger):
+        """Payload includes DPoP proof and device ID from headers."""
+        kwargs = {
+            "litellm_call_id": "call-123",
+            "model": "gpt-4",
+            "response_cost": 0.03,
+            "user": "user-456",
+            "litellm_params": {
+                "metadata": {
+                    "headers": {
+                        "x-scopeblind-dpop": "eyJ0eXAiOiJkcG9wK2p3dCJ9.test",
+                        "x-scopeblind-device-id": "device-789",
+                    }
+                }
+            },
+        }
+        response_obj = MagicMock()
+        response_obj.__class__.__name__ = "ModelResponse"
+
+        payload = logger._build_payload(
+            kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01", "llm_call_success"
+        )
+
+        assert payload["event"] == "llm_call_success"
+        assert payload["model"] == "gpt-4"
+        assert payload["user"] == "user-456"
+        assert payload["cost"] == 0.03
+        assert payload["dpop_proof"] == "eyJ0eXAiOiJkcG9wK2p3dCJ9.test"
+        assert payload["device_id"] == "device-789"
+        assert payload["litellm_call_id"] == "call-123"
+
+    def test_build_payload_without_device_headers(self, logger):
+        """Payload has None for device fields when no headers present."""
+        kwargs = {
+            "litellm_call_id": "call-123",
+            "model": "gpt-4",
+            "litellm_params": {"metadata": {}},
+        }
+        response_obj = MagicMock()
+
+        payload = logger._build_payload(
+            kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01", "llm_call_success"
+        )
+
+        assert payload["dpop_proof"] is None
+        assert payload["device_id"] is None
+
+    def test_build_payload_with_x_device_identity_header(self, logger):
+        """Payload extracts X-Device-Identity header."""
+        kwargs = {
+            "litellm_call_id": "call-123",
+            "model": "gpt-4",
+            "litellm_params": {
+                "metadata": {
+                    "headers": {
+                        "x-device-identity": 'ScopeBlind/1.0; info="https://scopeblind.com/verify-agent"',
+                    }
+                }
+            },
+        }
+        response_obj = MagicMock()
+
+        payload = logger._build_payload(
+            kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01", "llm_call_success"
+        )
+
+        assert payload["device_id"] == 'ScopeBlind/1.0; info="https://scopeblind.com/verify-agent"'
+
+
+class TestScopeBlindEventSending:
+    """Test that events are sent or skipped correctly."""
+
+    @pytest.fixture
+    def logger(self):
+        with patch.dict(
+            os.environ,
+            {"SCOPEBLIND_API_KEY": "sb_test_key"},
+        ):
+            from litellm.integrations.scopeblind import ScopeBlindLogger
+
+            instance = ScopeBlindLogger()
+            instance.sync_http_handler = MagicMock()
+            return instance
+
+    def test_skips_event_without_device_identity(self, logger):
+        """log_success_event should not send when no device headers."""
+        kwargs = {
+            "litellm_call_id": "call-123",
+            "model": "gpt-4",
+            "litellm_params": {"metadata": {}},
+        }
+        response_obj = MagicMock()
+
+        logger.log_success_event(
+            kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01"
+        )
+
+        logger.sync_http_handler.post.assert_not_called()
+
+    def test_sends_event_with_device_identity(self, logger):
+        """log_success_event should send when DPoP header is present."""
+        kwargs = {
+            "litellm_call_id": "call-123",
+            "model": "gpt-4",
+            "litellm_params": {
+                "metadata": {
+                    "headers": {
+                        "x-scopeblind-dpop": "eyJ0eXAiOiJkcG9wK2p3dCJ9.test",
+                    }
+                }
+            },
+        }
+        response_obj = MagicMock()
+
+        logger.log_success_event(
+            kwargs, response_obj, "2024-01-01T00:00:00", "2024-01-01T00:00:01"
+        )
+
+        logger.sync_http_handler.post.assert_called_once()
+        call_kwargs = logger.sync_http_handler.post.call_args
+        assert "scopeblind" in call_kwargs.kwargs["url"]
+
+
+class TestScopeBlindRegistration:
+    """Test that ScopeBlind is registered as a named callback."""
+
+    def test_scopeblind_in_callbacks_literal(self):
+        """scopeblind should be in the known callbacks list."""
+        import litellm
+
+        assert "scopeblind" in litellm._known_custom_logger_compatible_callbacks