Multi-Skill Collaborative Workflow |

This project integrates the first three hands-on projects into a complete "Data Report Generation" workflow.

Objective: Master multi-Skill orchestration, and learn how to design an orchestrator Skill that coordinates the entire process.

Workflow Goal

User uploads a business data file, and the system automatically completes the following end-to-end process, ultimately delivering a complete business report.

Orchestrator Skill Design Principles

The orchestrator Skill does not perform specific business processing; it only schedules other Skills' outputs, passes intermediate results, and reports progress to users at key milestones.

Orchestration Script

Example

# File path: scripts/orchestrator.py

# Multi-Skill collaborative orchestration entry script

import subprocess

import sys

import json

import os

import shutil

SKILLS_BASE = "/mnt/skills/public"

WORK_DIR    = "/home/claude/report_work"

OUTPUT_DIR  = "/mnt/user-data/outputs"

def run(script_path: str, *args) -> dict:

    """Run specified script, return JSON result"""

    cmd = [sys.executable, script_path] + list(args)

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)

    try:

        return json.loads(result.stdout)

    except json.JSONDecodeError:

        return {

            "status": "error",

            "message": result.stderr or result.stdout or "Script produced no output"

        }

def orchestrate(input_file: str) -> dict:

    """Execute complete report generation workflow"""

    os.makedirs(WORK_DIR, exist_ok=True)

    # ── Stage 1: Data Cleaning ─────────────────────────────────

    print("【1/3】Cleaning data...", flush=True)

    clean_result = run(

        f"{SKILLS_BASE}/data-cleaner/scripts/clean_data.py",

        input_file,

        os.path.join(WORK_DIR, "cleaned.csv")

    )

    if clean_result.get("status") != "success":

        return {"status": "error", "stage": "Data Cleaning",

                "message": clean_result.get("message")}

    print(f"  Cleaned: removed {clean_result.get('dup_removed', 0)} duplicate rows, "

          f"filled {clean_result.get('nulls_filled', 0)} null values", flush=True)

    # ── Stage 2: Statistical Analysis ─────────────────────────────────

    print("【2/3】Generating statistical report...", flush=True)

    stats_result = run(

        f"{SKILLS_BASE}/data-cleaner/scripts/calc_stats.py",

        clean_result,

        os.path.join(WORK_DIR, "stats.json")

    )

    if stats_result.get("status") != "success":

        return {"status": "error", "stage": "Statistical Analysis",

                "message": stats_result.get("message")}

    # ── Stage 3: Generate Excel Report ──────────────────────────

    print("【3/3】Generating Excel report...", flush=True)

    output_name = os.path.splitext(os.path.basename(input_file)) + "_report.xlsx"

    output_path = os.path.join(OUTPUT_DIR, output_name)

    report_result = run(

        f"{SKILLS_BASE}/data-cleaner/scripts/gen_report.py",

        clean_result,

        stats_result.get("output", ""),

        output_path

    )

    if report_result.get("status") != "success":

        return {"status": "error", "stage": "Report Generation",

                "message": report_result.get("message")}

    # Clean up working directory

    shutil.rmtree(WORK_DIR, ignore_errors=True)

    return {

        "status": "success",

        "output":  output_path,

        "clean_stats": {

            "dup_removed": clean_result.get("dup_removed", 0),

            "nulls_filled": clean_result.get("nulls_filled", 0)

        }

    }

if __name__ == "__main__":

    if len(sys.argv) < 2:

        print(json.dumps({"status": "error", "message": "Usage: python orchestrator.py "}))

        sys.exit(1)

    result = orchestrate(sys.argv)

    print(json.dumps(result, ensure_ascii=False, indent=2))

【1/3】Cleaning data...  Cleaned: removed 5 duplicate rows, filled 18 null values
【2/3】Generating statistical report...
【3/3】Generating Excel report...
{
  "status": "success",
  "output": "/mnt/user-data/outputs/tutorial_sales_report.xlsx",
  "clean_stats": {"dup_removed": 5, "nulls_filled": 18}
}

Orchestrator Skill's SKILL.md

---
name: report-pipeline
version: 1.0.0
description: >
  Complete data report generation pipeline: automatically performs data cleaning,
  statistical analysis, and generates Excel report for uploaded CSV/Excel files.
  Trigger when user needs one-click report generation or direct analytical report
  from raw data.
---

# Data Report Pipeline

## Prerequisites

Requires the following Skills to be installed:
- data-cleaner v1.0+

This Skill's scripts are located at /mnt/skills/public/report-pipeline/

## Execution

After obtaining the user-uploaded file path, run the orchestration script directly:

```bash
python scripts/orchestrator.py 

The script outputs real-time progress for each stage; upon completion, outputs the final file path.
After parsing the JSON output, call present_files to display the report file.

## Error Handling

If any stage fails, output contains error information including stage and message,
informing the user which stage failed and possible causes; do not proceed with subsequent stages.