Python SDK - Esper Docs

Start With Integrations For Faster SetupIf you want a recipe-backed path for a specific platform, start with the Integrations guides. The SDK is best when you want direct control inside your own framework or service boundary.

Installation

pip install esper-py

For development:

pip install esper-py[dev]

Basic Usage

from esper import EsperClient, EsperConfig, MitigationRequest, MitigationAction
from esper import BeaconEvent

# Initialize client
config = EsperConfig(api_key="your-api-key")
client = EsperClient(config)

# Send the first request's traffic to Esper beacon for analysis.
event = BeaconEvent(
    type="http_request",
    ip="192.168.1.1",
    user_agent="Mozilla/5.0...",
    path="/login",
    method="POST"
)
client.send_beacon_event(event)

# Check a later request against active mitigation state.
request = MitigationRequest(
    ip="192.168.1.1",
    user_agent="Mozilla/5.0...",
    path="/checkout",
    method="POST",
    metadata={"request_id": "req-2", "session_id": "session-123"},
)

response = client.check_mitigation(request)

if response.action == MitigationAction.BLOCK:
    print("Request blocked:", response.reason)
elif response.action == MitigationAction.CHALLENGE:
    print("Challenge required:", response.challenge)
else:
    print("Request allowed")

Async Support

import asyncio
from esper import AsyncEsperClient, EsperConfig, MitigationRequest

async def check_request():
    config = EsperConfig(api_key="your-api-key")

    async with AsyncEsperClient(config) as client:
        request = MitigationRequest(ip="192.168.1.1")
        response = await client.check_mitigation(request)
        return response

# Run async function
response = asyncio.run(check_request())

Configuration

from esper import EsperConfig

config = EsperConfig(
    api_key="your-api-key",              # Required
    api_url="https://api.esperr.com",    # Optional
    timeout=30.0,                         # Optional (seconds)
    max_retries=3,                        # Optional
    retry_delay=1.0                       # Optional (seconds)
)

API Methods

check_mitigation(request)

Check if a request should be mitigated.

from esper import MitigationRequest

request = MitigationRequest(
    ip="192.168.1.1",
    user_agent="Mozilla/5.0...",
    path="/api/endpoint",
    method="POST",
    headers={"X-Custom": "value"},
    metadata={"user_id": "123"}
)

response = client.check_mitigation(request)
print(f"Action: {response.action}")
print(f"Score: {response.score}")
print(f"Reason: {response.reason}")

verify_challenge(token, response)

Verify a user’s challenge response.

is_valid = client.verify_challenge(
    token="challenge-token",
    response="user-response"
)

if is_valid:
    print("Challenge passed")
else:
    print("Challenge failed")

send_beacon_event(event)

Send request traffic to the beacon server so Esper can analyze it and update state for later mitigation checks.

from esper import BeaconEvent

event = BeaconEvent(
    type="page_view",
    ip="192.168.1.1",
    user_agent="Mozilla/5.0...",
    session_id="session-123",
    data={"page": "/home", "referrer": "/login"}
)

client.send_beacon_event(event)

get_usage()

Get API usage statistics.

usage = client.get_usage()
print(f"Total requests: {usage.requests}")
print(f"Blocked: {usage.blocked}")
print(f"Challenged: {usage.challenged}")
print(f"Allowed: {usage.allowed}")

Framework Integration

Django Middleware

# middleware.py
from django.http import JsonResponse
from django.conf import settings
from esper import EsperClient, EsperConfig, MitigationRequest, MitigationAction
import logging

logger = logging.getLogger(__name__)

class EsperMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        config = EsperConfig(api_key=settings.ESPER_API_KEY)
        self.client = EsperClient(config)

    def __call__(self, request):
        # Check mitigation
        mitigation_request = MitigationRequest(
            ip=self.get_client_ip(request),
            user_agent=request.META.get('HTTP_USER_AGENT'),
            path=request.path,
            method=request.method
        )

        try:
            response = self.client.check_mitigation(mitigation_request)

            if response.action == MitigationAction.BLOCK:
                return JsonResponse(
                    {"error": "Access denied"},
                    status=403
                )
            elif response.action == MitigationAction.CHALLENGE:
                return JsonResponse(
                    {"challenge": response.challenge.dict()},
                    status=429
                )
        except Exception as e:
            logger.error(f"Esper error: {e}")

        return self.get_response(request)

    def get_client_ip(self, request):
        x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
        if x_forwarded_for:
            return x_forwarded_for.split(',')[0]
        return request.META.get('REMOTE_ADDR')

# settings.py
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'your_app.middleware.EsperMiddleware',  # Add this
    # ... other middleware
]

FastAPI Middleware

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from esper import AsyncEsperClient, EsperConfig, MitigationRequest
import logging

logger = logging.getLogger(__name__)

app = FastAPI()
config = EsperConfig(api_key="your-api-key")
esper = AsyncEsperClient(config)

@app.middleware("http")
async def esper_middleware(request: Request, call_next):
    # Check mitigation
    mitigation_request = MitigationRequest(
        ip=request.client.host,
        user_agent=request.headers.get("user-agent"),
        path=request.url.path,
        method=request.method
    )

    try:
        response = await esper.check_mitigation(mitigation_request)

        if response.action == MitigationAction.BLOCK:
            raise HTTPException(
                status_code=403,
                detail="Access denied"
            )
        elif response.action == MitigationAction.CHALLENGE:
            return JSONResponse(
                status_code=429,
                content={"challenge": response.challenge.dict()}
            )
    except Exception as e:
        logger.error(f"Esper error: {e}")

    return await call_next(request)

@app.on_event("shutdown")
async def shutdown_event():
    await esper.close()

Flask Extension

# flask_esper.py
from flask import request, jsonify, abort, g
from esper import EsperClient, EsperConfig, MitigationRequest, MitigationAction
from functools import wraps
import logging

logger = logging.getLogger(__name__)

class FlaskEsper:
    def __init__(self, app=None, config=None):
        self.client = None
        if app is not None:
            self.init_app(app, config)

    def init_app(self, app, config=None):
        if config is None:
            config = EsperConfig(
                api_key=app.config['ESPER_API_KEY']
            )

        self.client = EsperClient(config)
        app.before_request(self.check_request)

    def check_request(self):
        if request.endpoint and request.endpoint.startswith('static'):
            return

        mitigation_request = MitigationRequest(
            ip=request.remote_addr,
            user_agent=request.headers.get('User-Agent'),
            path=request.path,
            method=request.method
        )

        try:
            response = self.client.check_mitigation(mitigation_request)
            g.esper_result = response

            if response.action == MitigationAction.BLOCK:
                abort(403, description="Access denied")
            elif response.action == MitigationAction.CHALLENGE:
                return jsonify({
                    "challenge": response.challenge.dict()
                }), 429
        except Exception as e:
            logger.error(f"Esper error: {e}")

# app.py
from flask import Flask
from flask_esper import FlaskEsper

app = Flask(__name__)
app.config['ESPER_API_KEY'] = 'your-api-key'

esper = FlaskEsper(app)

Error Handling

from esper import (
    EsperAPIError,
    EsperConnectionError,
    EsperTimeoutError,
    EsperValidationError
)

try:
    response = client.check_mitigation(request)
except EsperAPIError as e:
    print(f"API error: {e.message}")
    print(f"Status code: {e.status_code}")
    print(f"Error code: {e.error_code}")
except EsperConnectionError as e:
    print(f"Connection error: {e}")
except EsperTimeoutError as e:
    print(f"Timeout error: {e}")
except EsperValidationError as e:
    print(f"Validation error: {e}")

Type Hints

The SDK includes full type hints for better IDE support:

from esper import (
    EsperClient,
    EsperConfig,
    MitigationRequest,
    MitigationResponse,
    MitigationAction,
    BeaconEvent,
    ChallengeData
)

def process_request(client: EsperClient, request: MitigationRequest) -> MitigationResponse:
    """Process request with full type hints."""
    response: MitigationResponse = client.check_mitigation(request)

    if response.action == MitigationAction.BLOCK:
        raise Exception("Blocked")

    return response

Testing

# test_esper.py
import pytest
from unittest.mock import Mock, patch
from esper import EsperClient, EsperConfig, MitigationRequest, MitigationAction, MitigationAction

@pytest.fixture
def client():
    config = EsperConfig(api_key="test-key")
    return EsperClient(config)

@pytest.fixture
def mock_response():
    return Mock(
        action=MitigationAction.ALLOW,
        score=10.0,
        reason="Clean traffic"
    )

def test_check_mitigation(client, mock_response):
    with patch.object(client, 'check_mitigation', return_value=mock_response):
        request = MitigationRequest(ip="192.168.1.1")
        response = client.check_mitigation(request)

        assert response.action == MitigationAction.ALLOW
        assert response.score == 10.0

@pytest.mark.asyncio
async def test_async_check_mitigation():
    from esper import AsyncEsperClient

    config = EsperConfig(api_key="test-key")
    async with AsyncEsperClient(config) as client:
        with patch.object(client, 'check_mitigation') as mock_check:
            mock_check.return_value = Mock(
                action=MitigationAction.BLOCK
            )

            request = MitigationRequest(ip="10.0.0.1")
            response = await client.check_mitigation(request)

            assert response.action == MitigationAction.BLOCK

Advanced Usage

Custom Retry Logic

from esper import EsperConfig, EsperClient

config = EsperConfig(
    api_key="your-api-key",
    max_retries=5,
    retry_delay=2.0  # 2 seconds base delay
)

client = EsperClient(config)

Batch Processing

import asyncio
from esper import AsyncEsperClient, EsperConfig, MitigationRequest

async def batch_check(requests: list[MitigationRequest]):
    config = EsperConfig(api_key="your-api-key")

    async with AsyncEsperClient(config) as client:
        tasks = [
            client.check_mitigation(req)
            for req in requests
        ]
        results = await asyncio.gather(*tasks)
        return results

# Usage
requests = [
    MitigationRequest(ip="192.168.1.1"),
    MitigationRequest(ip="192.168.1.2"),
    MitigationRequest(ip="192.168.1.3"),
]

results = asyncio.run(batch_check(requests))

Examples

Full examples are available in the GitHub repository.

Documentation Index

​Installation

​Basic Usage

​Async Support

​Configuration

​API Methods

​check_mitigation(request)

​verify_challenge(token, response)

​send_beacon_event(event)

​get_usage()

​Framework Integration

​Django Middleware

​FastAPI Middleware

​Flask Extension

​Error Handling

​Type Hints

​Testing

​Advanced Usage

​Custom Retry Logic

​Batch Processing

​Examples

Installation

Basic Usage

Async Support

Configuration

API Methods

check_mitigation(request)

verify_challenge(token, response)

send_beacon_event(event)

get_usage()

Framework Integration

Django Middleware

FastAPI Middleware

Flask Extension

Error Handling

Type Hints

Testing

Advanced Usage

Custom Retry Logic

Batch Processing

Examples