Migrate elevenlabs agents to new workspace

migrate_agent.py

import os
import requests
import json
import copy
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

OLD_ELEVENLABS_API_KEY = os.getenv("OLD_ELEVENLABS_API_KEY")
NEW_ELEVENLABS_API_KEY = os.getenv("NEW_ELEVENLABS_API_KEY")

BASE_URL = "https://api.elevenlabs.io/v1/convai"

def check_api_keys():
    """Validates that API keys are loaded."""
    if not OLD_ELEVENLABS_API_KEY:
        print("🚨 Error: OLD_ELEVENLABS_API_KEY not found in .env file.")
        return False
    if not NEW_ELEVENLABS_API_KEY:
        print("🚨 Error: NEW_ELEVENLABS_API_KEY not found in .env file.")
        return False
    print("✅ API keys loaded successfully.")
    return True

def list_all_agents(api_key, workspace_name="Old"):
    """Fetches all agent summaries from the specified workspace."""
    agents = []
    cursor = None
    headers = {
        "xi-api-key": api_key,
        "Content-Type": "application/json"
    }
    
    page_num = 1
    while True:
        url = f"{BASE_URL}/agents"
        params = {}
        if cursor:
            params["cursor"] = cursor
        
        print(f"⏳ Fetching page {page_num} of agents from {workspace_name} workspace...")
        try:
            response = requests.get(url, headers=headers, params=params)
            response.raise_for_status()  # Raises an HTTPError for bad responses (4XX or 5XX)
            data = response.json()
            
            current_page_agents = data.get("agents", [])
            if not current_page_agents and page_num == 1 and not data.get("has_more"):
                print(f"ℹ️ No agents found in {workspace_name} workspace.")
                return [] # Return empty list if no agents at all

            agents.extend(current_page_agents)
            
            if data.get("has_more") and data.get("next_cursor"):
                cursor = data["next_cursor"]
                page_num += 1
            else:
                break
        except requests.exceptions.RequestException as e:
            print(f"🚨 Error listing agents from {workspace_name} workspace: {e}")
            if response is not None:
                print(f"Response content: {response.text}")
            return None # Indicate failure
            
    print(f"✅ Found {len(agents)} agent(s) in {workspace_name} workspace.")
    return agents

def get_agent_details(api_key, agent_id, workspace_name="Old"):
    """Fetches detailed configuration for a specific agent."""
    headers = {
        "xi-api-key": api_key,
        "Content-Type": "application/json"
    }
    url = f"{BASE_URL}/agents/{agent_id}"
    
    print(f"⏳ Fetching details for agent {agent_id} from {workspace_name} workspace...")
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"🚨 Error getting details for agent {agent_id} from {workspace_name} workspace: {e}")
        if response is not None:
            print(f"Response content: {response.text}")
        return None

def create_agent(api_key, agent_payload, workspace_name="New"):
    """Creates an agent in the specified workspace."""
    headers = {
        "xi-api-key": api_key,
        "Content-Type": "application/json"
    }
    url = f"{BASE_URL}/agents/create"
    
    agent_name_for_log = agent_payload.get("name", "Unnamed Agent")
    print(f"⏳ Attempting to create agent '{agent_name_for_log}' in {workspace_name} workspace...")
    
    try:
        response = requests.post(url, headers=headers, json=agent_payload)
        response.raise_for_status()
        created_agent_data = response.json()
        print(f"✅ Successfully created agent '{agent_name_for_log}' in {workspace_name} workspace. New Agent ID: {created_agent_data.get('agent_id')}")
        return created_agent_data
    except requests.exceptions.RequestException as e:
        print(f"🚨 Error creating agent '{agent_name_for_log}' in {workspace_name} workspace: {e}")
        if response is not None:
            print(f"Response status code: {response.status_code}")
            try:
                print(f"Response content: {response.json()}") # Try to print JSON error
            except json.JSONDecodeError:
                print(f"Response content: {response.text}") # Fallback to text
        return None

def prepare_payload_for_create(agent_details):
    """
    Prepares the payload for the create agent endpoint from fetched agent details.
    The payload for creating an agent should only include:
    - conversation_config (Required)
    - name (Optional)
    - tags (Optional)
    - platform_settings (Optional)
    """
    if not agent_details:
        return None

    payload = {}

    payload["conversation_config"] = agent_details["conversation_config"] #required

    if "name" in agent_details:
        payload["name"] = agent_details["name"]
    
    if "tags" in agent_details:
        payload["tags"] = agent_details["tags"] # Ensure it's a list of strings as per API

    if "platform_settings" in agent_details:
        platform_settings = agent_details["platform_settings"]
                
    return payload

def main():
    print("🚀 Starting ElevenLabs Agent Migration Script...")
    if not check_api_keys():
        return

    # --- Step 1: List all agents from the Old Workspace ---
    print("\n--- Step 1: Listing agents from Old Workspace ---")
    old_agents_summary = list_all_agents(OLD_ELEVENLABS_API_KEY, "Old")
    
    if old_agents_summary is None: # Indicates an error occurred during listing
        print("❌ Failed to list agents from the old workspace due to an error. Exiting.")
        return
    if not old_agents_summary: # Empty list, but no error
        print("🏁 No agents found in the old workspace to migrate.")
        return

    # --- Step 2: Migrate Agents ---
    print("\n--- Step 2: Migrating Agents to New Workspace ---")
    successful_migrations = 0
    failed_migrations = 0

    for agent_summary in old_agents_summary:
        agent_id_old = agent_summary.get("agent_id")
        agent_name_old = agent_summary.get("name", f"Unnamed Agent (ID: {agent_id_old})")
        print(f"\n🔄 Processing agent: '{agent_name_old}' (Old ID: {agent_id_old})")

        # 2a. Get full agent details from the old workspace
        agent_details_old = get_agent_details(OLD_ELEVENLABS_API_KEY, agent_id_old, "Old")
        if not agent_details_old:
            print(f"🔻 Failed to fetch details for agent {agent_id_old}. Skipping.")
            failed_migrations += 1
            continue

        # 2b. Prepare payload for creation in the new workspace
        payload_for_new_agent = prepare_payload_for_create(agent_details_old)
        
        if not payload_for_new_agent:
            # Error message already printed in prepare_payload_for_create if conversation_config was missing
            print(f"🔻 Failed to prepare payload for agent '{agent_name_old}' (Old ID: {agent_id_old}). Skipping.")
            failed_migrations += 1
            continue
        
        # 2c. Create the agent in the new workspace
        created_agent_info = create_agent(NEW_ELEVENLABS_API_KEY, payload_for_new_agent, "New")
        if created_agent_info and created_agent_info.get("agent_id"):
            successful_migrations += 1
        else:
            failed_migrations += 1
            # Error message already printed in create_agent
            print(f"🔻 Failed to migrate agent '{agent_name_old}' (Old ID: {agent_id_old}).")

    # --- Migration Summary ---
    print("\n--- 📊 Migration Summary ---")
    print(f"Total agents found in old workspace: {len(old_agents_summary)}")
    print(f"✅ Successfully migrated: {successful_migrations}")
    print(f"❌ Failed to migrate: {failed_migrations}")
    print("🏁 Migration script finished.")

if __name__ == "__main__":
    main()