How to Implement Rate Limiting with CrewAI (Step by Step)
We’re building a rate limiting solution for CrewAI that not only keeps your application within safe operational bounds but also enhances the user experience. Knowing how to crewai implement rate limiting is crucial for applications that scale.
Prerequisites
- Python 3.11+
- crewAI version 1.0 or higher
- Redis (for storing rate limit data)
- Flask (to set up a simple web server)
Step-by-Step
Step 1: Set Up Your Environment
# First, install the necessary packages
pip install crewai flask redis
This command sets up a basic environment for working with CrewAI. You need both Flask for handling requests and Redis for storing counters. If you don’t have Redis installed, expect errors during runtime because rate limiting relies on a persistent data store. Missed that? It’s a classic rookie mistake.
Step 2: Initialize Redis
import redis
# Create a connection to the Redis server
redis_client = redis.Redis(host='localhost', port=6379, db=0)
Initializing Redis is essential. Here, we’re connecting to the local instance of Redis. If your server isn’t running, you’ll run into connection errors. Make sure to have Redis up before proceeding.
Step 3: Create the Rate Limiting Logic
def is_rate_limited(user_id):
# Define the rate limit and timeframe here
rate_limit = 5 # Maximum requests
timeframe = 60 # In seconds
pipeline = redis_client.pipeline()
current_time = redis_client.time()[0] # Use Redis server time
request_times = redis_client.lrange(user_id, 0, -1)
# Remove timestamps that are outside the timeframe
for time in request_times:
if current_time - int(time) > timeframe:
pipeline.lrem(user_id, 1, time)
# Add current request time
pipeline.rpush(user_id, current_time)
pipeline.expire(user_id, timeframe) # Ensure data expiry to free up memory
pipeline.execute()
return len(redis_client.lrange(user_id, 0, -1)) > rate_limit
In this function, we check if the user’s request count exceeds the allowed limit. The code efficiently manages timestamps in a Redis list and removes any entries outside the allowed timeframe. Missing or misunderstanding the data structure could crash the service.
Step 4: Integrate the Rate Limiting in Your Flask Application
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/api/resource', methods=['GET'])
def get_resource():
user_id = request.headers.get('X-User-ID') # Get user ID from request header
if is_rate_limited(user_id):
return jsonify({'error': 'Rate limit exceeded. Try again later.'}), 429
# Handle the request
return jsonify({'message': 'Success!'})
if __name__ == '__main__':
app.run(debug=True)
This snippet sets up a basic API route that checks rate limits before processing requests. You can send requests with a custom header for identification. Forgetting to include the header will send your requests into a denial-of-service pattern — nothing like a self-inflicted headache!
Step 5: Test Your Implementation
# Make concurrent requests to test rate limits
for i in {1..10}; do
curl -H "X-User-ID: 12345" http://localhost:5000/api/resource &
done
wait
Running this loop sends 10 rapid requests to your Flask app. You should notice that, after the rate limit is exceeded, you get a 429 error. If not, check your Redis setup or rate limiting logic. A misconfiguration will make all your hard work useless — like my old blog posts that nobody reads anymore.
The Gotchas
- Data Expiry: If you don’t set expiry for your Redis entries, you’ll clutter the store with old data, leading to inaccurate rate limit checking.
- Concurrent Requests: Handling simultaneous requests from the same user can lead to unexpected behavior unless properly managed with atomic operations.
- Testing Limits: Be cautious when testing rate limits; a flood of requests can degrade service for everyone else, especially in production.
- Logging Rate Limit Events: Failing to log when users hit their limits can make troubleshooting challenging.
- Rate Limit Adjustments: Hard-coding rate limits can backfire; allow dynamic configuration through an admin dashboard.
Full Code
import redis
from flask import Flask, request, jsonify
# Initialize the Redis client
redis_client = redis.Redis(host='localhost', port=6379, db=0)
# Function to check if a user is rate limited
def is_rate_limited(user_id):
rate_limit = 5
timeframe = 60
pipeline = redis_client.pipeline()
current_time = redis_client.time()[0]
request_times = redis_client.lrange(user_id, 0, -1)
for time in request_times:
if current_time - int(time) > timeframe:
pipeline.lrem(user_id, 1, time)
pipeline.rpush(user_id, current_time)
pipeline.expire(user_id, timeframe)
pipeline.execute()
return len(redis_client.lrange(user_id, 0, -1)) > rate_limit
# Create a Flask application
app = Flask(__name__)
# Define API endpoint
@app.route('/api/resource', methods=['GET'])
def get_resource():
user_id = request.headers.get('X-User-ID')
if is_rate_limited(user_id):
return jsonify({'error': 'Rate limit exceeded. Try again later.'}), 429
return jsonify({'message': 'Success!'})
if __name__ == '__main__':
app.run(debug=True)
What’s Next
Consider implementing a logging mechanism for monitoring user behavior and adjusting your rate limits. This can help in fine-tuning your application’s performance under load.
FAQ
- What happens if Redis goes down? If your Redis instance fails, your rate limiting will break. You might want to implement a fallback or alert system.
- How can I adjust rate limits on the fly? Store rate limit configurations in a database and read them at application startup or periodically.
- Can I apply different rate limits to different users? Absolutely. You can modify the `is_rate_limited` function to read user roles or attributes and set limits accordingly.
Data Sources
Last updated April 09, 2026. Data sourced from official docs and community benchmarks.
🕒 Published: