devin_cli.py

#!/usr/bin/env python3
"""
Devin CLI - Enhanced version with auth management
A command line tool for creating Devin sessions with built-in authentication management
"""

import os
import sys
import json
import requests
import click
from pathlib import Path
from typing import Optional, List


class DevinAPIError(Exception):
    """Custom exception for Devin API errors"""
    pass


def get_config_dir() -> Path:
    """Get the configuration directory for storing auth tokens"""
    config_dir = Path.home() / '.devin-cli'
    config_dir.mkdir(exist_ok=True)
    return config_dir


def get_token_file() -> Path:
    """Get the path to the token file"""
    return get_config_dir() / 'token'


def save_token(token: str) -> None:
    """Save the API token to a secure file"""
    token_file = get_token_file()
    
    # Write token to file with restricted permissions
    with open(token_file, 'w') as f:
        f.write(token.strip())
    
    # Set file permissions to be readable only by owner (600)
    os.chmod(token_file, 0o600)
    click.echo(f"✅ Token saved securely to {token_file}")


def load_token() -> Optional[str]:
    """Load the API token from file or environment variable"""
    # First try environment variable
    env_token = os.getenv('DEVIN_API_KEY')
    if env_token:
        return env_token.strip()
    
    # Then try saved token file
    token_file = get_token_file()
    if token_file.exists():
        try:
            with open(token_file, 'r') as f:
                return f.read().strip()
        except Exception as e:
            click.echo(f"⚠️  Warning: Could not read saved token: {e}", err=True)
    
    return None


def get_api_key() -> str:
    """Get API key from environment variable or saved file"""
    api_key = load_token()
    if not api_key:
        raise click.ClickException(
            "No Devin API token found.\n\n"
            "To set up authentication, run:\n"
            "  devin-cli auth\n\n"
            "Or set the environment variable:\n"
            "  export DEVIN_API_KEY=your_token_here"
        )
    return api_key


def test_token(token: str) -> bool:
    """Test if a token is valid by making a test API request"""
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json'
    }
    
    # Make a minimal test request
    try:
        response = requests.post(
            'https://api.devin.ai/v1/sessions',
            headers=headers,
            json={'prompt': 'test'},
            timeout=10
        )
        # Check for authentication errors
        if response.status_code in [401, 403]:
            return False  # Invalid token
        # If we get any other response (including errors), assume token is valid
        return True
    except Exception:
        # Network errors, assume token format is OK
        return True


def make_api_request(payload: dict) -> dict:
    """Make API request to create Devin session"""
    api_key = get_api_key()
    
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.post(
            'https://api.devin.ai/v1/sessions',
            headers=headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        raise DevinAPIError(f"API request failed: {e}")


def get_session_details(session_id: str) -> dict:
    """Get details of an existing Devin session"""
    api_key = get_api_key()
    
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.get(
            f'https://api.devin.ai/v1/sessions/{session_id}',
            headers=headers,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        raise DevinAPIError(f"API request failed: {e}")


def send_message_to_session(session_id: str, payload: dict) -> dict:
    """Send a message to an existing Devin session"""
    api_key = get_api_key()
    
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.post(
            f'https://api.devin.ai/v1/sessions/{session_id}/message',
            headers=headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        raise DevinAPIError(f"API request failed: {e}")


def parse_list_input(value: str) -> List[str]:
    """Parse comma-separated string into list"""
    if not value:
        return []
    return [item.strip() for item in value.split(',') if item.strip()]


@click.group(invoke_without_command=True)
@click.pass_context
@click.version_option(version='1.1.0')
def cli(ctx):
    """Devin CLI - Create and manage Devin sessions"""
    if ctx.invoked_subcommand is None:
        # Default behavior - show help
        click.echo(ctx.get_help())


@cli.command()
@click.option('--test', is_flag=True, help='Test the current token')
def auth(test):
    """Set or test your Devin API token"""
    if test:
        token = load_token()
        if token:
            click.echo("Testing token...")
            if test_token(token):
                source = "environment variable" if os.getenv('DEVIN_API_KEY') else "saved file"
                click.echo(f"✅ Token is valid (from {source})")
            else:
                click.echo("❌ Token is invalid")
        else:
            click.echo("❌ No token found")
        return
    
    # Set token
    click.echo("Enter your Devin API token (get it from: https://api.devin.ai)")
    token = click.prompt('Token', hide_input=True, type=str)
    
    click.echo("Testing token...")
    if test_token(token):
        save_token(token)
        click.echo("✅ Token saved and verified!")
    else:
        click.echo("❌ Token is invalid. Please check and try again.")
        sys.exit(1)


@cli.command()
@click.option('--prompt', '-p', help='The task description for Devin')
@click.option('--snapshot-id', help='ID of a machine snapshot to use')
@click.option('--unlisted', is_flag=True, default=None, help='Make the session unlisted')
@click.option('--idempotent', is_flag=True, default=None, help='Enable idempotent session creation')
@click.option('--max-acu-limit', type=int, help='Maximum ACU limit for the session')
@click.option('--secret-ids', help='Comma-separated list of secret IDs to use')
@click.option('--knowledge-ids', help='Comma-separated list of knowledge IDs to use')
@click.option('--tags', help='Comma-separated list of tags to add to the session')
@click.option('--title', help='Custom title for the session')
@click.option('--output', '-o', type=click.Choice(['json', 'table']), default='table', help='Output format')
def create(prompt, snapshot_id, unlisted, idempotent, max_acu_limit, 
          secret_ids, knowledge_ids, tags, title, output):
    """Create a new Devin session (interactive by default)"""
    
    # If no prompt provided, go into full interactive mode
    if not prompt:
        prompt = click.prompt('Task description for Devin', type=str)
        
        # Prompt for all optional fields in interactive mode
        if snapshot_id is None:
            snapshot_id = click.prompt('Snapshot ID (optional)', default='', show_default=False)
            snapshot_id = snapshot_id if snapshot_id else None
        
        if unlisted is None:
            unlisted = click.confirm('Make session unlisted?', default=False)
        
        if idempotent is None:
            idempotent = click.confirm('Enable idempotent session creation?', default=False)
        
        if max_acu_limit is None:
            max_acu_limit_input = click.prompt('Maximum ACU limit (optional)', default='', show_default=False)
            max_acu_limit = int(max_acu_limit_input) if max_acu_limit_input else None
        
        if secret_ids is None:
            secret_ids = click.prompt('Secret IDs (comma-separated, optional)', default='', show_default=False)
        
        if knowledge_ids is None:
            knowledge_ids = click.prompt('Knowledge IDs (comma-separated, optional)', default='', show_default=False)
        
        if tags is None:
            tags = click.prompt('Tags (comma-separated, optional)', default='', show_default=False)
        
        if title is None:
            title = click.prompt('Custom title (optional)', default='', show_default=False)
            title = title if title else None

    # Build the request payload
    payload = {'prompt': prompt}
    
    # Add optional parameters if provided
    if snapshot_id:
        payload['snapshot_id'] = snapshot_id
    if unlisted is not None:
        payload['unlisted'] = unlisted
    if idempotent is not None:
        payload['idempotent'] = idempotent
    if max_acu_limit is not None:
        payload['max_acu_limit'] = max_acu_limit
    if secret_ids:
        payload['secret_ids'] = parse_list_input(secret_ids)
    if knowledge_ids:
        payload['knowledge_ids'] = parse_list_input(knowledge_ids)
    if tags:
        payload['tags'] = parse_list_input(tags)
    if title:
        payload['title'] = title

    try:
        # Make the API request
        click.echo("Creating Devin session...")
        result = make_api_request(payload)
        
        # Output the result
        if output == 'json':
            click.echo(json.dumps(result, indent=2))
        else:
            # Table format
            click.echo("\n✅ Session created successfully!")
            click.echo(f"Session ID: {result.get('session_id', 'N/A')}")
            click.echo(f"URL: {result.get('url', 'N/A')}")
            click.echo(f"New Session: {result.get('is_new_session', 'N/A')}")
            
    except DevinAPIError as e:
        click.echo(f"❌ Error: {e}", err=True)
        sys.exit(1)
    except Exception as e:
        click.echo(f"❌ Unexpected error: {e}", err=True)
        sys.exit(1)


@cli.command()
@click.argument('session_id')
@click.option('--output', '-o', type=click.Choice(['json', 'table']), default='table', help='Output format')
def get(session_id, output):
    """Get details of an existing Devin session"""
    try:
        click.echo(f"Retrieving session details for {session_id}...")
        result = get_session_details(session_id)
        
        # Output the result
        if output == 'json':
            click.echo(json.dumps(result, indent=2))
        else:
            # Table format
            click.echo("\n📄 Session Details:")
            click.echo(f"Session ID: {result.get('session_id', 'N/A')}")
            click.echo(f"Status: {result.get('status', 'N/A')}")
            click.echo(f"Title: {result.get('title', 'N/A')}")
            click.echo(f"Created: {result.get('created_at', 'N/A')}")
            click.echo(f"Updated: {result.get('updated_at', 'N/A')}")
            
            if result.get('tags'):
                click.echo(f"Tags: {', '.join(result['tags'])}")
            
            messages = result.get('messages', [])
            if messages:
                click.echo(f"\n💬 Messages ({len(messages)} total):")
                for i, msg in enumerate(messages[:3]):  # Show first 3 messages
                    click.echo(f"  {i+1}. {msg.get('role', 'unknown')}: {msg.get('content', '')[:100]}...")
                if len(messages) > 3:
                    click.echo(f"  ... and {len(messages) - 3} more messages")
                    
    except DevinAPIError as e:
        click.echo(f"❌ Error: {e}", err=True)
        sys.exit(1)
    except Exception as e:
        click.echo(f"❌ Unexpected error: {e}", err=True)
        sys.exit(1)


@cli.command()
@click.argument('session_id')
@click.option('--message', '-m', help='Message to send to the session')
@click.option('--output', '-o', type=click.Choice(['json', 'table']), default='table', help='Output format')
def message(session_id, message, output):
    """Send a message to an existing Devin session"""
    
    # If no message provided, prompt for it
    if not message:
        message = click.prompt('Message to send to the session', type=str)
    
    payload = {'message': message}
    
    try:
        click.echo(f"Sending message to session {session_id}...")
        result = send_message_to_session(session_id, payload)
        
        # Output the result
        if output == 'json':
            click.echo(json.dumps(result, indent=2))
        else:
            # Table format
            click.echo("\n✅ Message sent successfully!")
            if 'message_id' in result:
                click.echo(f"Message ID: {result['message_id']}")
            if 'status' in result:
                click.echo(f"Session Status: {result['status']}")
                
    except DevinAPIError as e:
        click.echo(f"❌ Error: {e}", err=True)
        sys.exit(1)
    except Exception as e:
        click.echo(f"❌ Unexpected error: {e}", err=True)
        sys.exit(1)


@cli.command()
@click.option('--target-dir', '-t', default='.', help='Target directory to copy files to (default: current directory)')
@click.option('--force', '-f', is_flag=True, help='Overwrite existing files without prompting')
def setup(target_dir, force):
    """Download latest Devin workflow and session guide to your repository"""
    
    # GitHub raw URLs for the public repo
    base_url = "https://raw.githubusercontent.com/parkerduff/devin-cli/main"
    files_to_download = [
        {
            'url': f"{base_url}/devin-session-guide.md",
            'target_path': 'devin-session-guide.md',
            'description': 'Devin Session Guide'
        },
        {
            'url': f"{base_url}/.windsurf/workflows/create-session.md",
            'target_path': '.windsurf/workflows/create-session.md',
            'description': 'Create Session Workflow'
        }
    ]
    
    target_base = Path(target_dir).resolve()
    success_count = 0
    
    for file_info in files_to_download:
        target_file = target_base / file_info['target_path']
        
        # Create parent directories if they don't exist
        target_file.parent.mkdir(parents=True, exist_ok=True)
        
        # Check if file exists and handle overwrite
        if target_file.exists() and not force:
            if not click.confirm(f"{file_info['description']} already exists at {target_file}. Overwrite?"):
                click.echo(f"⏭️  Skipped {file_info['description']}")
                continue
        
        # Download the file
        try:
            click.echo(f"📥 Downloading {file_info['description']}...")
            response = requests.get(file_info['url'], timeout=10)
            response.raise_for_status()
            
            # Write the content to file
            with open(target_file, 'w', encoding='utf-8') as f:
                f.write(response.text)
            
            click.echo(f"✅ Downloaded {file_info['description']} → {target_file}")
            success_count += 1
            
        except requests.exceptions.RequestException as e:
            click.echo(f"❌ Failed to download {file_info['description']}: {e}", err=True)
        except Exception as e:
            click.echo(f"❌ Error writing {file_info['description']}: {e}", err=True)
    
    # Summary
    if success_count > 0:
        click.echo(f"\n📝 Files downloaded to: {target_base}")
        click.echo("\nYou can now use:")
        click.echo("  • /create-session workflow in Windsurf")
        click.echo("  • devin-session-guide.md for prompt examples")
    elif success_count == 0:
        click.echo("\n⚠️  No files were downloaded.")


if __name__ == '__main__':
    cli()