mcp-orchestration

name	MCP_ORCHESTRATION
description	Tool discovery, routing, chaining, error handling, and composition for the 34+ MCP scheduling tools. Use when orchestrating complex multi-tool workflows, handling MCP errors, or discovering available capabilities.
model_tier	haiku
parallel_hints	{"can_parallel_with":["code-review","test-writer","security-audit"],"must_serialize_with":["database-migration"],"preferred_batch_size":5}

MCP Orchestration Skill

Expert orchestration of Model Context Protocol (MCP) tools for medical residency scheduling. Handles tool discovery, intelligent routing, error recovery, and complex multi-tool composition.

When This Skill Activates

• Multi-step workflows requiring 2+ MCP tools
• Error recovery from failed MCP calls
• Tool capability discovery needed
• Complex scheduling operations requiring orchestration
• Debugging MCP integration issues
• Performance optimization of tool chains

Overview

The MCP server exposes 34+ specialized tools across 6 categories:

Category	Tools	Purpose
Core Scheduling	5	Validation, generation, conflict detection, swaps
Resilience Framework	13	Utilization, contingency, defense levels, homeostasis
Background Tasks	4	Celery task management (start, status, cancel, list)
Deployment	7	Validation, security, smoke tests, rollback
Empirical Testing	5	Benchmarking solvers, constraints, modules
Resources	2	Schedule status, compliance summary

Total: 36 tools available for orchestration.

Key Orchestration Phases

Phase 1: Discovery

1. Identify available tools matching task requirements
2. Check tool availability and health
3. Verify prerequisites (DB connection, API availability)
4. Map inputs/outputs between dependent tools

Phase 2: Planning

1. Create execution DAG (directed acyclic graph)
2. Identify parallel vs sequential dependencies
3. Plan error handling checkpoints
4. Estimate execution time and resource usage

Phase 3: Execution

1. Execute tools in dependency order
2. Handle transient errors with retry logic
3. Propagate results through tool chain
4. Monitor progress and resource utilization

Phase 4: Recovery

1. Detect permanent vs transient failures
2. Execute fallback strategies
3. Rollback partial state changes if needed
4. Log errors for human escalation

Orchestration Patterns

Pattern 1: Sequential Chain

Tool A → Tool B → Tool C

Each tool depends on previous tool's output.

Example: Schedule Generation Pipeline

validate_deployment → generate_schedule → validate_schedule → run_smoke_tests

Pattern 2: Parallel Fan-Out

       → Tool B
Tool A → Tool C
       → Tool D

Multiple tools execute concurrently on same input.

Example: Comprehensive Schedule Analysis

                → validate_schedule
schedule_status → detect_conflicts
                → check_utilization_threshold

Pattern 3: Map-Reduce

Tool A → [Tool B, Tool B, Tool B] → Tool C

Parallel execution followed by aggregation.

Example: Multi-Person Swap Analysis

For each faculty:
    analyze_swap_candidates → aggregate_results → rank_by_score

Pattern 4: Conditional Routing

Tool A → Decision → Tool B (if condition)
               → Tool C (else)

Example: Deployment Workflow

validate_deployment → (if valid) → promote_to_production
                   → (else)      → rollback_deployment

Key Files

File	Purpose
Workflows/tool-discovery.md	MCP endpoint scanning and capability mapping
Workflows/error-handling.md	Retry logic, fallback strategies, escalation
Workflows/tool-composition.md	DAG patterns, parallel execution, result synthesis
Reference/mcp-tool-index.md	Complete tool catalog with I/O schemas
Reference/tool-error-patterns.md	Known failure modes and workarounds
Reference/composition-examples.md	Real-world multi-tool chains

Output

This skill produces:

1. Execution Plans: DAG of tool dependencies with timing estimates
2. Error Reports: Categorized failures with recovery recommendations
3. Performance Metrics: Latency, throughput, resource usage
4. Capability Maps: Which tools can satisfy which requirements

Error Handling Strategy

See Workflows/error-handling.md for complete strategy. Key principles:

1. Retry Transient Errors: Network timeouts, rate limits, DB locks
2. Fail Fast on Permanent Errors: Invalid inputs, missing resources
3. Graceful Degradation: Use cached data or reduced functionality
4. Human Escalation: Alert on unrecoverable errors

Integration with MCP Server

The MCP server runs in Docker container mcp-server and exposes tools via:

• STDIO Transport: For Claude Desktop integration
• HTTP Transport (dev mode): Port 8080 for debugging

Health Check

docker-compose logs -f mcp-server
docker-compose exec mcp-server python -c \
  "from scheduler_mcp.server import mcp; print(f'Tools: {len(mcp.tools)}')"

API Connectivity Test

docker-compose exec mcp-server curl -s http://backend:8000/health

Common Workflows

1. Schedule Safety Check

Goal: Comprehensive validation before deployment

Parallel:
  - validate_schedule(date_range)
  - detect_conflicts(date_range)
  - check_utilization_threshold()
  - run_contingency_analysis_resilience(N-1, N-2)

Aggregate results → Generate safety report

2. Emergency Coverage Response

Goal: Handle faculty absence with minimal disruption

1. run_contingency_analysis(scenario="faculty_absence", person_ids=[...])
2. For each resolution_option:
     analyze_swap_candidates(requester_id, assignment_id)
3. execute_sacrifice_hierarchy(target_level="yellow", simulate=True)
4. get_static_fallbacks() → Identify pre-computed schedules

3. Deployment Pipeline

Goal: Safe production deployment

1. validate_deployment(env="staging", git_ref="main")
2. run_security_scan(git_ref="main")
3. If all passed:
     run_smoke_tests(env="staging", suite="full")
4. If smoke tests passed:
     promote_to_production(staging_version, approval_token)
5. Monitor: get_deployment_status(deployment_id)

4. Performance Optimization

Goal: Identify and remove low-value code

Parallel:
  - benchmark_solvers(scenario_count=20)
  - benchmark_constraints(test_schedules="historical")
  - benchmark_resilience(modules=all)
  - module_usage_analysis(entry_points=["main", "api", "scheduling"])

Aggregate → Generate cut list → ablation_study(module_path)

---

Concrete Implementation Examples

The workflows above are high-level. This section provides detailed, runnable code examples.

Example 1: Parallel Safety Check with Error Handling

"""Complete safety check implementation with error handling."""
import asyncio
from typing import Dict, List, Any
from datetime import date, datetime

async def comprehensive_safety_check(
    start_date: date,
    end_date: date,
    timeout_seconds: int = 60
) -> Dict[str, Any]:
    """
    Orchestrate parallel safety checks with robust error handling.

    Args:
        start_date: Schedule validation start date
        end_date: Schedule validation end date
        timeout_seconds: Maximum time to wait for all checks

    Returns:
        Safety report with pass/fail status and recommendations
    """
    # Phase 1: Define all checks with timeout wrapper
    async def safe_call(coro, check_name: str):
        """Wrapper that handles timeouts and exceptions."""
        try:
            result = await asyncio.wait_for(coro, timeout=30)
            return {"success": True, "data": result, "check": check_name}
        except asyncio.TimeoutError:
            return {"success": False, "error": "Timeout after 30s", "check": check_name}
        except Exception as e:
            return {: , : (e), : check_name}

    
    checks = [
        safe_call(validate_schedule_mcp(start_date, end_date), ),
        safe_call(detect_conflicts_mcp(start_date, end_date), ),
        safe_call(check_utilization_threshold_mcp(), ),
        safe_call(run_contingency_analysis_mcp([, ]), )
    ]

    results =  asyncio.gather(*checks)

    
    report = {
        : datetime.utcnow().isoformat(),
        : ,
        : (results),
        : (  r  results  r[]),
        : {},
        : [],
        : [],
        : 
    }

     result  results:
        check_name = result[]
         result[]:
            report[][check_name] = result[]

            
             check_name ==    result[].get():
                report[].append()
                report[] = 

             check_name ==   result[].get(, ) > :
                report[].append(
                    
                )
                report[] = 

             check_name ==   result[].get():
                report[].append()

        :
            
            report[].append()
            report[][check_name] = {: result[]}

    
     report[] == :
        report[] = 
        report[] = [, ]
     report[]:
        report[] = 
        report[] = [, ]
    :
        report[] = 
        report[] = []

     report

Example 2: Emergency Coverage with Tiered Strategies

"""Multi-tier emergency coverage orchestration."""

async def handle_emergency_absence(
    absent_faculty_id: str,
    absence_start: date,
    absence_end: date
) -> Dict[str, Any]:
    """
    Handle faculty absence using tiered fallback strategies.

    Strategy Tiers (in order):
    1. Swap-based coverage (least disruptive)
    2. Sacrifice hierarchy (controlled degradation)
    3. Static fallback schedule (pre-approved backup)
    4. Manual escalation (all automation failed)

    Returns:
        Resolution plan with selected strategy
    """
    response = {
        "absent_faculty_id": absent_faculty_id,
        "strategies_attempted": [],
        "selected_strategy": None,
        "execution_steps": []
    }

    # TIER 1: Try swap-based coverage
    print("Tier 1: Attempting swap matching...")
    try:
        # Get affected assignments
        contingency = await run_contingency_analysis_mcp(
            scenario="faculty_absence",
            person_ids=[absent_faculty_id],
            start_date=absence_start,
            end_date=absence_end
        )

        affected = contingency.get("affected_assignments", [])
        swap_success_count = 0

        # Try to find swaps for each affected assignment
        for assignment_id in affected[:10]:  # Limit to first 10
            try:
                candidates = await analyze_swap_candidates_mcp(
                    requester_id=absent_faculty_id,
                    assignment_id=assignment_id
                )

                if candidates.get("candidates"):
                    swap_success_count += 

             Exception  e:
                ()

        
        coverage_ratio = swap_success_count / (affected)  affected  
        response[].append({
            : ,
            : ,
            : ,
            : coverage_ratio >= 
        })

         coverage_ratio >= :
            response[] = 
            response[] = [
                ,
                ,
                
            ]
             response

     Exception  e:
        response[].append({
            : ,
            : ,
            : (e)
        })

    
    ()
    :
        sacrifice =  execute_sacrifice_hierarchy_mcp(
            target_level=,
            simulate=
        )

        has_violations = (sacrifice.get())
        response[].append({
            : ,
            : ,
            :  has_violations,
            :  has_violations
        })

          has_violations:
            response[] = 
            response[] = [
                ,
                ,
                ,
                
            ]
             response

     Exception  e:
        response[].append({
            : ,
            : ,
            : (e)
        })

    
    ()
    :
        fallbacks =  get_static_fallbacks_mcp()

         fallback  fallbacks.get(, []):
             (fallback[] <= absence_start.isoformat() 
                fallback[] >= absence_end.isoformat()):

                response[].append({
                    : ,
                    : ,
                    : fallback[],
                    : 
                })

                response[] = 
                response[] = [
                    ,
                    ,
                    ,
                    
                ]
                 response

        response[].append({
            : ,
            : ,
            : ,
            : 
        })

     Exception  e:
        response[].append({
            : ,
            : ,
            : (e)
        })

    
    ()
    response[] = 
    response[] = [
        ,
        ,
        
    ]

     response

---

Common Failure Modes and Solutions

Real-world orchestration failures and how to handle them.

Failure Mode 1: MCP Server Unresponsive

Symptoms:

• All MCP tool calls timeout
• Connection refused errors
• No response from health check

Detection:

async def check_mcp_health() -> bool:
    """Verify MCP server is responsive."""
    try:
        # Simple health check with short timeout
        result = await asyncio.wait_for(
            schedule_status_mcp(),
            timeout=5
        )
        return True
    except asyncio.TimeoutError:
        print("ERROR: MCP server not responding")
        return False
    except Exception as e:
        print(f"ERROR: MCP health check failed: {e}")
        return False

Recovery Steps:

# 1. Check container status
docker-compose ps mcp-server

# 2. Check logs for errors
docker-compose logs --tail=50 mcp-server

# 3. Restart container if needed
docker-compose restart mcp-server

# 4. Verify backend connectivity from MCP
docker-compose exec mcp-server curl http://backend:8000/health

Prevention:

• Implement health checks before orchestration
• Use circuit breaker pattern for repeated failures
• Set reasonable timeouts (30s default, not 5min)

Failure Mode 2: Partial Results from Parallel Fan-Out

Symptoms:

• Some tools succeed, others fail
• Incomplete data for aggregation
• Missing expected fields in results

Example Scenario:

# 3 out of 4 checks succeeded - what do we do?
results = await asyncio.gather(*checks, return_exceptions=True)
# [
#   {"data": {...}},           # Success
#   TimeoutError(),             # Failed
#   {"data": {...}},           # Success
#   ConnectionError()           # Failed
# ]

Solution Pattern:

async def robust_parallel_execution(
    tools: List[tuple[str, Callable]],
    min_success_ratio: float = 0.75
) -> Dict[str, Any]:
    """
    Execute tools in parallel with partial failure tolerance.

    Args:
        tools: List of (name, async_callable) tuples
        min_success_ratio: Minimum fraction that must succeed

    Returns:
        Aggregated results with success indicators

    Raises:
        OrchestraionError: If too many tools fail
    """
    tasks = [tool[1]() for tool in tools]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = []
    failures = []

    for (name, _), result in zip(tools, results):
        if isinstance(result, Exception):
            failures.append({"tool": name, "error": str(result)})
        else:
            successes.append({"tool": name, "data": result})

    success_ratio = len(successes) / len(tools)

    report = {
        "total": len(tools),
        "succeeded": len(successes),
        "failed": len(failures),
        "success_ratio": success_ratio,
        "successes": successes,
        "failures": failures
    }

    if success_ratio < min_success_ratio:
         OrchestrationError(
            
            
        )

     report

Failure Mode 3: Tool Output Schema Mismatch

Symptoms:

• KeyError when accessing expected fields
• Type errors when processing results
• Unexpected None values

Example:

# Expected schema
expected = {"is_valid": bool, "violations": list}

# Actual response (missing field)
actual = {"is_valid": True}  # Where's violations?

# Code crashes
violations = result["violations"]  # KeyError!

Solution Pattern:

from typing import Optional
from pydantic import BaseModel, Field

class ValidationResult(BaseModel):
    """Expected schema for validation tool."""
    is_valid: bool
    violations: list = Field(default_factory=list)
    timestamp: Optional[str] = None

def safe_extract(raw_result: Dict[str, Any]) -> ValidationResult:
    """
    Safely parse tool output with schema validation.

    Raises:
        ValidationError: If required fields missing or wrong type
    """
    try:
        return ValidationResult(**raw_result)
    except ValidationError as e:
        print(f"WARNING: Tool output schema mismatch: {e}")
        # Return safe default
        return ValidationResult(is_valid=False, violations=["Schema parse error"])

Failure Mode 4: Dependency Chain Breaks Mid-Execution

Symptoms:

• Tool B needs output from Tool A, but Tool A failed
• Cascading failures down the dependency chain
• Incomplete state changes

Example:

# Sequential dependency: A → B → C
result_a = await tool_a()  # Succeeds
result_b = await tool_b(result_a["value"])  # FAILS - what about C?
result_c = await tool_c(result_b["value"])  # Can't run - result_b doesn't exist

Solution Pattern:

async def execute_dependency_chain(
    tools: List[tuple[str, Callable]],
    rollback_on_failure: bool = True
) -> Dict[str, Any]:
    """
    Execute tools with dependencies, rolling back on failure.

    Args:
        tools: List of (name, tool_func) in dependency order
        rollback_on_failure: Whether to undo previous steps on failure

    Returns:
        Chain execution report
    """
    results = {}
    executed_tools = []

    for tool_name, tool_func in tools:
        try:
            # Execute with previous results as context
            result = await tool_func(results)
            results[tool_name] = result
            executed_tools.append(tool_name)

        except Exception as e:
            # Chain broken - decide rollback
            report = {
                "status": "FAILED",
                "failed_at": tool_name,
                "completed": executed_tools,
                "error": str(e)
            }

            if rollback_on_failure:
                print(f"Rolling back {len(executed_tools)} completed steps...")
                for completed_tool in reversed(executed_tools):
                    try:
                        # Call rollback if available
                        rollback_func = getattr(
                            tool_func,
                            ,
                            
                        )
                         rollback_func:
                             rollback_func(results[completed_tool])
                     Exception  rollback_error:
                        ()

                report[] = 

             report

     {
        : ,
        : executed_tools,
        : results
    }

---

Integration Examples

How to call MCP tools from different contexts.

From FastAPI Endpoint

"""MCP orchestration from API route."""
from fastapi import APIRouter, HTTPException, BackgroundTasks
from datetime import date

router = APIRouter()

@router.post("/schedule/safety-check")
async def api_safety_check(
    start_date: date,
    end_date: date,
    background_tasks: BackgroundTasks
):
    """
    Trigger safety check via API.

    For long-running checks, use background task.
    """
    try:
        # Quick check - run synchronously
        if (end_date - start_date).days <= 7:
            report = await comprehensive_safety_check(start_date, end_date)
            return report

        # Long check - run in background
        else:
            task_id = str(uuid.uuid4())
            background_tasks.add_task(
                run_safety_check_background,
                task_id,
                start_date,
                end_date
            )
            return {
                "task_id": task_id,
                "status": "RUNNING",
                "message": "Check started in background"
            }

    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Safety check failed: {str(e)}"
        )

From Celery Task

"""MCP orchestration from Celery background task."""
from celery import shared_task

@shared_task(bind=True, max_retries=3)
def celery_emergency_coverage(
    self,
    absent_faculty_id: str,
    absence_start: str,
    absence_end: str
):
    """
    Handle emergency coverage in background.

    Retries up to 3 times on transient failures.
    """
    try:
        # Celery tasks aren't async - use asyncio.run
        result = asyncio.run(
            handle_emergency_absence(
                absent_faculty_id,
                date.fromisoformat(absence_start),
                date.fromisoformat(absence_end)
            )
        )

        return {
            "status": "COMPLETED",
            "strategy": result["selected_strategy"],
            "execution_plan": result["execution_steps"]
        }

    except Exception as e:
        # Retry on transient errors
        if "timeout" in str(e).lower() or "connection" in str(e).lower():
            self.retry(exc=e, countdown=30)  # Retry after 30s
        else:
            # Permanent error - don't retry
            return {
                "status": "FAILED",
                : (e)
            }

From CLI Script

"""MCP orchestration from command-line script."""
import asyncio
import sys
from datetime import date

async def main():
    """CLI entrypoint for safety check."""
    if len(sys.argv) != 3:
        print("Usage: safety_check.py START_DATE END_DATE")
        print("Example: safety_check.py 2025-01-15 2025-02-14")
        sys.exit(1)

    start = date.fromisoformat(sys.argv[1])
    end = date.fromisoformat(sys.argv[2])

    print(f"Running safety check: {start} to {end}")
    print("-" * 60)

    report = await comprehensive_safety_check(start, end)

    print(f"\nStatus: {report['status']}")
    print(f"Recommendation: {report['recommendation']}")

    if report['blocking_issues']:
        print("\nBlocking Issues:")
        for issue in report['blocking_issues']:
            print(f"  ❌ {issue}")

    if report['warnings']:
        print("\nWarnings:")
         warning  report[]:
            ()

    ()
     step  report[]:
        ()

    
    sys.exit(  report[] ==   )

 __name__ == :
    asyncio.run(main())

---

Best Practices

1. Always check tool prerequisites before execution
2. Use background tasks for long-running operations (>30s)
3. Poll task status instead of blocking on Celery tasks
4. Implement timeouts for all MCP calls (default: 30s)
5. Log all tool inputs/outputs for debugging
6. Cache tool results when appropriate (compliance summary, static fallbacks)
7. Parallelize independent tools to reduce latency
8. Handle partial failures gracefully in fan-out patterns

Troubleshooting

MCP Server Not Responding

# Check container status
docker-compose ps mcp-server

# View logs
docker-compose logs -f mcp-server

# Restart server
docker-compose restart mcp-server

Backend API Unreachable from MCP

# Test connectivity from MCP container
docker-compose exec mcp-server curl -s http://backend:8000/health

# Check network
docker network inspect autonomous-assignment-program-manager_default

Tool Returns Unexpected Result

1. Check tool signature in Reference/mcp-tool-index.md
2. Verify input schema matches expected format
3. Check backend API logs: docker-compose logs backend
4. Review error patterns in Reference/tool-error-patterns.md

Real-World Scenario: Multi-Skill Integration

Complete end-to-end example showing MCP orchestration integrated with other skills.

Scenario: Pre-Deployment Safety Validation

User Request: "Validate Block 10 schedule before deploying to production"

Orchestration Flow:

MCP_ORCHESTRATION (this skill)
    ↓
    Invokes: comprehensive_safety_check() using MCP tools
    ↓
    ├─→ constraint-preflight (verify all constraints registered)
    ├─→ schedule-validator (ACGME compliance verification)
    └─→ safe-schedule-generation (ensure backup exists)
    ↓
    If PASS: production-incident-responder (deployment monitoring)
    If FAIL: systematic-debugger (investigate failures)

Implementation:

async def validate_block_10_deployment():
    """
    Complete pre-deployment validation orchestration.

    Integrates multiple skills and MCP tools.
    """
    print("Step 1: MCP_ORCHESTRATION - Running comprehensive safety check...")

    # Use MCP tools for parallel checks
    safety_report = await comprehensive_safety_check(
        start_date=date(2025, 2, 3),
        end_date=date(2025, 3, 2)
    )

    if safety_report["status"] == "FAIL":
        print("\n❌ Safety check FAILED")
        print("Step 2: Invoking systematic-debugger skill...")

        # Systematic debugging of failures
        for issue in safety_report["blocking_issues"]:
            print(f"  Investigating: {issue}")
            # Debugger skill would explore root causes

        return {
            "deployment_approved": False,
            "reason": "Safety check failed",
            "next_action": "Fix issues identified by systematic-debugger"
        }

    # Additional validation via other skills
    print("\n✓ MCP safety checks passed")
    print("Step 2: Running constraint-preflight...")

    # Ensure all constraints are properly registered
    preflight_ok =  run_constraint_preflight()

      preflight_ok:
         {
            : ,
            : ,
            : 
        }

    ()

    
    validation_result =  run_schedule_validator(
        start_date=date(, , ),
        end_date=date(, , )
    )

      validation_result[]:
         {
            : ,
            : ,
            : validation_result[],
            : 
        }

    ()

    
    backup_status =  check_schedule_backup()

      backup_status[]:
        ()
         create_schedule_backup()

    
    ()

     {
        : ,
        : safety_report,
        : ,
        : backup_status.get(),
        : 
    }

Output Example:

Step 1: MCP_ORCHESTRATION - Running comprehensive safety check...
  ✓ Schedule validation: PASS
  ✓ Conflict detection: PASS (0 conflicts)
  ✓ Utilization check: PASS (78.3% < 80%)
  ⚠ Contingency analysis: 1 N-1 failure detected

✓ MCP safety checks passed
Step 2: Running constraint-preflight...
  ✓ All 47 constraints registered
  ✓ No orphaned constraints detected

Step 3: Running schedule-validator...
  ✓ 80-hour rule: PASS (all weeks compliant)
  ✓ 1-in-7 rule: PASS (all residents have 1 day off per week)
  ✓ Supervision ratios: PASS

Step 4: Ensuring database backup exists...
  Backup exists: backup_20250202_143022

✅ All validation passed - DEPLOYMENT APPROVED

Next Action: Proceed to production deployment

---

Related Skills

• constraint-preflight: Verify constraints before schedule generation (integrates with validation workflow)
• safe-schedule-generation: Database backup before write operations (ensures rollback capability)
• production-incident-responder: Crisis response using MCP tools (handles deployment failures)
• systematic-debugger: Root cause analysis of tool failures (investigates orchestration errors)
• schedule-validator: ACGME compliance verification (complements MCP validation tools)
• acgme-compliance: Regulatory expertise (validates MCP tool compliance checks)

Version

• Created: 2025-12-26
• MCP Server Version: 0.1.0
• Total Tools: 36
• Backend API: FastAPI 0.109.0

---

For detailed tool signatures and schemas, see Reference/mcp-tool-index.md For error handling procedures, see Workflows/error-handling.md For composition patterns, see Reference/composition-examples.md