MCP with Gemini Integration

#!/usr/bin/env python3 """ Rate Limit Monitoring Script for Gemini MCP Integration. Provides real-time monitoring of API usage and rate limiting status. """ import asyncio import json import time from datetime import datetime, timedelta from pathlib import Path import sys # Add src to path for imports sys.path.append(str(Path(__file__).parent / "src")) import sys import os sys.path.append(os.path.join(os.path.dirname(__file__), '..')) from src.utils.persistent_rate_limiter import persistent_rate_limiter def print_status(): """Print current rate limiting status.""" status = persistent_rate_limiter.get_status() print("\n" + "="*60) print("GEMINI API RATE LIMITING STATUS") print("="*60) print(f"Database: {status['database_path']}") print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print() # Current usage print("CURRENT USAGE:") print(f" Requests per minute: {status['current_rpm']}/{status['safe_rpm']}") print(f" Requests per day: {status['current_rpd']}/{status['safe_rpd']}") print(f" Tokens per minute: {status['current_tpm']}/{status['safe_tpm']}") print() # Available capacity print("AVAILABLE CAPACITY:") print(f" RPM available: {status['rpm_available']}") print(f" RPD available: {status['rpd_available']}") print(f" TPM available: {status['tpm_available']}") print() # Usage percentages rpm_percent = (status['current_rpm'] / status['safe_rpm']) * 100 if status['safe_rpm'] > 0 else 0 rpd_percent = (status['current_rpd'] / status['safe_rpd']) * 100 if status['safe_rpd'] > 0 else 0 tpm_percent = (status['current_tpm'] / status['safe_tpm']) * 100 if status['safe_tpm'] > 0 else 0 print("USAGE PERCENTAGES:") print(f" RPM: {rpm_percent:.1f}%") print(f" RPD: {rpd_percent:.1f}%") print(f" TPM: {tpm_percent:.1f}%") print("="*60) def print_usage_history(hours: int = 24): """Print usage history for the specified number of hours.""" history = persistent_rate_limiter.get_usage_history(hours) if "error" in history: print(f"Error getting usage history: {history['error']}") return print(f"\n" + "="*60) print(f"USAGE HISTORY (Last {hours} hours)") print("="*60) print(f"Total requests: {history['total_requests']}") print(f"Total tokens: {history['total_tokens'] or 0:,}") print() if history['hourly_breakdown']: print("HOURLY BREAKDOWN:") print(f"{'Hour':<20} {'Requests':<10} {'Tokens':<15}") print("-" * 45) for entry in history['hourly_breakdown']: hour = entry['hour'] requests = entry['requests'] tokens = entry['tokens'] or 0 print(f"{hour:<20} {requests:<10} {tokens:<15,}") else: print("No usage data in the specified time period.") print("="*60) def print_detailed_analysis(): """Print detailed analysis of rate limiting patterns.""" print("\n" + "="*60) print("DETAILED ANALYSIS") print("="*60) # Get status for analysis status = persistent_rate_limiter.get_status() # Calculate rates current_time = time.time() # Check if we're approaching limits rpm_warning = status['current_rpm'] >= status['safe_rpm'] * 0.8 rpd_warning = status['current_rpd'] >= status['safe_rpd'] * 0.8 tpm_warning = status['current_tpm'] >= status['safe_tpm'] * 0.8 print("LIMIT WARNINGS:") if rpm_warning: print(f" ⚠️ RPM approaching limit: {status['current_rpm']}/{status['safe_rpm']}") else: print(f" ✅ RPM within safe range: {status['current_rpm']}/{status['safe_rpm']}") if rpd_warning: print(f" ⚠️ RPD approaching limit: {status['current_rpd']}/{status['safe_rpd']}") else: print(f" ✅ RPD within safe range: {status['current_rpd']}/{status['safe_rpd']}") if tpm_warning: print(f" ⚠️ TPM approaching limit: {status['current_tpm']}/{status['safe_tpm']}") else: print(f" ✅ TPM within safe range: {status['current_tpm']}/{status['safe_tpm']}") print() # Recommendations print("RECOMMENDATIONS:") if rpm_warning or rpd_warning or tpm_warning: print(" 🚨 Consider reducing API call frequency") print(" 📊 Monitor usage patterns more closely") print(" ⏰ Implement longer delays between requests") else: print(" ✅ Current usage is within safe limits") print(" 📈 You can safely make more API calls") print("="*60) async def monitor_realtime(interval: int = 30): """Monitor rate limiting in real-time.""" print(f"\nStarting real-time monitoring (updates every {interval} seconds)") print("Press Ctrl+C to stop") try: while True: print_status() await asyncio.sleep(interval) except KeyboardInterrupt: print("\nMonitoring stopped.") def main(): """Main function for the monitoring script.""" import argparse parser = argparse.ArgumentParser(description="Monitor Gemini API rate limiting") parser.add_argument("--status", action="store_true", help="Show current status") parser.add_argument("--history", type=int, default=24, help="Show usage history for N hours (default: 24)") parser.add_argument("--analysis", action="store_true", help="Show detailed analysis") parser.add_argument("--monitor", type=int, default=30, help="Monitor in real-time with N second intervals") args = parser.parse_args() if args.status: print_status() if args.history: print_usage_history(args.history) if args.analysis: print_detailed_analysis() if args.monitor: asyncio.run(monitor_realtime(args.monitor)) # If no specific action, show status if not any([args.status, args.history, args.analysis, args.monitor]): print_status() if __name__ == "__main__": main()