技能编排服务

# DON'T: Pass 10MB CSV through context
# DO: Process locally, return summary
await execute_code({
    "code": """
import pandas as pd
df = pd.read_csv('/path/to/large.csv')
summary = {
    'rows': len(df),
    'columns': list(df.columns),
    'sample': df.head(3).to_dict()
}
__result__ = summary
""",
    "return_mode": "summary"  # Returns preview + artifact handle
})

# Discovered a good codebase mapping approach? Save it:
await create_skill({
    "name": "context_efficient_codebase_mapper",
    "description": "Maps unfamiliar codebases without flooding context",
    "code": "...",  # The pattern you discovered
    "tags": ["codebase", "analysis", "context-efficiency"]
})
# Now available in all future conversations!

# End of session - save context
await set_state({
    "key": "session.last_task",
    "value": {"file": "server.py", "line": 450, "task": "implementing composition"},
    "namespace": "project_x"
})

# Next session - recall context
state = await get_state("session.last_task", "project_x")
# AI knows exactly where you left off

# A parent skill that orchestrates multiple child skills
comprehension = invoke_skill('domain_context_switcher', {'action': 'detect', 'message': user_input})
session_data = invoke_skill('persistent_session_context', {'action': 'recall', 'domain': comprehension['domain']})
tasks = invoke_skill('omni_task_manager', {'action': 'list', 'domain': comprehension['domain']})

# Combine results into intelligent response
__result__ = {
    'context': comprehension,
    'session': session_data,
    'relevant_tasks': tasks,
    'call_depth': __call_depth__
}

# Call another skill and get its __result__
result = invoke_skill('child_skill', {'param': 'value'}, {'context': 'data'})

# List all available skills (name -> description dict)
skills = list_available_skills()

# Get metadata about a skill without invoking it
info = get_skill_info('some_skill')

# Context variables available in all skills:
# __call_depth__ - Current nesting depth (1 for top-level, 2 for first child, etc.)
# __parent_skill__ - Name of the calling skill (None for top-level)
# __skill_name__ - Current skill's name

# Parent skill that orchestrates multiple children
await create_skill({
    "name": "smart_assistant",
    "description": "Context-aware assistant using skill composition",
    "code": """
message = params.get('message', '')

# 1. Understand the domain context
context = invoke_skill('domain_context_switcher', {
    'action': 'detect',
    'message': message
})

# 2. Load session history
session = invoke_skill('persistent_session_context', {
    'action': 'recall',
    'domain': context.get('primary_domain')
})

# 3. Get relevant tasks
tasks = invoke_skill('omni_task_manager', {
    'action': 'list',
    'filters': {'domain': context.get('primary_domain')}
})

__result__ = {
    'response': f'Processed: {message}',
    'domain_context': context,
    'session_history': session,
    'relevant_tasks': tasks,
    'composition_depth': __call_depth__
}
""",
    "category": "workflow",
    "tags": ["assistant", "composition", "context-aware"]
})

# Inspect all available variables in execution context
debug_context(**locals())
# Output:
# === Execution Context Debug ===
#   params: dict = {'file': 'data.csv'}
#   context: dict = {'env': 'production'}
#   __call_depth__: int = 1
# ===============================

# Pretty-print JSON-serializable objects
pretty_print({"users": [{"name": "Alice"}, {"name": "Bob"}]})
# Output:
# {
#   "users": [
#     {"name": "Alice"},
#     {"name": "Bob"}
#   ]
# }

**Error:** name 'undefined_var' is not defined
```traceback...```

**Available context variables:** params, context, __call_depth__, custom_var
**Tip:** Use `debug_context(**locals())` in your skill code to inspect available variables.

# Per-call timeout (up to max)
await invoke_skill({
    "skill_name": "long_running_skill",
    "timeout": 150,  # seconds
    "return_mode": "full"
})

~/.skill_orchestrator/
├── skills/
│   ├── skill_name.json          # Full skill definition (name, code, metadata)
│   ├── .skill_index.json        # Metadata-only index (fast search, excludes code)
│   └── .skill_usage.json        # Invocation counts (write-light tracking)
├── state/
│   └── namespace_name.json      # Key-value pairs for each namespace
├── sessions/                     # Reserved for future use
└── artifacts/
    ├── {id}.meta.json           # Artifact metadata (MIME type, size, TTL)
    └── {id}.data                # Artifact binary content

# Clone the repository
git clone https://github.com/your-username/skill-orchestrator-mcp
cd skill-orchestrator-mcp

# Install with uv (recommended)
uv venv
source .venv/bin/activate  # or .venv\Scripts\activate on Windows
uv sync

# Using the CLI
skill-orchestrator-mcp

# Or directly with Python
python -m skill_orchestrator_mcp.server

# With uvx for development
uvx --from . skill-orchestrator-mcp

{
  "mcpServers": {
    "skill-orchestrator": {
      "command": "path/to/.venv/bin/skill-orchestrator-mcp"
    }
  }
}

{
  "mcpServers": {
    "skill-orchestrator": {
      "command": "uvx",
      "args": ["--from", "/path/to/skill-orchestrator-mcp", "skill-orchestrator-mcp"]
    }
  }
}

await create_skill({
    "name": "process_csv_summary",
    "description": "Load a CSV and return summary statistics without flooding context",
    "code": """
import csv
from pathlib import Path

file_path = params.get('file_path')
sample_rows = params.get('sample_rows', 5)

with open(file_path, 'r') as f:
    reader = csv.DictReader(f)
    rows = list(reader)

__result__ = {
    "file": Path(file_path).name,
    "row_count": len(rows),
    "columns": list(rows[0].keys()) if rows else [],
    "sample": rows[:sample_rows]
}
""",
    "category": "data_processing",
    "tags": ["csv", "data", "summary"],
    "examples": ["Summarize the sales.csv file", "Show me what's in data.csv"]
})

# Search by metadata - doesn't load code (context efficient)
await search_skills({
    "query": "csv",
    "category": "data_processing",
    "limit": 10,
    "response_format": "json",
    "fields": ["name", "description", "tags"]
})

# Only load code when you need to inspect/modify
await get_skill("process_csv_summary")

await invoke_skill({
    "skill_name": "process_csv_summary",
    "parameters": {
        "file_path": "/path/to/data.csv",
        "sample_rows": 3
    },
    "return_mode": "summary",  # preview + artifact handle (default)
    "timeout": 60
})

await execute_code({
    "code": """
# Process data without passing through context
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
filtered = [x for x in data if x > 5]
total = sum(filtered)

print(f"Filtered: {filtered}")
print(f"Sum: {total}")

__result__ = {"filtered": filtered, "sum": total}
""",
    "timeout": 10,
    "return_mode": "summary"
})

# Store workflow progress
await set_state({
    "key": "workflow.current_step",
    "value": {"step": 3, "status": "processing", "started": "2025-01-16T10:00:00Z"},
    "namespace": "my_project"
})

# Retrieve later (even in a different conversation)
state = await get_state("workflow.current_step", "my_project")

# List all keys in a namespace
await list_state({
    "namespace": "my_project",
    "response_format": "json"
})

┌─────────────────────────────────────────────────────────────────┐
│                      Claude / AI Agent                          │
├─────────────────────────────────────────────────────────────────┤
│                     MCP Protocol Layer                          │
├─────────────────────────────────────────────────────────────────┤
│                Skill Orchestrator MCP Server                    │
├────────────┬────────────┬────────────┬────────────┬────────────┤
│   Skill    │    Code    │   State    │  Artifact  │   Usage    │
│  Registry  │  Executor  │  Manager   │   Store    │  Tracker   │
│            │ (subprocess)│            │ (chunked)  │ (batched)  │
├────────────┴────────────┴────────────┴────────────┴────────────┤
│                      Skill Index                                │
│              (metadata-only, progressive disclosure)            │
├─────────────────────────────────────────────────────────────────┤
│                   File System Storage                           │
│          ~/.skill_orchestrator/{skills,state,artifacts}         │
└─────────────────────────────────────────────────────────────────┘

# Use subprocess mode (default, recommended)
# SKILL_ORCHESTRATOR_EXEC_MODE=subprocess

# Legacy in-process mode (not recommended - no timeout guarantee)
# SKILL_ORCHESTRATOR_EXEC_MODE=inprocess