Skills Orchestrator MCP

技能编排器MCP服务器

一个自我进化的MCP（模型上下文协议）服务器，使AI代理能够动态创建、管理和执行可重用的技能。实现Anthropic的“使用MCP执行代码”模式，以实现上下文高效的代理工具交互。

______________________________________________________________________

目录

• 愿景：与你共同成长的人工智能
• 关键能力
• 为什么这很重要：令牌效率和上下文管理
• AI代理如何利用该系统
• 可用工具
• 技能构成体系
• 调试与开发
• 配置
• 存储体系结构
• 技能类别
• 快速开始
• 使用示例
• 建筑
• 安全注意事项
• 未来路线图
• 背景与灵感
• 贡献
• 许可证

______________________________________________________________________

愿景：与你共同成长的人工智能

传统的人工智能交互是无状态的——每次对话都是从头开始的，复杂的操作会用中间数据淹没上下文窗口。技能编排者改变了这一基本限制：

之前： 人工智能在上下文中处理一切，在会话之间丢失工作，反复重新发现解决方案。

之后： 人工智能创造了积累知识的持久、可组合的技能。它减轻了繁重的计算负担，跨会话维护内存，并随着时间的推移构建领域专业知识。

这不仅仅是一个工具服务器，它还是 人工智能自我提升 和 上下文高效操作.

______________________________________________________________________

关键能力

动态技能注册表

• 创建 在对话过程中实时可重用的代码模式
• 商店 在整个课程中持续使用的技能
• 搜索 按名称、描述、类别或标签列出的技能（渐进式发现）
• 调用 参数技能
• 编排 通过调用其他技能（技能链）来获取技能
• 版本 更新时自动技能

代码执行引擎

• 执行 孤立子流程环境中的Python代码
• 注入 将上下文变量放入执行命名空间
• 捕捉 stdout、stderr和返回值
• 超时 具有真实进程边界的保护
• 调试助手 内置（debug_context、pretty_print）

状态持久性

• 商店 使用点符号键进行对话状态
• 组织 为不同的项目/域提供命名空间
• 轨道 工作流进度、偏好和决策
• 分享 技能和会话之间的状态

伪影系统（带外存储）

• 商店 无洪泛上下文的大输出
• 检索 根据需要分块显示内容
• 基于TTL 自动清理到期
• 基于手柄 参考文献artifact://...)为了提高效率

渐进式发现

• 搜索 不加载代码的元数据技能
• 过滤器 按类别、标签或文本查询
• 分页 大型技能库的结果
• 负载 仅在需要检查时提供完整定义

______________________________________________________________________

为什么这很重要：令牌效率和上下文管理

传统的MCP用法通过上下文窗口传递工具定义和结果，消耗了过多的令牌。代码执行模式解决了这个问题：

现实世界影响： Anthropic的研究表明，对于数据量大的操作，这种模式可以将上下文使用率降低98%以上。

______________________________________________________________________

AI代理如何利用该系统

1.上下文卸载

与其在对话上下文中处理大型数据集：

# 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
})

2.通过技能创造实现自我提升

当发现一个有用的模式时，将其编码为一种技能：

# 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!

3.建立领域专业知识

为不同领域创造专业技能：

• 游戏开发： 决策记录、资产管道、设计模式助手
• 数据科学： 处理管道、可视化生成器、报表生成器
• 写作： 风格一致的生成器、编辑工作流程、研究组织者
• DevOps： 部署自动机、监控分析器、事件响应者

4.记忆与连续性

状态系统启用持久内存：

# 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

5.复杂工作流程的技能构成

从简单的原语构建复杂的管道：

# 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__
}

______________________________________________________________________

可用工具（15个MCP工具）

技能管理

代码执行

工件管理

状态管理

公用事业

MCP资源

______________________________________________________________________

技能构成体系

技能可以调用其他技能，实现强大的组合模式。这将孤立的实用程序转换为可组合的构建块。

组合功能（所有技能都有）

# 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"}
#   ]
# }

增强的错误消息

当技能失败时，错误消息包括：

• 带有行号的完整回溯
• 可用上下文变量列表
• 使用提示 debug_context() 供检查

**Error:** name 'undefined_var' is not defined

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

______________________________________________________________________

## 配置

### 环境变量

|变量|默认值|描述|
|----------|---------|-------------|
| `SKILL_ORCHESTRATOR_DEFAULT_EXECUTION_TIME_SECONDS` |30|代码执行的默认超时|
| `SKILL_ORCHESTRATOR_MAX_EXECUTION_TIME_SECONDS` |180|允许的最大超时时间|
| `SKILL_ORCHESTRATOR_TERMINATION_GRACE_SECONDS` |5 |强制杀人前的宽限期|
| `SKILL_ORCHESTRATOR_MAX_CALL_DEPTH` |10|最大技能嵌套深度|
| `SKILL_ORCHESTRATOR_EXEC_MODE` |子进程|执行模式（子进程或进程内）|

### 超时配置

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

### 存储设计原则

- **渐进式披露：** 技能指数仅在需要时排除代码加载
- **写灯：** 使用情况统计在单独的文件中，以避免每次调用时重写技能
- **原子操作：** 临时文件+重命名模式以确保数据完整性
- **线程安全：** 锁定所有共享状态（索引、使用情况、工件）

______________________________________________________________________

## 技能类别

|类别|描述|示例用例|
|----------|-------------|-------------------|
| `data_processing` |数据转换和分析| CSV处理、JSON转换|
| `file_operations` |文件系统操作|批量重命名、目录组织|
| `api_integration` |外部API连接|REST客户端、webhook处理程序|
| `text_analysis` |文本处理和NLP |摘要、提取、格式化|
| `automation` |工作流自动化|构建管道、部署脚本|
| `visualization` |图表和视觉输出|数据图、图表|
| `computation` |数学和科学计算|计算、模拟|
| `workflow` |多步骤流水线|编排、链接|
| `utility` |通用工具|助手、转换器|
| `custom` |未分类|特定领域技能|

______________________________________________________________________

## 快速开始

### 安装

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

### 为Claude桌面配置

添加到您的 `claude_desktop_config.json`:

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

或者使用uvx：

{ "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} │ └─────────────────────────────────────────────────────────────────┘

### 执行流程

1. **子进程模式（默认）：** 使用运行器脚本和有效负载创建临时目录，生成子进程，通过进程边界强制执行真正的超时
1. **技能构成：** 传递给子流程的所有技能的注册表，启用 `invoke_skill()` 技能范围内的呼叫
1. **工件处理：** 将大量输出存储到磁盘，返回轻量级句柄
1. **渐进式发现：** 搜索索引时不加载代码，按需加载完整定义

______________________________________________________________________

## 安全注意事项

- 代码使用单独的Python解释器在隔离的子进程中运行
- 通过进程边界强制执行超时（不能绕过）
- 标准库访问（网络/文件操作时请谨慎使用）
- 输出截断可防止内存问题
- 具有可配置宽限期的优雅终止

Use subprocess mode (default, recommended)

SKILL_ORCHESTRATOR_EXEC_MODE=subprocess

Legacy in-process mode (not recommended - no timeout guarantee)

SKILL_ORCHESTRATOR_EXEC_MODE=inprocess

______________________________________________________________________

## 未来路线图

- \[\]基于Docker的沙盒执行，增强安全性
- \[x\] 技能组合和链接（已实现！）
- \[\]常见模式的内置技能模板
- \[\]远程技能分享和发现
- \[\]执行分析和性能跟踪
- \[\]用于API连接技能的OAuth集成
- \[\]用于流输出的WebSocket传输
- \[\]语义技能发现（基于嵌入的搜索）
- \[\]支持回滚的技能版本控制
- \[\]技能测试框架

______________________________________________________________________

## 背景与灵感

此服务器实现了以下模式：

- [Anthropic的“MCP代码执行”](https://www.anthropic.com/engineering/code-execution-with-mcp) -大幅减少上下文使用
-  -协调多个MCP服务器
- [Cloudflare的“代码模式”](https://blog.cloudflare.com/code-mode/) -TypeScript沙盒执行

______________________________________________________________________

## 贡献

欢迎投稿！请随时提交拉取请求。

## 许可证

MIT许可证-有关详细信息，请参阅许可证文件。

## 致谢

- MCP规范和代码执行模式的拟人化
- Docker对MCP网关的启示
- Cloudflare用于代码模式洞察
- FastMCP社区