Python SDK

Link to section: IntroductionIntroduction

The LockLLM Python SDK is a production-ready library that provides comprehensive AI security for your LLM applications. Built with Python type hints and designed for modern Python development, it offers both synchronous and asynchronous APIs with drop-in replacements for popular AI provider SDKs with automatic prompt injection detection and jailbreak prevention.

Key features:

• Real-time security scanning with minimal latency (<250ms)
• Dual sync/async API for maximum flexibility
• Drop-in replacements for 17+ AI providers (custom endpoint support for each)
• Full type hints with mypy support
• Works with Python 3.8 through 3.12
• Streaming-compatible with all providers
• Context manager support
• Completely free with unlimited usage

Use cases:

• Production LLM applications requiring security
• AI agents and autonomous systems
• Chatbots and conversational interfaces
• RAG (Retrieval Augmented Generation) systems
• Multi-tenant AI applications
• Enterprise AI deployments
• FastAPI, Django, and Flask applications

Link to section: InstallationInstallation

Install the SDK using your preferred package manager:

# pip
pip install lockllm

# pip3
pip3 install lockllm

# poetry
poetry add lockllm

# pipenv
pipenv install lockllm

Requirements:

• Python 3.8 or higher (supports 3.8, 3.9, 3.10, 3.11, 3.12)
• httpx (installed automatically as a dependency)

Link to section: Optional DependenciesOptional Dependencies

For provider wrapper functions, install the relevant official SDKs:

# For OpenAI and OpenAI-compatible providers
pip install openai

# For Anthropic Claude
pip install anthropic

# For Cohere
pip install cohere

Provider SDK mapping:

• openai - OpenAI, Groq, DeepSeek, Mistral, Perplexity, OpenRouter, Together AI, xAI, Fireworks, Anyscale, Hugging Face, Gemini, Azure, Bedrock, Vertex AI
• anthropic - Anthropic Claude
• cohere - Cohere (optional)

These SDKs are only required if you use the wrapper functions for those providers.

Link to section: Quick StartQuick Start

Link to section: Step 1: Get Your API KeysStep 1: Get Your API Keys

1. Visit lockllm.com and create a free account
2. Navigate to API Keys section and copy your LockLLM API key
3. Go to Proxy Settings and add your provider API keys (OpenAI, Anthropic, etc.)

Your provider keys are encrypted and stored securely. You'll only need your LockLLM API key in your code.

Link to section: Step 2: Basic UsageStep 2: Basic Usage

Choose from three integration methods: wrapper functions (easiest), direct scan API, or official SDKs with custom base URL.

Link to section: Wrapper Functions (Recommended)Wrapper Functions (Recommended)

The simplest way to add security - replace your SDK initialization:

Synchronous:

from lockllm import create_openai
import os

# Before:
# from openai import OpenAI
# openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# After (one line change):
openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

# Everything else works exactly the same
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": user_input}]
)

print(response.choices[0].message.content)

Asynchronous:

from lockllm import create_async_openai
import os
import asyncio

async def main():
    openai = create_async_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

    response = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_input}]
    )

    print(response.choices[0].message.content)

asyncio.run(main())

Link to section: Direct Scan APIDirect Scan API

For manual control and custom workflows:

Synchronous:

from lockllm import LockLLM
import os

lockllm = LockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

# Scan user input before processing
result = lockllm.scan(
    input=user_prompt,
    sensitivity="medium"  # "low" | "medium" | "high"
)

if not result.safe:
    print("Malicious input detected!")
    print(f"Injection score: {result.injection}%")
    print(f"Confidence: {result.confidence}%")
    print(f"Request ID: {result.request_id}")

    # Handle security incident
    return {"error": "Invalid input detected"}

# Safe to proceed
response = your_llm_call(user_prompt)

Asynchronous:

from lockllm import AsyncLockLLM
import os
import asyncio

async def main():
    lockllm = AsyncLockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

    result = await lockllm.scan(
        input=user_prompt,
        sensitivity="medium"
    )

    if not result.safe:
        print(f"Malicious prompt detected: {result.injection}%")
        return

    # Safe to proceed
    response = await your_llm_call(user_prompt)

asyncio.run(main())

Link to section: Official SDKs with ProxyOfficial SDKs with Proxy

Use official SDKs with LockLLM's proxy:

from openai import OpenAI
from lockllm import get_proxy_url
import os

client = OpenAI(
    api_key=os.getenv("LOCKLLM_API_KEY"),
    base_url=get_proxy_url('openai')
)

# Works exactly like the official SDK
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)

Link to section: Provider WrappersProvider Wrappers

LockLLM provides drop-in replacements for 17+ AI providers with custom endpoint support. All wrappers work identically to the official SDKs with automatic security scanning.

Link to section: OpenAI (Sync)OpenAI (Sync)

from lockllm import create_openai
import os

openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

# Chat completions
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": user_input}
    ],
    temperature=0.7,
    max_tokens=1000
)

print(response.choices[0].message.content)

# Streaming
stream = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Count from 1 to 10"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end='')

# Function calling
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "What's the weather in Boston?"}],
    functions=[{
        "name": "get_weather",
        "description": "Get the current weather in a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }]
)

Link to section: OpenAI (Async)OpenAI (Async)

from lockllm import create_async_openai
import os
import asyncio

async def main():
    openai = create_async_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

    # Chat completions
    response = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_input}]
    )

    # Async streaming
    stream = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Write a story"}],
        stream=True
    )

    async for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end='')

asyncio.run(main())

Link to section: Anthropic Claude (Sync)Anthropic Claude (Sync)

from lockllm import create_anthropic
import os

anthropic = create_anthropic(api_key=os.getenv("LOCKLLM_API_KEY"))

# Messages API
message = anthropic.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": user_input}
    ]
)

print(message.content[0].text)

# Streaming
with anthropic.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a poem"}]
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)

Link to section: Anthropic Claude (Async)Anthropic Claude (Async)

from lockllm import create_async_anthropic
import os
import asyncio

async def main():
    anthropic = create_async_anthropic(api_key=os.getenv("LOCKLLM_API_KEY"))

    # Async messages
    message = await anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_input}]
    )

    # Async streaming
    async with anthropic.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Write a poem"}]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end='', flush=True)

asyncio.run(main())

Link to section: Groq, DeepSeek, PerplexityGroq, DeepSeek, Perplexity

from lockllm import create_groq, create_deepseek, create_perplexity
import os

# Groq - Fast inference with Llama models
groq = create_groq(api_key=os.getenv("LOCKLLM_API_KEY"))
response = groq.chat.completions.create(
    model='llama-3.1-70b-versatile',
    messages=[{'role': 'user', 'content': user_input}]
)

# DeepSeek - Advanced reasoning models
deepseek = create_deepseek(api_key=os.getenv("LOCKLLM_API_KEY"))
response = deepseek.chat.completions.create(
    model='deepseek-chat',
    messages=[{'role': 'user', 'content': user_input}]
)

# Perplexity - Models with internet access
perplexity = create_perplexity(api_key=os.getenv("LOCKLLM_API_KEY"))
response = perplexity.chat.completions.create(
    model='llama-3.1-sonar-huge-128k-online',
    messages=[{'role': 'user', 'content': user_input}]
)

Link to section: All Supported ProvidersAll Supported Providers

LockLLM supports 17+ providers with ready-to-use wrappers. All providers support custom endpoint URLs configured via the dashboard.

Import any wrapper function:

Synchronous wrappers:

from lockllm import (
    create_openai,        # OpenAI GPT models
    create_anthropic,     # Anthropic Claude
    create_groq,          # Groq LPU inference
    create_deepseek,      # DeepSeek models
    create_perplexity,    # Perplexity (with internet)
    create_mistral,       # Mistral AI
    create_openrouter,    # OpenRouter (multi-provider)
    create_together,      # Together AI
    create_xai,           # xAI Grok
    create_fireworks,     # Fireworks AI
    create_anyscale,      # Anyscale Endpoints
    create_huggingface,   # Hugging Face Inference
    create_gemini,        # Google Gemini
    create_cohere,        # Cohere
    create_azure,         # Azure OpenAI
    create_bedrock,       # AWS Bedrock
    create_vertex_ai      # Google Vertex AI
)

Asynchronous wrappers:

from lockllm import (
    create_async_openai,
    create_async_anthropic,
    create_async_groq,
    create_async_deepseek,
    create_async_perplexity,
    create_async_mistral,
    create_async_openrouter,
    create_async_together,
    create_async_xai,
    create_async_fireworks,
    create_async_anyscale,
    create_async_huggingface,
    create_async_gemini,
    create_async_cohere,
    create_async_azure,
    create_async_bedrock,
    create_async_vertex_ai
)

Provider compatibility:

• 15+ providers use OpenAI-compatible API (require openai package)
• Anthropic uses its own SDK (requires anthropic)
• Cohere uses its own SDK (requires cohere, optional)
• All providers support custom endpoint URLs via dashboard

Link to section: ConfigurationConfiguration

Link to section: LockLLM Client ConfigurationLockLLM Client Configuration

Synchronous:

from lockllm import LockLLM, LockLLMConfig
import os

config = LockLLMConfig(
    api_key=os.getenv("LOCKLLM_API_KEY"),  # Required
    base_url="https://api.lockllm.com",     # Optional: custom endpoint
    timeout=60.0,                            # Optional: request timeout (seconds)
    max_retries=3                            # Optional: max retry attempts
)

lockllm = LockLLM(api_key=config.api_key, base_url=config.base_url, timeout=config.timeout)

Asynchronous:

from lockllm import AsyncLockLLM
import os

lockllm = AsyncLockLLM(
    api_key=os.getenv("LOCKLLM_API_KEY"),
    base_url="https://api.lockllm.com",
    timeout=60.0,
    max_retries=3
)

Link to section: Sensitivity LevelsSensitivity Levels

Control detection strictness with the sensitivity parameter:

from lockllm import LockLLM
import os

lockllm = LockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

# Low sensitivity - fewer false positives (threshold: 40)
# Use for: creative applications, exploratory use cases
low_result = lockllm.scan(input=user_prompt, sensitivity="low")

# Medium sensitivity - balanced detection (threshold: 25) - DEFAULT
# Use for: general user inputs, standard applications
medium_result = lockllm.scan(input=user_prompt, sensitivity="medium")

# High sensitivity - maximum protection (threshold: 10)
# Use for: sensitive operations, admin panels, data exports
high_result = lockllm.scan(input=user_prompt, sensitivity="high")

Choosing sensitivity:

• High: Critical systems (admin, payments, sensitive data)
• Medium: General applications (default, recommended)
• Low: Creative tools (writing assistants, brainstorming)

Link to section: Custom EndpointsCustom Endpoints

All providers support custom endpoint URLs for:

• Self-hosted LLM deployments (OpenAI-compatible APIs)
• Azure OpenAI resources with custom endpoints
• Alternative API gateways and reverse proxies
• Private cloud or air-gapped deployments
• Development and staging environments

How it works: Configure custom endpoints in the LockLLM dashboard when adding any provider API key. The SDK wrappers automatically use your custom endpoint URL.

# The wrapper automatically uses your custom endpoint
azure = create_azure(api_key=os.getenv("LOCKLLM_API_KEY"))

# Your custom Azure endpoint is configured in the dashboard:
# - Endpoint: https://your-resource.openai.azure.com
# - Deployment: gpt-4
# - API Version: 2024-10-21

Example - Self-hosted model: If you have a self-hosted model with an OpenAI-compatible API, configure it in the dashboard using one of the OpenAI-compatible provider wrappers (e.g., OpenAI, Groq) with your custom endpoint URL.

# Use OpenAI wrapper with custom endpoint configured in dashboard
openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

# Dashboard configuration:
# - Provider: OpenAI
# - Custom Endpoint: https://your-self-hosted-llm.com/v1
# - API Key: your-model-api-key

Link to section: Request OptionsRequest Options

Override configuration per-request:

# Per-request timeout
result = lockllm.scan(
    input=user_prompt,
    sensitivity="high",
    timeout=30.0  # 30 second timeout for this request
)

Link to section: Advanced FeaturesAdvanced Features

Link to section: Streaming ResponsesStreaming Responses

All provider wrappers support streaming:

Synchronous streaming:

openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

stream = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Write a story"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end='')

Asynchronous streaming:

import asyncio

async def main():
    openai = create_async_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

    stream = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Write a story"}],
        stream=True
    )

    async for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end='')

asyncio.run(main())

Link to section: Function CallingFunction Calling

OpenAI function calling works seamlessly:

openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "What's the weather in Boston?"}],
    functions=[{
        "name": "get_weather",
        "description": "Get the current weather in a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }],
    function_call="auto"
)

if response.choices[0].message.function_call:
    function_call = response.choices[0].message.function_call
    import json
    args = json.loads(function_call.arguments)

    # Call your function with the parsed arguments
    weather = get_weather(args['location'], args.get('unit', 'celsius'))

    # Send function result back to LLM
    final_response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "user", "content": "What's the weather in Boston?"},
            response.choices[0].message,
            {"role": "function", "name": "get_weather", "content": json.dumps(weather)}
        ]
    )

Link to section: Multi-Turn ConversationsMulti-Turn Conversations

Maintain conversation context with message history:

openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "What is the capital of France?"}
]

response = openai.chat.completions.create(
    model="gpt-4",
    messages=messages
)

# Add assistant response to history
messages.append({
    "role": "assistant",
    "content": response.choices[0].message.content
})

# Continue conversation
messages.append({
    "role": "user",
    "content": "What is its population?"
})

response = openai.chat.completions.create(
    model="gpt-4",
    messages=messages
)

Link to section: FastAPI IntegrationFastAPI Integration

Integrate with FastAPI for automatic request scanning:

from fastapi import FastAPI, HTTPException, Depends
from lockllm import AsyncLockLLM
from pydantic import BaseModel
import os

app = FastAPI()
lockllm = AsyncLockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

class ChatRequest(BaseModel):
    prompt: str

async def scan_prompt(request: ChatRequest):
    """Dependency to scan prompts"""
    result = await lockllm.scan(
        input=request.prompt,
        sensitivity="medium"
    )

    if not result.safe:
        raise HTTPException(
            status_code=400,
            detail={
                "error": "Malicious input detected",
                "injection": result.injection,
                "confidence": result.confidence,
                "request_id": result.request_id
            }
        )

    return result

@app.post("/chat")
async def chat(
    request: ChatRequest,
    scan_result = Depends(scan_prompt)
):
    # Request already scanned by dependency
    openai = create_async_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

    response = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": request.prompt}]
    )

    return {
        "response": response.choices[0].message.content,
        "scan_result": {
            "safe": scan_result.safe,
            "request_id": scan_result.request_id
        }
    }

Link to section: Django IntegrationDjango Integration

Integrate with Django middleware:

# middleware.py
from lockllm import LockLLM
import os
import json

class LockLLMMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.lockllm = LockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

    def __call__(self, request):
        if request.method == 'POST' and request.content_type == 'application/json':
            try:
                body = json.loads(request.body)
                if 'prompt' in body:
                    result = self.lockllm.scan(
                        input=body['prompt'],
                        sensitivity="medium"
                    )

                    if not result.safe:
                        from django.http import JsonResponse
                        return JsonResponse({
                            'error': 'Malicious input detected',
                            'details': {
                                'injection': result.injection,
                                'confidence': result.confidence,
                                'request_id': result.request_id
                            }
                        }, status=400)

                    # Attach scan result to request
                    request.scan_result = result
            except Exception:
                pass

        response = self.get_response(request)
        return response

# settings.py
MIDDLEWARE = [
    # ... other middleware
    'yourapp.middleware.LockLLMMiddleware',
]

Link to section: Flask IntegrationFlask Integration

from flask import Flask, request, jsonify
from lockllm import LockLLM
import os

app = Flask(__name__)
lockllm = LockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

@app.before_request
def scan_request():
    if request.method == 'POST' and request.is_json:
        data = request.get_json()
        if 'prompt' in data:
            result = lockllm.scan(
                input=data['prompt'],
                sensitivity="medium"
            )

            if not result.safe:
                return jsonify({
                    'error': 'Malicious input detected',
                    'details': {
                        'injection': result.injection,
                        'confidence': result.confidence,
                        'request_id': result.request_id
                    }
                }), 400

            # Store scan result for the request
            request.scan_result = result

@app.route('/chat', methods=['POST'])
def chat():
    # Request already scanned by before_request
    data = request.get_json()

    openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": data['prompt']}]
    )

    return jsonify({
        'response': response.choices[0].message.content
    })

if __name__ == '__main__':
    app.run()

Link to section: Batch ProcessingBatch Processing

Process multiple requests concurrently with asyncio:

import asyncio
from lockllm import create_async_openai
import os

async def process_prompt(openai, prompt):
    """Process a single prompt"""
    try:
        response = await openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"Error: {str(e)}"

async def batch_process(prompts):
    """Process multiple prompts concurrently"""
    openai = create_async_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

    # Process all prompts concurrently
    tasks = [process_prompt(openai, prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks)

    return results

# Usage
prompts = [
    "What is AI?",
    "Explain machine learning",
    "What is deep learning?"
]

results = asyncio.run(batch_process(prompts))
for i, result in enumerate(results):
    print(f"Prompt {i+1}: {result}\n")

Link to section: Context Manager SupportContext Manager Support

Use context managers for automatic resource cleanup:

Synchronous:

from lockllm import LockLLM
import os

# Context manager ensures proper cleanup
with LockLLM(api_key=os.getenv("LOCKLLM_API_KEY")) as client:
    result = client.scan(input="test prompt")
    print(f"Safe: {result.safe}")

Asynchronous:

from lockllm import AsyncLockLLM
import os
import asyncio

async def main():
    # Async context manager
    async with AsyncLockLLM(api_key=os.getenv("LOCKLLM_API_KEY")) as client:
        result = await client.scan(input="test prompt")
        print(f"Safe: {result.safe}")

asyncio.run(main())

Link to section: Error HandlingError Handling

LockLLM provides typed exceptions for comprehensive error handling:

Link to section: Error TypesError Types

from lockllm import (
    LockLLMError,           # Base error class
    AuthenticationError,    # 401 - Invalid API key
    RateLimitError,         # 429 - Rate limit exceeded
    PromptInjectionError,   # 400 - Malicious input detected
    UpstreamError,          # 502 - Provider API error
    ConfigurationError,     # 400 - Invalid configuration
    NetworkError            # 0 - Network/connection error
)

Link to section: Complete Error HandlingComplete Error Handling

from lockllm import create_openai
from lockllm import (
    PromptInjectionError,
    AuthenticationError,
    RateLimitError,
    UpstreamError,
    ConfigurationError,
    NetworkError
)
import os

openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

try:
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_input}]
    )

    print(response.choices[0].message.content)

except PromptInjectionError as error:
    # Security threat detected
    print("Malicious input detected!")
    print(f"Injection score: {error.scan_result.injection}%")
    print(f"Confidence: {error.scan_result.confidence}%")
    print(f"Request ID: {error.request_id}")

    # Log security incident
    import logging
    logging.warning(f"Prompt injection blocked: {error.request_id}")

    # Return user-friendly error
    return {"error": "Your input could not be processed for security reasons."}

except AuthenticationError as error:
    print("Invalid API key")
    # Check your LOCKLLM_API_KEY environment variable

except RateLimitError as error:
    print("Rate limit exceeded")
    print(f"Retry after (ms): {error.retry_after}")

    # Wait and retry
    if error.retry_after:
        import time
        time.sleep(error.retry_after / 1000)
        # Retry request...

except UpstreamError as error:
    print("Provider API error")
    print(f"Provider: {error.provider}")
    print(f"Status: {error.upstream_status}")
    print(f"Message: {error.message}")

    # Handle provider-specific errors
    if error.provider == 'openai' and error.upstream_status == 429:
        # OpenAI rate limit
        pass

except ConfigurationError as error:
    print(f"Configuration error: {error.message}")
    # Check provider key is added in dashboard

except NetworkError as error:
    print(f"Network error: {error.message}")
    # Check internet connection, firewall, etc.

Link to section: Exponential BackoffExponential Backoff

Implement exponential backoff for transient errors:

import time
from lockllm import RateLimitError, NetworkError

def call_with_backoff(fn, max_retries=5):
    """Call function with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return fn()
        except (RateLimitError, NetworkError) as error:
            if attempt == max_retries - 1:
                raise

            delay = min(1.0 * (2 ** attempt), 30.0)
            print(f"Retry attempt {attempt + 1} after {delay}s")
            time.sleep(delay)

    raise Exception('Max retries exceeded')

# Usage
openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))

response = call_with_backoff(lambda: openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": user_input}]
))

Link to section: Type HintsType Hints

Full type hint support with mypy:

Link to section: Type AnnotationsType Annotations

from lockllm import (
    LockLLM,
    LockLLMConfig,
    ScanRequest,
    ScanResponse,
    ScanResult,
    Sensitivity
)
from typing import Optional

# Configuration types
def create_client(api_key: str, base_url: Optional[str] = None) -> LockLLM:
    return LockLLM(api_key=api_key, base_url=base_url)

# Scan function with type hints
def scan_prompt(text: str, level: Sensitivity = "medium") -> ScanResponse:
    lockllm: LockLLM = LockLLM(api_key="...")
    result: ScanResponse = lockllm.scan(input=text, sensitivity=level)
    return result

# Response types
response: ScanResponse = scan_prompt("test")
is_safe: bool = response.safe
score: float = response.injection
request_id: str = response.request_id

Link to section: mypy Supportmypy Support

The SDK includes a py.typed marker for mypy:

# Run mypy type checking
mypy your_code.py

# Example output:
# your_code.py:10: error: Argument "sensitivity" has incompatible type "str"; expected "Literal['low', 'medium', 'high']"

Link to section: IDE AutocompleteIDE Autocomplete

Full IDE autocomplete with type stubs:

from lockllm import LockLLM

lockllm = LockLLM(api_key="...")

# IDE suggests: scan(...)
lockllm.s  # <-- autocomplete suggestions

# IDE suggests: input, sensitivity, **options
lockllm.scan(i  # <-- autocomplete for parameters

# IDE suggests: "low", "medium", "high"
lockllm.scan(input="test", sensitivity="m  # <-- autocomplete for values

Link to section: API ReferenceAPI Reference

Link to section: ClassesClasses

Link to section: LockLLMLockLLM

Synchronous client class for scanning prompts.

class LockLLM:
    def __init__(
        self,
        api_key: str,
        base_url: Optional[str] = None,
        timeout: Optional[float] = None,
        max_retries: Optional[int] = None
    )

    def scan(
        self,
        input: str,
        sensitivity: Literal["low", "medium", "high"] = "medium",
        **options
    ) -> ScanResponse

Link to section: AsyncLockLLMAsyncLockLLM

Asynchronous client class for scanning prompts.

class AsyncLockLLM:
    def __init__(
        self,
        api_key: str,
        base_url: Optional[str] = None,
        timeout: Optional[float] = None,
        max_retries: Optional[int] = None
    )

    async def scan(
        self,
        input: str,
        sensitivity: Literal["low", "medium", "high"] = "medium",
        **options
    ) -> ScanResponse

Link to section: Data ClassesData Classes

Link to section: LockLLMConfigLockLLMConfig

@dataclass
class LockLLMConfig:
    api_key: str              # Your LockLLM API key (required)
    base_url: Optional[str]   # Custom API endpoint (optional)
    timeout: Optional[float]  # Request timeout in seconds (default: 60.0)
    max_retries: Optional[int]# Max retry attempts (default: 3)

Link to section: ScanResponseScanResponse

@dataclass
class ScanResponse:
    safe: bool                # true if safe, false if malicious
    label: Literal[0, 1]     # 0=safe, 1=malicious
    confidence: float         # Confidence score 0-100
    injection: float          # Injection risk score 0-100
    sensitivity: str          # Sensitivity level used
    request_id: str           # Unique request identifier
    usage: Usage              # Usage statistics
    debug: Optional[Debug]    # Debug info (Pro plan only)

Link to section: FunctionsFunctions

Link to section: Wrapper Functions (Sync)Wrapper Functions (Sync)

def create_openai(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_anthropic(api_key: str, base_url: Optional[str] = None, **kwargs) -> Anthropic
def create_groq(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_deepseek(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_perplexity(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_mistral(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_openrouter(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_together(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_xai(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_fireworks(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_anyscale(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_huggingface(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_gemini(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_cohere(api_key: str, base_url: Optional[str] = None, **kwargs) -> Cohere
def create_azure(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_bedrock(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI
def create_vertex_ai(api_key: str, base_url: Optional[str] = None, **kwargs) -> OpenAI

Link to section: Wrapper Functions (Async)Wrapper Functions (Async)

def create_async_openai(api_key: str, **kwargs) -> AsyncOpenAI
def create_async_anthropic(api_key: str, **kwargs) -> AsyncAnthropic
def create_async_groq(api_key: str, **kwargs) -> AsyncOpenAI
# ... and 14 more async variants

Link to section: Utility FunctionsUtility Functions

def get_proxy_url(provider: str) -> str
def get_all_proxy_urls() -> dict[str, str]

Link to section: Exception ClassesException Classes

class LockLLMError(Exception):
    message: str
    type: str
    code: Optional[str]
    status: Optional[int]
    request_id: Optional[str]

class AuthenticationError(LockLLMError): pass
class RateLimitError(LockLLMError):
    retry_after: Optional[int]
class PromptInjectionError(LockLLMError):
    scan_result: ScanResult
class UpstreamError(LockLLMError):
    provider: Optional[str]
    upstream_status: Optional[int]
class ConfigurationError(LockLLMError): pass
class NetworkError(LockLLMError): pass

Link to section: Best PracticesBest Practices

Link to section: SecuritySecurity

1. Never hardcode API keys - Use environment variables or secret managers
2. Log security incidents - Track blocked requests with request IDs
3. Set appropriate sensitivity - Balance security needs with false positives
4. Handle errors gracefully - Provide user-friendly error messages
5. Monitor request patterns - Watch for attack trends in dashboard
6. Rotate keys regularly - Update API keys periodically
7. Use HTTPS only - Never send API keys over unencrypted connections

Link to section: PerformancePerformance

1. Use wrapper functions - Most efficient integration method
2. Use async for I/O-bound workloads - Better concurrency with AsyncLockLLM
3. Implement caching - Cache LLM responses when appropriate
4. Set reasonable timeouts - Balance user experience with reliability
5. Use connection pooling - SDK handles this automatically with httpx
6. Batch when possible - Group similar requests with asyncio.gather()

Link to section: Async ProgrammingAsync Programming

1. Use async context managers - Ensure proper resource cleanup
2. Avoid blocking calls in async - Don't mix sync and async code
3. Handle event loop properly - Use asyncio.run() or existing loop
4. Be cautious with global state - Async can expose race conditions
5. Use asyncio.gather() for concurrency - Process multiple requests in parallel

Link to section: Production DeploymentProduction Deployment

1. Test sensitivity levels - Validate with real user data
2. Implement monitoring - Track blocked requests and false positives
3. Set up alerting - Get notified of security incidents
4. Review logs regularly - Analyze attack patterns
5. Keep SDK updated - Benefit from latest improvements (pip install -U lockllm)
6. Document incidents - Maintain security incident log
7. Load test - Verify performance under expected load

Link to section: Migration GuidesMigration Guides

Link to section: From Direct API IntegrationFrom Direct API Integration

If you're currently calling LLM APIs directly:

# Before: Direct OpenAI API call
from openai import OpenAI
import os

openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": user_input}]
)

# After: With LockLLM security (one line change)
from lockllm import create_openai
import os

openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))  # Only change

# Everything else stays the same
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": user_input}]
)

Link to section: From OpenAI LibraryFrom OpenAI Library

Minimal changes required:

# Before
from openai import OpenAI
openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# After
from lockllm import create_openai
openai = create_openai(api_key=os.getenv("LOCKLLM_API_KEY"))  # Use LockLLM key

# All other code remains unchanged

Link to section: From Anthropic LibraryFrom Anthropic Library

Minimal changes required:

# Before
from anthropic import Anthropic
anthropic = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# After
from lockllm import create_anthropic
anthropic = create_anthropic(api_key=os.getenv("LOCKLLM_API_KEY"))  # Use LockLLM key

# All other code remains unchanged

Link to section: Async Programming GuideAsync Programming Guide

Link to section: Event Loop ManagementEvent Loop Management

import asyncio
from lockllm import AsyncLockLLM

# Method 1: Using asyncio.run() (recommended for scripts)
async def main():
    lockllm = AsyncLockLLM(api_key="...")
    result = await lockllm.scan(input="test")
    print(result.safe)

asyncio.run(main())

# Method 2: Using existing event loop (for frameworks)
loop = asyncio.get_event_loop()
result = loop.run_until_complete(
    lockllm.scan(input="test")
)

# Method 3: In Jupyter notebooks
await lockllm.scan(input="test")  # Works directly

Link to section: Concurrency PatternsConcurrency Patterns

import asyncio
from lockllm import AsyncLockLLM

async def concurrent_scans():
    lockllm = AsyncLockLLM(api_key="...")

    # Scan multiple prompts concurrently
    prompts = ["prompt1", "prompt2", "prompt3"]

    # Method 1: asyncio.gather()
    results = await asyncio.gather(*[
        lockllm.scan(input=prompt)
        for prompt in prompts
    ])

    # Method 2: asyncio.create_task()
    tasks = [
        asyncio.create_task(lockllm.scan(input=prompt))
        for prompt in prompts
    ]
    results = await asyncio.gather(*tasks)

    return results

Link to section: Resource CleanupResource Cleanup

from lockllm import AsyncLockLLM
import asyncio

# Always use async context managers
async def main():
    async with AsyncLockLLM(api_key="...") as client:
        result = await client.scan(input="test")
        # Client is automatically closed when exiting the block
        return result

asyncio.run(main())

Link to section: TroubleshootingTroubleshooting

Link to section: Common IssuesCommon Issues

"Invalid API key" error (401)

• Verify your LockLLM API key is correct
• Check the key hasn't been revoked in the dashboard
• Ensure you're using your LockLLM key, not your provider key

"No provider API key configured" error (400)

• Add your provider API key (OpenAI, Anthropic, etc.) in the dashboard
• Navigate to Proxy Settings and configure provider keys
• Ensure the provider key is enabled (toggle switch on)

"Could not extract prompt from request" error (400)

• Verify request body format matches provider API spec
• Check you're using the correct SDK version
• Ensure messages array is properly formatted

High latency

• Check your network connection
• Verify LockLLM API status
• Consider adjusting timeout settings
• Review provider API latency

mypy errors

• Ensure Python 3.8+ is installed
• Check peer dependencies are installed (openai, anthropic)
• Run pip install types-requests
• Verify SDK is installed: pip show lockllm

Async event loop issues

• Don't mix sync and async code
• Use asyncio.run() for scripts
• Use existing event loop in frameworks
• Close resources properly with async context managers

Link to section: Debugging TipsDebugging Tips

Enable detailed logging:

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

from lockllm import LockLLM

lockllm = LockLLM(api_key=os.getenv("LOCKLLM_API_KEY"))

try:
    result = lockllm.scan(input=user_prompt)

    print(f"Scan result: safe={result.safe}, injection={result.injection}%, request_id={result.request_id}")
except Exception as error:
    print(f"Error: {type(error).__name__}: {error}")
    if hasattr(error, 'request_id'):
        print(f"Request ID: {error.request_id}")

Link to section: Getting HelpGetting Help

• Documentation: https://www.lockllm.com/docs
• GitHub Issues: https://github.com/lockllm/lockllm-pip/issues
• Email Support: [email protected]
• Status Page: status.lockllm.com

Link to section: FAQFAQ

Link to section: How do I install the SDK?How do I install the SDK?

Install using pip, poetry, or pipenv:

pip install lockllm
poetry add lockllm
pipenv install lockllm

The SDK requires Python 3.8+ and works with Python 3.8, 3.9, 3.10, 3.11, and 3.12.

Link to section: Does the SDK work as a drop-in replacement for OpenAI and Anthropic?Does the SDK work as a drop-in replacement for OpenAI and Anthropic?

Yes. Use create_openai() or create_anthropic() to get wrapped clients that work exactly like the official SDKs. All methods, streaming, and function calling are supported. Prompts are automatically scanned before being sent to the provider.

Link to section: What's the difference between sync and async?What's the difference between sync and async?

The SDK provides both synchronous (LockLLM, create_openai) and asynchronous (AsyncLockLLM, create_async_openai) APIs. Use async for I/O-bound workloads and better concurrency. Use sync for simple scripts and synchronous applications.

Link to section: What type hint support is available?What type hint support is available?

The SDK includes full type hints with a py.typed marker for mypy. It supports mypy strict mode, provides IDE autocomplete, and includes type stubs for all APIs. Python 3.8+ type hints are used throughout.

Link to section: Which AI providers are supported?Which AI providers are supported?

17+ providers are supported with both sync and async variants: OpenAI, Anthropic, Groq, DeepSeek, Perplexity, Mistral, OpenRouter, Together AI, xAI (Grok), Fireworks AI, Anyscale, Hugging Face, Google Gemini, Cohere, Azure OpenAI, AWS Bedrock, Google Vertex AI. All providers support custom endpoint URLs for self-hosted and private deployments.

Link to section: How do I handle errors?How do I handle errors?

The SDK provides 7 typed exception classes: AuthenticationError, RateLimitError, PromptInjectionError, UpstreamError, ConfigurationError, NetworkError, and base LockLLMError. Use try-except blocks with specific exception types for proper error handling.

Link to section: Does the SDK support streaming?Does the SDK support streaming?

Yes. All provider wrappers fully support streaming responses in both sync and async modes. Use stream=True in your requests and iterate with for (sync) or async for (async) loops.

Link to section: Is it really free?Is it really free?

Yes. LockLLM is completely free with unlimited usage and no rate limits for the free tier (1,000 requests/minute). You bring your own provider API keys (BYOK model), so you only pay your provider (OpenAI, Anthropic, etc.) for LLM usage.

Link to section: GitHub RepositoryGitHub Repository

View the source code, report issues, and contribute:

Repository: https://github.com/lockllm/lockllm-pip

PyPI Package: https://pypi.org/project/lockllm/

Link to section: Next StepsNext Steps

• View Python SDK integration page for more examples
• Read API Reference for REST API details
• Explore Best Practices for production deployments
• Check out Proxy Mode for alternative integration
• Configure Webhooks for security alerts
• Browse Dashboard documentation