mvp-factory-openhands/test-scripts/openhands-sdk-wrapper.py

#!/usr/bin/env python3
"""
OpenHands SDK Wrapper for n8n Integration
Usage: python3 /home/bam/openhands-sdk-wrapper.py "task description"

This script runs OpenHands in SDK mode without Docker containers,
perfect for n8n workflow integration.
"""

import sys
import os
import json
import argparse
from datetime import datetime

# Add SDK to path
SDK_PATH = '/tmp/software-agent-sdk'
sys.path.insert(0, SDK_PATH)

try:
    from openhands.sdk import LLM, Agent, Conversation, Tool
    from openhands.tools.file_editor import FileEditorTool
    from openhands.tools.task_tracker import TaskTrackerTool
except ImportError as e:
    print(f"❌ Failed to import OpenHands SDK: {e}")
    print("Make sure SDK is built: cd /tmp/software-agent-sdk && make build")
    sys.exit(1)

def run_openhands_task(task, workspace="/home/bam", verbose=True):
    """
    Execute an OpenHands task using the SDK

    Args:
        task: Task description string
        workspace: Working directory path
        verbose: Print detailed logs

    Returns:
        dict: Results including success status and output
    """
    results = {
        'success': False,
        'task': task,
        'workspace': workspace,
        'timestamp': datetime.now().isoformat(),
        'error': None,
        'files_created': [],
        'log_output': []
    }

    try:
        if verbose:
            print(f"🚀 Starting OpenHands SDK execution...")
            print(f"📋 Task: {task}")
            print(f"📁 Workspace: {workspace}")
            print("-" * 50)

        # Load environment
        env_file = '/home/bam/openhands/.env'
        if os.path.exists(env_file):
            with open(env_file, 'r') as f:
                for line in f:
                    if '=' in line and not line.startswith('#'):
                        key, value = line.strip().split('=', 1)
                        os.environ[key] = value

        # Configure LLM
        api_key = os.getenv('MINIMAX_API_KEY')
        if not api_key:
            raise ValueError("MINIMAX_API_KEY not found in environment")

        llm = LLM(
            model="openai/MiniMax-M2",
            api_key=api_key,
            base_url="https://api.minimax.io/v1"
        )

        if verbose:
            print("✅ LLM configured successfully")

        # Create agent with tools
        agent = Agent(
            llm=llm,
            tools=[
                Tool(name=FileEditorTool.name),
                Tool(name=TaskTrackerTool.name),
            ],
        )

        if verbose:
            print("✅ Agent created with tools")

        # Start conversation
        conversation = Conversation(agent=agent, workspace=workspace)

        if verbose:
            print("💬 Conversation started")

        # Send task and run
        conversation.send_message(task)

        if verbose:
            print("📤 Task sent to agent")
            print("⏳ Running agent execution...")

        conversation.run()

        if verbose:
            print("✅ Agent execution completed")

        # Check for created files
        for file in os.listdir(workspace):
            file_path = os.path.join(workspace, file)
            if os.path.isfile(file_path):
                results['files_created'].append(file)

        results['success'] = True

        if verbose:
            print("-" * 50)
            print(f"✅ Task completed successfully!")
            print(f"📄 Files created: {results['files_created']}")

    except Exception as e:
        results['error'] = str(e)
        if verbose:
            print(f"❌ Task failed: {e}")
            import traceback
            traceback.print_exc()

    return results

def main():
    """Main entry point for the wrapper script"""
    parser = argparse.ArgumentParser(description='OpenHands SDK Wrapper for n8n')
    parser.add_argument('task', help='Task description to execute')
    parser.add_argument('--workspace', default='/home/bam', help='Working directory')
    parser.add_argument('--quiet', action='store_true', help='Suppress verbose output')
    parser.add_argument('--json', action='store_true', help='Output results as JSON')

    args = parser.parse_args()

    # Run the task
    results = run_openhands_task(
        task=args.task,
        workspace=args.workspace,
        verbose=not args.quiet
    )

    # Output results
    if args.json:
        print(json.dumps(results, indent=2))
    else:
        if results['success']:
            print("SUCCESS")
            if results['files_created']:
                print(f"Files created: {', '.join(results['files_created'])}")
        else:
            print("FAILED")
            if results['error']:
                print(f"Error: {results['error']}")

    # Exit with appropriate code
    sys.exit(0 if results['success'] else 1)

if __name__ == "__main__":
    # If called without arguments, assume direct execution (for testing)
    if len(sys.argv) == 1:
        # Default test task
        task = "Create a file named sdk-wrapper-test.txt with content: SDK wrapper test successful!"
        print(f"🧪 Running default test task: {task}")
        results = run_openhands_task(task)

        print("\n" + "="*50)
        print("RESULTS:")
        print(f"Success: {results['success']}")
        print(f"Files created: {results['files_created']}")
        if results['error']:
            print(f"Error: {results['error']}")

        # Check if file was created
        if results['success'] and os.path.exists('/home/bam/sdk-wrapper-test.txt'):
            print("\n✅ File content:")
            with open('/home/bam/sdk-wrapper-test.txt', 'r') as f:
                print(f.read())
    else:
        main()