Mcp Code Execution

# MCP Subagent Coordination Patterns

> **Version Note (Claude Code 2.1.14+)**: Parallel subagent execution is significantly more stable in Claude Code 2.1.14+, which fixed memory issues that could cause crashes when running parallel subagents. Earlier versions may experience heap out of memory errors with 3+ concurrent agents.

## Pipeline Coordination

Execute subagents in sequence with MECW monitoring:

```python
def coordinate_pipeline_subagents(subagents, input_data):
    """Execute subagents in sequence with MECW monitoring"""

    current_data = input_data
    results = []

    for subagent in subagents:
        # Monitor context before each subagent
        if estimate_context_usage() > get_mecw_limit() * 0.8:
            apply_emergency_compaction()

        # Execute subagent with minimal context
        subagent_result = subagent.execute_focused_task({
            'data': current_data,
            'context_limit': get_mecw_limit() * 0.4
        })

        current_data = subagent_result.get('next_input', current_data)
        results.append(subagent_result)

        # Store intermediate results externally
        store_intermediate_result(subagent.purpose, subagent_result)

    return results
```

### When to Use Pipeline
- **Sequential dependencies**: Each step depends on previous result
- **Linear workflows**: Clear progression from input to output
- **Token conservation**: Share minimal context between steps
- **Error isolation**: Failures don't cascade to all subagents

## Parallel Coordination

Execute multiple subagents simultaneously:

```python
def coordinate_parallel_subagents(subagents, input_data):
    """Execute multiple subagents simultaneously"""

    # Split input data for parallel processing
    data_splits = split_input_for_parallel(input_data, len(subagents))

    # Launch subagents with minimal context
    futures = []
    for subagent, data_split in zip(subagents, data_splits):
        future = subagent.execute_async({
            'data': data_split,
            'context_limit': get_mecw_limit() // len(subagents)
        })
        futures.append(future)

    # Collect results with external storage
    results = []
    for future in futures:
        result = future.get_result()
        store_external_result(result.subagent_id, result.data)
        results.append(result)

    return synthesize_parallel_results(results)
```

### When to Use Parallel
- **Independent tasks**: No dependencies between subagents
- **Time-sensitive**: Need faster completion
- **Resource distribution**: Share MECW budget across subagents
- **Diverse expertise**: Different domains being processed

## Hybrid Coordination

Combine pipeline and parallel patterns:

```python
def coordinate_hybrid_subagents(phase_groups):
    """Execute phases sequentially, subagents within each phase in parallel"""

    all_results = []

    for phase in phase_groups:
        # Execute subagents in this phase in parallel
        phase_results = coordinate_parallel_subagents(
            phase.subagents,
            phase.input_data
        )

        # Synthesize phase results before next phase
        synthesized = synthesize_phase_results(phase_results)
        all_results.append(synthesized)

        # Pass synthesized results to next phase
        if phase.has_next():
            phase.next().set_input_data(synthesized)

    return combine_phase_results(all_results)
```

### When to Use Hybrid
- **Complex workflows**: Mix of sequential and parallel steps
- **Phase dependencies**: Groups of parallel tasks with inter-group dependencies
- **Resource optimization**: Balance speed (parallel) with coordination (sequential)

## Emergency Patterns

### Context Overflow Recovery

```python
def handle_context_overflow(subagent, current_state):
    """Emergency handling when subagent exceeds MECW limits"""

    # 1. Immediately store current state externally
    store_emergency_state(subagent.id, current_state)

    # 2. Split task into smaller sub-s