Imported from dusk-time/tech-openclaw (skills/intelligent-task-coordinator/SKILL.md). Install upstream with npx skills add dusk-time/tech-openclaw --skill intelligent-task-coordinator. Copyright stays with the author.

Intelligent Task Coordinator

Overview

This skill provides intelligent task decomposition and dynamic agent orchestration. It analyzes complex tasks, breaks them down into manageable subtasks, discovers available skills and agents, matches them to subtasks based on capabilities, coordinates execution, and aggregates results.

Key Features

• Automatic Task Decomposition - Break complex tasks into logical subtasks
• Skill Discovery - Scan available skills and match to task requirements
• Agent Selection - Choose optimal agents based on capabilities and availability
• Dynamic Orchestration - Adapt workflow based on execution results
• Result Aggregation - Combine results from multiple agents into coherent output

Architecture

┌─────────────────┐
│  User Request   │
│  "Analyze global│
│   assets and    │
│   create report"│
└────────┬────────┘
         │
         ▼
┌─────────────────────────────────┐
│  Intelligent Task Coordinator   │
│  1. Analyze task intent         │
│  2. Decompose into subtasks     │
│  3. Discover skills & agents    │
│  4. Match & assign              │
│  5. Coordinate execution        │
│  6. Aggregate results           │
└────────┬────────────────────────┘
         │
         ├──────────────┬──────────────┬──────────────┬──────────────┐
         ▼              ▼              ▼              ▼              ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Skill:      │ │ Skill:      │ │ Skill:      │ │ Skill:      │ │ Skill:      │
│ stock-      │ │ yahoo-      │ │ plotly      │ │ python-     │ │ browser-    │
│ analysis    │ │ finance     │ │             │ │ project     │ │ use         │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
         │              │              │              │              │
         ▼              ▼              ▼              ▼              ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Agent:      │ │ Agent:      │ │ Agent:      │ │ Agent:      │ │ Agent:      │
│ Data        │ │ Analysis    │ │ Chart       │ │ Code        │ │ Research    │
│ Fetcher     │ │ Specialist  │ │ Generator   │ │ Executor    │ │ Assistant   │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
         │              │              │              │              │
         └──────────────┴──────────────┴──────────────┴──────────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │  Result         │
                    │  Aggregator     │
                    │  & Reporter     │
                    └─────────────────┘

Core Components

1. Task Analyzer

class TaskAnalyzer:
    """Analyze task complexity and requirements"""
    
    def analyze(self, task_description: str) -> TaskAnalysis:
        """
        Analyze task and return:
        - Complexity score (1-10)
        - Required capabilities
        - Estimated subtasks
        - Dependencies
        """
        
        # Extract keywords and intent
        keywords = self.extract_keywords(task_description)
        intent = self.classify_intent(task_description)
        
        # Determine complexity
        complexity = self.estimate_complexity(task_description)
        
        # Identify required capabilities
        capabilities = self.identify_capabilities(keywords, intent)
        
        return TaskAnalysis(
            original_task=task_description,
            complexity=complexity,
            intent=intent,
            required_capabilities=capabilities,
            estimated_subtasks=self.suggest_subtasks(intent, capabilities)
        )

2. Skill Discovery Engine

class SkillDiscoveryEngine:
    """Discover and catalog available skills"""
    
    def __init__(self):
        self.skills_dir = Path.home() / '.openclaw' / 'workspace' / 'skills'
        self.skill_registry = {}
    
    def scan_skills(self) -> Dict[str, SkillInfo]:
        """Scan workspace for available skills"""
        
        for skill_dir in self.skills_dir.iterdir():
            if not skill_dir.is_dir():
                continue
            
            skill_md = skill_dir / 'SKILL.md'
            if not skill_md.exists():
                continue
            
            # Parse skill metadata
            skill_info = self.parse_skill(skill_md)
            self.skill_registry[skill_info.name] = skill_info
        
        return self.skill_registry
    
    def match_skills_to_task(self, task_requirements: List[str]) -> List[SkillMatch]:
        """Match skills to task requirements"""
        
        matches = []
        for skill_name, skill_info in self.skill_registry.items():
            score = self.calculate_match_score(skill_info, task_requirements)
            if score > 0.5:  # Threshold
                matches.append(SkillMatch(
                    skill=skill_info,
                    confidence=score,
                    matched_capabilities=self.find_matches(skill_info, task_requirements)
                ))
        
        return sorted(matches, key=lambda m: m.confidence, reverse=True)

3. Agent Selector

class AgentSelector:
    """Select optimal agents for subtasks"""
    
    def __init__(self):
        self.agent_pool = []
        self.session_history = []
    
    def discover_agents(self) -> List[AgentInfo]:
        """Discover available agents (sessions_spawn, existing sessions)"""
        
        # Check active sessions
        sessions = sessions_list(activeMinutes=60)
        
        # Check available agent IDs
        agent_ids = agents_list()
        
        # Build agent pool
        for session in sessions:
            agent_info = AgentInfo(
                id=session['sessionKey'],
                type='session',
                capabilities=session.get('capabilities', []),
                status='busy' if session.get('active') else 'available',
                load=self.estimate_load(session)
            )
            self.agent_pool.append(agent_info)
        
        return self.agent_pool
    
    def select_agent(self, subtask: Subtask) -> AgentSelection:
        """Select best agent for subtask"""
        
        candidates = []
        for agent in self.agent_pool:
            if agent.status != 'available':
                continue
            
            # Calculate fit score
            fit_score = self.calculate_fit(agent, subtask)
            
            # Check capacity
            if agent.load > 0.8:
                continue
            
            candidates.append((agent, fit_score))
        
        if not candidates:
            # Spawn new agent if needed
            return self.spawn_agent_for_task(subtask)
        
        # Select best candidate
        best_agent, best_score = max(candidates, key=lambda x: x[1])
        
        return AgentSelection(
            agent=best_agent,
            confidence=best_score,
            strategy='reuse_existing'
        )

4. Task Decomposer

class TaskDecomposer:
    """Decompose complex tasks into subtasks"""
    
    DECOMPOSITION_PATTERNS = {
        'data_analysis': [
            'fetch_data',
            'clean_data',
            'analyze_data',
            'visualize_results',
            'generate_report'
        ],
        'research': [
            'define_research_question',
            'search_sources',
            'extract_information',
            'synthesize_findings',
            'write_report'
        ],
        'development': [
            'understand_requirements',
            'design_solution',
            'implement_code',
            'test_implementation',
            'document_solution'
        ],
        'asset_analysis': [
            'fetch_historical_data',
            'calculate_technical_indicators',
            'perform_fundamental_analysis',
            'generate_price_predictions',
            'create_visualizations',
            'compile_investment_recommendations'
        ]
    }
    
    def decompose(self, task_analysis: TaskAnalysis) -> List[Subtask]:
        """Decompose task into subtasks"""
        
        # Select decomposition pattern
        pattern = self.select_pattern(task_analysis.intent)
        
        subtasks = []
        for i, step in enumerate(pattern):
            subtask = Subtask(
                id=f"subtask_{i+1}",
                description=self.generate_description(step, task_analysis),
                required_capabilities=self.get_required_capabilities(step),
                dependencies=[f"subtask_{i}"] if i > 0 else [],
                estimated_duration=self.estimate_duration(step),
                priority=self.calculate_priority(step, pattern)
            )
            subtasks.append(subtask)
        
        return subtasks

5. Workflow Orchestrator

class WorkflowOrchestrator:
    """Orchestrate execution of subtasks across agents"""
    
    def __init__(self):
        self.execution_graph = nx.DiGraph()
        self.results = {}
    
    def build_execution_graph(self, subtasks: List[Subtask]) -> nx.DiGraph:
        """Build DAG of task dependencies"""
        
        for subtask in subtasks:
            self.execution_graph.add_node(
                subtask.id,
                subtask=subtask,
                status='pending'
            )
            
            for dep in subtask.dependencies:
                self.execution_graph.add_edge(dep, subtask.id)
        
        return self.execution_graph
    
    async def execute(self) -> ExecutionResult:
        """Execute workflow with parallel execution where possible"""
        
        # Find ready tasks (no pending dependencies)
        ready_tasks = self.get_ready_tasks()
        
        while ready_tasks:
            # Execute ready tasks in parallel
            tasks = [
                self.execute_subtask(task_id)
                for task_id in ready_tasks
            ]
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Update graph
            for task_id, result in zip(ready_tasks, results):
                self.execution_graph.nodes[task_id]['status'] = 'completed'
                self.execution_graph.nodes[task_id]['result'] = result
                self.results[task_id] = result
            
            # Find next ready tasks
            ready_tasks = self.get_ready_tasks()
        
        return ExecutionResult(
            success=True,
            results=self.results,
            execution_time=self.calculate_execution_time()
        )

Usage Examples

Example 1: Global Asset Analysis

# User request
task = "Analyze 7 major asset classes (Bitcoin, Gold, Silver, Oil, Copper, Lithium, DRAM) 
        with 20-year historical data, technical indicators, price predictions for 1M/3M/6M/1Y/2Y/5Y, 
        and generate interactive charts with investment recommendations"

# Coordinator automatically:
# 1. Analyzes task complexity → Score: 8/10
# 2. Decomposes into subtasks:
#    - Fetch historical data for 7 assets
#    - Calculate technical indicators (RSI, MACD, Bollinger)
#    - Run Monte Carlo simulations for predictions
#    - Generate interactive charts
#    - Compile investment recommendations
# 3. Discovers matching skills:
#    - stock-analysis
#    - yahoo-finance
#    - plotly
#    - python-project
# 4. Selects agents:
#    - Data fetching agent (existing session)
#    - Analysis agent (sessions_spawn)
#    - Chart generation agent (sessions_spawn)
#    - Report writing agent (existing session)
# 5. Executes workflow and aggregates results

Example 2: Research Report

task = "Research the impact of AI on semiconductor industry, 
        find top 10 companies, analyze their financials, 
        and create investment ranking"

# Automatic decomposition:
# 1. Search for AI semiconductor companies
# 2. Filter top 10 by market cap
# 3. Fetch financial data for each
# 4. Calculate valuation metrics
# 5. Rank by investment attractiveness
# 6. Generate report

# Skills matched:
# - web_search (for company research)
# - yahoo-finance (for financial data)
# - stock-analysis (for valuation)
# - python-project (for ranking algorithm)

Implementation

Main Coordinator Class

class IntelligentTaskCoordinator:
    """Main coordinator that ties everything together"""
    
    def __init__(self):
        self.analyzer = TaskAnalyzer()
        self.decomposer = TaskDecomposer()
        self.skill_discovery = SkillDiscoveryEngine()
        self.agent_selector = AgentSelector()
        self.orchestrator = WorkflowOrchestrator()
        
        # Initialize
        self.skill_discovery.scan_skills()
        self.agent_selector.discover_agents()
    
    async def execute_task(self, task_description: str) -> CoordinatorResult:
        """Execute a complex task with automatic decomposition and orchestration"""
        
        # Step 1: Analyze task
        print(f"📊 Analyzing task: {task_description[:100]}...")
        task_analysis = self.analyzer.analyze(task_description)
        
        # Step 2: Decompose into subtasks
        print(f"🔧 Decomposing into {len(task_analysis.estimated_subtasks)} subtasks...")
        subtasks = self.decomposer.decompose(task_analysis)
        
        # Step 3: Discover and match skills
        print("🔍 Discovering relevant skills...")
        skill_matches = self.skill_discovery.match_skills_to_task(
            task_analysis.required_capabilities
        )
        
        # Step 4: Select agents for each subtask
        print("🤖 Selecting optimal agents...")
        agent_assignments = []
        for subtask in subtasks:
            selection = self.agent_selector.select_agent(subtask)
            agent_assignments.append((subtask, selection))
        
        # Step 5: Build execution graph
        print("📋 Building execution workflow...")
        self.orchestrator.build_execution_graph(subtasks)
        
        # Step 6: Execute workflow
        print("⚡ Executing workflow...")
        execution_result = await self.orchestrator.execute()
        
        # Step 7: Aggregate results
        print("📦 Aggregating results...")
        final_result = self.aggregate_results(execution_result.results)
        
        return CoordinatorResult(
            success=execution_result.success,
            task_analysis=task_analysis,
            subtasks=subtasks,
            skill_matches=skill_matches,
            agent_assignments=agent_assignments,
            execution_result=execution_result,
            final_result=final_result
        )
    
    def aggregate_results(self, results: Dict[str, Any]) -> Any:
        """Aggregate results from multiple agents into coherent output"""
        
        # Combine data from all subtasks
        combined = {
            'summary': self.generate_summary(results),
            'details': results,
            'recommendations': self.generate_recommendations(results),
            'visualizations': self.collect_visualizations(results),
            'metadata': {
                'total_subtasks': len(results),
                'successful': sum(1 for r in results.values() if not isinstance(r, Exception)),
                'failed': sum(1 for r in results.values() if isinstance(r, Exception))
            }
        }
        
        return combined

Integration with OpenClaw

# Using sessions_spawn for dynamic agent creation
async def spawn_agent_for_subtask(self, subtask: Subtask) -> AgentInfo:
    """Spawn new agent for subtask if no existing agent available"""
    
    # Determine required model based on task type
    if 'analysis' in subtask.required_capabilities:
        model = 'bailian/qwen3.5-plus'
    elif 'chart' in subtask.required_capabilities:
        model = 'bailian/qwen3.5-plus'  # Good at code generation
    else:
        model = 'default'
    
    # Spawn agent
    result = await sessions_spawn(
        task=subtask.description,
        model=model,
        timeoutSeconds=subtask.estimated_duration * 60,
        cleanup='delete'
    )
    
    return AgentInfo(
        id=result['sessionKey'],
        type='spawned',
        capabilities=subtask.required_capabilities,
        status='running',
        load=0.5
    )

# Using message for inter-agent communication
async def send_task_to_agent(self, agent_id: str, task: Subtask):
    """Send task to specific agent session"""
    
    await sessions_send(
        sessionKey=agent_id,
        message=json.dumps({
            'type': 'task_assignment',
            'task_id': task.id,
            'description': task.description,
            'requirements': task.required_capabilities,
            'deadline': task.estimated_duration
        })
    )

Skill Matching Algorithm

def calculate_match_score(self, skill: SkillInfo, requirements: List[str]) -> float:
    """Calculate how well a skill matches task requirements"""
    
    score = 0.0
    matches = []
    
    # Check name match
    for req in requirements:
        if req.lower() in skill.name.lower():
            score += 0.3
            matches.append(('name', req))
    
    # Check description match
    if skill.description:
        for req in requirements:
            if req.lower() in skill.description.lower():
                score += 0.2
                matches.append(('description', req))
    
    # Check capabilities
    if hasattr(skill, 'capabilities'):
        for req in requirements:
            for cap in skill.capabilities:
                if req.lower() in cap.lower():
                    score += 0.25
                    matches.append(('capability', req))
    
    # Check tools mentioned
    if hasattr(skill, 'tools'):
        for req in requirements:
            for tool in skill.tools:
                if req.lower() in tool.lower():
                    score += 0.15
                    matches.append(('tool', req))
    
    return min(score, 1.0)  # Cap at 1.0

Monitoring and Debugging

class WorkflowMonitor:
    """Monitor workflow execution and provide insights"""
    
    def __init__(self):
        self.events = []
        self.metrics = {
            'total_tasks': 0,
            'completed_tasks': 0,
            'failed_tasks': 0,
            'avg_execution_time': 0,
            'skills_used': set(),
            'agents_used': set()
        }
    
    def log_event(self, event_type: str, details: Dict):
        """Log workflow event"""
        self.events.append({
            'timestamp': datetime.now().isoformat(),
            'type': event_type,
            'details': details
        })
    
    def generate_report(self) -> str:
        """Generate execution report"""
        report = f"""
# Workflow Execution Report

## Summary
- Total Tasks: {self.metrics['total_tasks']}
- Completed: {self.metrics['completed_tasks']}
- Failed: {self.metrics['failed_tasks']}
- Success Rate: {self.metrics['completed_tasks'] / max(self.metrics['total_tasks'], 1) * 100:.1f}%
- Avg Execution Time: {self.metrics['avg_execution_time']:.2f}s

## Skills Used
{', '.join(self.metrics['skills_used']) or 'None'}

## Agents Used
{', '.join(self.metrics['agents_used']) or 'None'}

## Timeline
"""
        for event in self.events:
            report += f"- [{event['timestamp']}] {event['type']}: {event['details']}\n"
        
        return report

Quick Start

# Initialize coordinator
coordinator = IntelligentTaskCoordinator()

# Execute complex task
result = await coordinator.execute_task(
    "Analyze global asset classes with historical data, "
    "technical indicators, price predictions, and generate "
    "interactive charts with investment recommendations"
)

# Review results
print(f"✅ Task completed: {result.success}")
print(f"📊 Subtasks executed: {len(result.subtasks)}")
print(f"🛠️  Skills used: {[m.skill.name for m in result.skill_matches]}")
print(f"🤖 Agents involved: {len(result.agent_assignments)}")

This skill automatically decomposes complex tasks, discovers relevant skills and agents, and orchestrates their execution! 🚀