A robust, async-first Python package for managing, rate-limiting, rotating, securely storing, and tracking usage of API keys across multiple providers (OpenAI, Anthropic, Google, and many more). Designed for LLM agent servers that need reliable, high-volume API key management.

✨ Core Features

📦 Installation

The package is published on PyPI.

[!TIP] Recommended: Install using uv for lightning-fast dependency resolution.

uv add api-service-handler

Or using pip:

pip install api-service-handler

# For MongoDB support (requires MongoDB 4.2+)
uv add "api-service-handler[mongodb]"

# For PostgreSQL support
uv add "api-service-handler[postgresql]"

# For SQLite support
uv add "api-service-handler[sqlite]"

# Everything at once
uv add "api-service-handler[all]"

⚙️ Configuration & Initialization

The main entry point to the library is the APIServiceHandler class. It manages the connection to your underlying database, loads the encryption settings, and serves as the facade for all operations.

import asyncio
from api_service_handler.client import APIServiceHandler

async def main():
    handler = APIServiceHandler(
        storage_backend="sqlite",                   # memory, sqlite, mongodb, postgresql
        connection_string="sqlite:///api_keys.db",  # Path or URI to database
        encrypt_keys=True,                          # Strongly recommended
        shared_secret="your-super-secret-32-byte-key-here!!!", # Used for AES-GCM
        rotation_strategy="round_robin",            # round_robin, least_used, random, weighted
        auto_reset_counters=True,                   # Auto-refresh quotas on day/month rollover
        soft_delete=True                            # Keep deleted keys as "revoked" for auditing
    )
    
    # ⚠️ You MUST initialize the handler to establish db connections
    await handler.initialize()
    
    print("🚀 API Service Handler is ready!")
    
    # Safely close connections before your app shuts down
    await handler.close()

if __name__ == "__main__":
    asyncio.run(main())

[!NOTE] Environment Variables: You can configure the client purely using environment variables (ASH_STORAGE_BACKEND, ASH_CONNECTION_STRING, ASH_SHARED_SECRET, ASH_ROTATION_STRATEGY).

🔑 Managing API Keys

Adding a Key

Add keys dynamically with precise controls over their environments and usage thresholds.

from api_service_handler.models import Environment

key = await handler.add_key(
    provider="openai",               # Supported provider enum string
    key_value="sk-proj-1234567890",  # The raw API key
    alias="prod-gpt4-key",           # Friendly name
    daily_limit=1000,                # Max 1000 uses per day
    monthly_limit=25000,             # Max 25000 uses per month
    max_concurrent=5,                # Max 5 active parallel requests
    environment=Environment.PRODUCTION,
    tags=["premium", "gpt-4"],
    metadata={"billing_account": "acct_123"},
    weight=10                        # Useful if using "weighted" rotation
)

Retrieving & Filtering

Keys are retrieved as APIKey Pydantic models. By default, key_value is encrypted at rest and will remain masked unless explicitly requested.

# Get a specific key (Decrypted)
key = await handler.get_key(key_id="uuid-string", decrypt=True)

# Complex Filtering
filtered_keys = await handler.get_all_keys(
    provider="google_gemini",
    tags=["premium"],
    environment="production",
    has_capacity=True # ✨ Magic! Only returns keys that haven't hit rate limits
)

🎯 Metadata Filtering (Deep Search)

You can attach arbitrary JSON metadata to any key and query against it. All storage backends (including Memory & SQLite) natively support filtering by nested metadata using dot-notation!

# Returns only keys assigned to the engineering team
engineering_keys = await handler.get_all_keys(
    metadata_filter={"team": "engineering"}
)

# Nested JSON matching using dot-notation!
enterprise_keys = await handler.get_all_keys(
    metadata_filter={"billing.tier": "enterprise"}
)

# You can even route requests strictly based on metadata!
async with handler.use_key(provider="openai", metadata_filter={"project_id": "proj-789"}) as p_key:
    pass

♻️ Using Keys (Rotation & Rate Limiting)

The core purpose of this library is to safely dispense API keys while preventing you from hitting provider rate limits.

The safest and easiest way to retrieve a key is using the use_key context manager. It automatically:

1. Filters out keys that have hit their Daily/Monthly limits.
2. Filters out keys that have hit their Max Concurrent limit.
3. Rotates between valid keys based on your strategy.
4. Safely tracks parallel execution blocks and historical usage.

from api_service_handler.exceptions import RateLimitExceededError, NoAvailableKeyError, MaxConcurrentExceededError

async def generate_text(prompt: str):
    try:
        # Request a key for Anthropic that is designated for production
        async with handler.use_key(provider="anthropic", environment="production") as api_key:
            
            # The key is automatically decrypted and ready to use
            raw_key = api_key.key_value
            print(f"Executing request with {api_key.alias}")
            
            # response = await my_llm_client.chat(api_key=raw_key, messages=prompt)
            
    except NoAvailableKeyError:
        print("❌ No active Anthropic keys found!")
    except RateLimitExceededError as e:
        print(f"🛑 All keys are rate limited! {e}")
    except MaxConcurrentExceededError as e:
        print(f"⚠️ Too many parallel requests active right now! {e}")

📊 Usage Tracking & Statistics

Track exactly how much your APIs are being utilized. Counters auto-reset at the start of a new calendar day/month based on UTC time.

# Get usage for a specific key
stats = await handler.get_usage_stats(key.id)
print(f"Daily remaining: {stats.daily_remaining}")

# Get aggregated usage across all keys for a provider
provider_stats = await handler.get_provider_stats("openai")
for stat in provider_stats:
    print(f"{stat.alias}: {stat.daily_usage_count} uses today")

💻 CLI Tool (ash)

A full-featured command-line utility is bundled with the package for administrative tasks.

[!IMPORTANT] Ensure you have your environment variables exported so the CLI connects to your production database! export ASH_CONNECTION_STRING="postgresql://user:pass@localhost/db"

🌐 Supported Providers

The library enforces strict string enums for API providers. You can use any of the following (case-insensitive):

📋 Changelog

v0.1.7 — Bug fixes & hardening

Breaking changes

1. decrypt_api_key now raises on failure

Previously, decrypting with a wrong secret silently returned the raw ciphertext. This masked misconfigurations and caused mysterious 401s downstream. Now it raises EncryptionError.

# Before (v0.1.6): wrong secret silently returned ciphertext — no error
# After  (v0.1.7): raises EncryptionError

from api_service_handler.exceptions import EncryptionError

try:
    plain = decrypt_api_key(value, secret)
except EncryptionError:
    # bad secret or corrupted value — handle explicitly

If you only use APIServiceHandler and never call decrypt_api_key directly, no changes are needed — the exception propagates as expected.

2. bulk_add_keys now respects 0 for rate limits

Previously, passing daily_limit=0, monthly_limit=0, or max_concurrent=0 to bulk_add_keys was silently ignored (treated as "not set") and the config default was applied instead. This matched neither the caller's intent nor the behaviour of add_key. Now 0 is respected as-is.

# Before: 0 was replaced by default_daily_limit=1000 → key got limit=1000
# After:  0 is kept → key is effectively blocked

await handler.bulk_add_keys([{"provider": "openai", "key_value": "sk-...", "daily_limit": 0}])

If you relied on passing 0 to inherit the config default, change those calls to pass None instead.

3. MongoDB backend requires MongoDB 4.2+

Concurrent-usage tracking now uses an atomic aggregation pipeline update, which requires MongoDB 4.2 or newer. If you are on an older version, upgrade before updating the package.

mongosh --eval "db.version()"

Bug fixes (no action required)

• Concurrency race fixed — max_concurrent enforcement is now atomic; the limit can no longer be exceeded under parallel load.
• has_capacity=False filter — SQLite backend now correctly returns only exhausted keys when has_capacity=False is passed (previously the filter was silently ignored).
• MongoDB empty-document crash — _dict_to_key now raises instead of returning None, preventing silent AttributeError crashes on malformed documents.
• Decrypt guard correctness — get_key, get_all_keys, get_keys_by_provider, get_next_key, and use_key now only attempt decryption when encrypt_keys=True, preventing accidental decryption of plaintext-stored keys when a shared_secret is configured but encryption is disabled.