groq-rotator

A drop-in async Groq client that automatically rotates API keys when you hit rate limits. Useful when you have multiple free-tier Groq keys and don't want your pipeline to crash.

Install

pip install groq-rotator

Usage

Use it exactly like the normal AsyncGroq client:

import asyncio
from groq_rotator import AsyncRotatingGroq

client = AsyncRotatingGroq(api_keys=["key1", "key2", "key3"])

async def main():
    response = await client.chat.completions.create(
        model="llama3-70b-8192",
        messages=[{"role": "user", "content": "Hello"}]
    )
    print(response.choices[0].message.content)

asyncio.run(main())

That's it. No extra setup. If key1 hits a rate limit, it silently switches to key2 and retries.

Parallel requests (main use case)

import asyncio
from groq_rotator import AsyncRotatingGroq

client = AsyncRotatingGroq(api_keys=["key1", "key2", "key3"])

async def process(prompt: str):
    response = await client.chat.completions.create(
        model="llama3-70b-8192",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

async def main():
    prompts = ["Summarize X", "Classify Y", "Extract Z"]
    results = await asyncio.gather(*[process(p) for p in prompts])
    print(results)

asyncio.run(main())

FastAPI

from fastapi import FastAPI
from groq_rotator import AsyncRotatingGroq

app = FastAPI()
client = AsyncRotatingGroq(api_keys=["key1", "key2", "key3"])

@app.post("/chat")
async def chat(prompt: str):
    response = await client.chat.completions.create(
        model="llama3-70b-8192",
        messages=[{"role": "user", "content": prompt}]
    )
    return {"response": response.choices[0].message.content}

Optional: know when a key rotates

def on_rotate(index: int):
    print(f"Switched to key index {index}")

client = AsyncRotatingGroq(api_keys=["key1", "key2"], on_rotate=on_rotate)

When is this useful?

• You have multiple free-tier Groq keys and want to maximize throughput
• You're running batch jobs or multi-agent pipelines with heavy Groq usage
• You're in a FastAPI or async backend and can't afford a crash on rate limit
• You want rotation to be invisible — no changes to how you write Groq code

When you don't need this

• You have a paid Groq key with high rate limits
• You're making one-off requests in a script
• You're not using async

Error handling

If all keys are rate limited, it raises groq.RateLimitError so you can handle it yourself:

from groq import RateLimitError

try:
    response = await client.chat.completions.create(...)
except RateLimitError:
    print("All keys exhausted")