mirror of https://github.com/kortix-ai/suna.git
180 lines
7.2 KiB
Python
180 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Suna Default Agent Management Script (Simplified)
|
||
|
||
This script provides administrative functions for managing Suna default agents across all users.
|
||
|
||
Usage:
|
||
# 🚀 MAIN COMMANDS
|
||
python manage_suna_agents.py install-all # Install Suna for all users who don't have it
|
||
python manage_suna_agents.py stats # Show Suna agent statistics
|
||
python manage_suna_agents.py install-user <id> # Install Suna for specific user
|
||
|
||
Examples:
|
||
python manage_suna_agents.py install-all
|
||
python manage_suna_agents.py stats
|
||
python manage_suna_agents.py install-user 123e4567-e89b-12d3-a456-426614174000
|
||
|
||
Note: Sync is no longer needed - Suna agents automatically use the current configuration from config.py
|
||
"""
|
||
|
||
import asyncio
|
||
import argparse
|
||
import sys
|
||
import json
|
||
from pathlib import Path
|
||
|
||
# Add the backend directory to the path so we can import modules
|
||
backend_dir = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(backend_dir))
|
||
|
||
from utils.suna_default_agent_service import SunaDefaultAgentService
|
||
from services.supabase import DBConnection
|
||
from utils.logger import logger
|
||
|
||
|
||
class SunaAgentManager:
|
||
def __init__(self):
|
||
self.service = SunaDefaultAgentService()
|
||
|
||
async def install_all_users(self):
|
||
"""Install Suna agent for all users who don't have it"""
|
||
print("🚀 Installing Suna default agent for all users who don't have it...")
|
||
|
||
result = await self.service.install_for_all_users()
|
||
|
||
print(f"✅ Installation completed!")
|
||
print(f" 📦 Installed: {result['installed_count']}")
|
||
print(f" ❌ Failed: {result['failed_count']}")
|
||
|
||
if result['failed_count'] > 0:
|
||
print("\n❌ Failed installations:")
|
||
for detail in result['details']:
|
||
if detail['status'] == 'failed':
|
||
print(f" - User {detail['account_id']}: {detail.get('error', 'Unknown error')}")
|
||
|
||
if result['installed_count'] > 0:
|
||
print(f"\n✅ Successfully installed Suna for {result['installed_count']} users")
|
||
|
||
async def update_config_info(self):
|
||
"""Show information about Suna configuration (no sync needed)"""
|
||
print("ℹ️ Suna Configuration Information")
|
||
print("=" * 50)
|
||
print("🔧 Suna agents automatically use the current configuration from config.py")
|
||
print("📝 No sync needed - changes are applied immediately when agents run")
|
||
print("💡 To update Suna behavior, simply modify backend/agent/suna/config.py")
|
||
print("\n✅ All Suna agents are always up-to-date with your latest configuration!")
|
||
|
||
async def install_user(self, account_id):
|
||
"""Install Suna agent for specific user"""
|
||
print(f"🚀 Installing Suna default agent for user {account_id}...")
|
||
|
||
agent_id = await self.service.install_suna_agent_for_user(account_id)
|
||
|
||
if agent_id:
|
||
print(f"✅ Successfully installed Suna agent {agent_id} for user {account_id}")
|
||
else:
|
||
print(f"❌ Failed to install Suna agent for user {account_id}")
|
||
|
||
async def replace_user_agent(self, account_id):
|
||
"""Replace Suna agent for specific user (in case of corruption)"""
|
||
print(f"🔄 Replacing Suna agent for user {account_id}...")
|
||
|
||
# Install/replace the agent with latest config
|
||
agent_id = await self.service.install_suna_agent_for_user(account_id, replace_existing=True)
|
||
|
||
if agent_id:
|
||
print(f"✅ Successfully replaced Suna agent {agent_id} for user {account_id}")
|
||
else:
|
||
print(f"❌ Failed to replace Suna agent for user {account_id}")
|
||
|
||
async def show_stats(self):
|
||
"""Show Suna agent statistics"""
|
||
print("📊 Suna Default Agent Statistics")
|
||
print("=" * 50)
|
||
|
||
stats = await self.service.get_suna_agent_stats()
|
||
|
||
if 'error' in stats:
|
||
print(f"❌ Error getting stats: {stats['error']}")
|
||
return
|
||
|
||
print(f"Total Agents: {stats.get('total_agents', 0)}")
|
||
print(f"Active Agents: {stats.get('active_agents', 0)}")
|
||
print(f"Inactive Agents: {stats.get('inactive_agents', 0)}")
|
||
|
||
version_dist = stats.get('version_distribution', {})
|
||
if version_dist:
|
||
print(f"\nVersion Distribution:")
|
||
for version, count in version_dist.items():
|
||
print(f" {version}: {count} agents")
|
||
|
||
creation_dates = stats.get('creation_dates', {})
|
||
if creation_dates:
|
||
print(f"\nCreation Dates (Last 12 months):")
|
||
for month, count in sorted(creation_dates.items(), reverse=True):
|
||
print(f" {month}: {count} agents")
|
||
|
||
|
||
async def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="Manage Suna default agents across all users (Simplified)",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=__doc__
|
||
)
|
||
|
||
subparsers = parser.add_subparsers(dest='command', help='Available commands')
|
||
|
||
# Main commands
|
||
subparsers.add_parser('install-all', help='Install Suna agent for all users who don\'t have it')
|
||
subparsers.add_parser('stats', help='Show Suna agent statistics')
|
||
subparsers.add_parser('config-info', help='Show information about Suna configuration')
|
||
|
||
# User-specific commands
|
||
install_user_parser = subparsers.add_parser('install-user', help='Install Suna agent for specific user')
|
||
install_user_parser.add_argument('account_id', help='Account ID to install Suna for')
|
||
|
||
replace_user_parser = subparsers.add_parser('replace-user', help='Replace Suna agent for specific user (if corrupted)')
|
||
replace_user_parser.add_argument('account_id', help='Account ID to replace Suna for')
|
||
|
||
# Legacy commands (deprecated but still functional)
|
||
subparsers.add_parser('sync', help='[DEPRECATED] No longer needed - config is always current')
|
||
subparsers.add_parser('update-all', help='[DEPRECATED] No longer needed - config is always current')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not args.command:
|
||
parser.print_help()
|
||
return
|
||
|
||
manager = SunaAgentManager()
|
||
|
||
try:
|
||
if args.command == 'install-all':
|
||
await manager.install_all_users()
|
||
elif args.command == 'stats':
|
||
await manager.show_stats()
|
||
elif args.command == 'config-info':
|
||
await manager.update_config_info()
|
||
elif args.command == 'install-user':
|
||
await manager.install_user(args.account_id)
|
||
elif args.command == 'replace-user':
|
||
await manager.replace_user_agent(args.account_id)
|
||
elif args.command == 'sync':
|
||
print("⚠️ DEPRECATED: Sync is no longer needed!")
|
||
await manager.update_config_info()
|
||
elif args.command == 'update-all':
|
||
print("⚠️ DEPRECATED: Update-all is no longer needed!")
|
||
await manager.update_config_info()
|
||
else:
|
||
parser.print_help()
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n⚠️ Operation cancelled by user")
|
||
except Exception as e:
|
||
print(f"❌ Error: {str(e)}")
|
||
logger.error(f"Script error: {str(e)}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |