openclaw/skills/multi-agent/spawn.py

#!/usr/bin/env python3
"""
Multi-Agent Spawner - Spawn any agent type via Gateway API
Bypasses sessions_spawn tool limitations
"""
import requests
import json
import argparse
import sys
import os

GATEWAY_URL = os.environ.get('OPENCLAW_GATEWAY_URL', 'https://localhost:18789')
GATEWAY_TOKEN = os.environ.get('OPENCLAW_GATEWAY_TOKEN', '916117a24ae61a926536bf992e38847421e53d16fb12ab73')

def spawn_agent(agent_id, task, label=None, timeout=3600, model=None, thinking='low', deliver=False):
    """
    Spawn an isolated agent session via Gateway API
    
    Args:
        agent_id: Agent ID to spawn (gemini-3-flash, gpt5-mini, sonnet, etc.)
        task: Task message/prompt for the agent
        label: Optional session label
        timeout: Timeout in seconds (default 3600 = 1 hour)
        model: Override model (optional)
        thinking: Thinking level (low/medium/high)
        deliver: Deliver result to channel
    
    Returns:
        dict with sessionKey, runId, status
    """
    
    # Build payload
    payload = {
        'agentId': agent_id,
        'task': task,
        'runTimeoutSeconds': timeout,
        'thinking': thinking,
        'deliver': deliver
    }
    
    if label:
        payload['label'] = label
    if model:
        payload['model'] = model
    
    # Call Gateway API directly (disable SSL verification for self-signed cert)
    response = requests.post(
        f'{GATEWAY_URL}/sessions/spawn',
        headers={
            'Authorization': f'Bearer {GATEWAY_TOKEN}',
            'Content-Type': 'application/json'
        },
        json=payload,
        timeout=10,
        verify=False  # Self-signed cert
    )
    
    if response.status_code == 200:
        result = response.json()
        return {
            'status': 'success',
            'sessionKey': result.get('childSessionKey'),
            'runId': result.get('runId'),
            'agentId': agent_id,
            'label': label,
            'task': task[:100]
        }
    else:
        return {
            'status': 'error',
            'code': response.status_code,
            'error': response.text
        }

def main():
    parser = argparse.ArgumentParser(description='Spawn OpenClaw sub-agents')
    parser.add_argument('--agent', required=True, help='Agent ID (gemini-3-flash, gpt5-mini, sonnet)')
    parser.add_argument('--task', required=True, help='Task message/prompt')
    parser.add_argument('--label', help='Session label')
    parser.add_argument('--timeout', type=int, default=3600, help='Timeout in seconds')
    parser.add_argument('--model', help='Override model')
    parser.add_argument('--thinking', default='low', choices=['low', 'medium', 'high'])
    parser.add_argument('--deliver', action='store_true', help='Deliver result to channel')
    parser.add_argument('--assign-kanban', help='Auto-assign to Kanban task ID')
    parser.add_argument('--json', action='store_true', help='Output JSON only')
    
    args = parser.parse_args()
    
    # Spawn agent
    result = spawn_agent(
        agent_id=args.agent,
        task=args.task,
        label=args.label,
        timeout=args.timeout,
        model=args.model,
        thinking=args.thinking,
        deliver=args.deliver
    )
    
    # Auto-assign to Kanban if requested
    if result['status'] == 'success' and args.assign_kanban:
        import subprocess
        session_id = result['sessionKey'].split(':')[-1]
        assignee = f"{args.agent}:{session_id[:8]}"
        
        subprocess.run([
            'python3', '/home/alex/kanban-board/kanban_helper.py',
            'update', args.assign_kanban,
            f'assignee={assignee}',
            'status=in_progress'
        ], check=False)
    
    # Output
    if args.json:
        print(json.dumps(result, indent=2))
    else:
        if result['status'] == 'success':
            print(f"✅ Spawned {args.agent}")
            print(f"   Session: {result['sessionKey']}")
            print(f"   Run ID: {result['runId']}")
            if args.assign_kanban:
                print(f"   Kanban: Assigned to task {args.assign_kanban}")
        else:
            print(f"❌ Failed to spawn agent")
            print(f"   Error: {result.get('error', 'Unknown')}")
            sys.exit(1)
    
    return result

if __name__ == '__main__':
    main()