SkillFlow MCP服务器
将MCP工具调用链转化为可重用的自动化技能。
  
概述
SkillFlow是模型上下文协议（MCP）服务器的完整实现，它能够将工具调用链作为可重用的“技能”进行记录、管理和重放。将复杂的多步骤操作转化为单命令自动化工作流程。
核心价值观：将重复的、多步骤的操作转化为可重用的自动化技能。
✨ 主要特点
🎯 核心能力
- 📹 记录模式：自动捕获工具调用序列
- 🔄 技能创造：将录制转换为参数化技能
- ⚡ 一键执行：使用单个命令运行复杂的工作流
- 🌐 DAG执行：支持并行执行和依赖关系管理
- 💾 零数据库：基于JSON的存储（不需要数据库）
- 🔌 完整的MCP协议：完全支持工具、资源和提示
🚀 高级功能（最新）
并发控制
- 顺序模式：逐一执行步骤（默认）
- 分阶段模式：并行执行的分组步骤
- 全并行模式：最大限度地提高依赖关系中的并行性
上游MCP集成
- 代理工具：自动公开上游服务器工具
- 资源访问：列出并读取连接的MCP服务器上的资源
- 提示访问：从上游服务器检索和使用提示模板
- 本地MCP客户端：具有完全协议控制的自定义实现
SkillFlow自己的MCP端点
- 资源：通过自定义URI方案公开技能、会话和运行日志
- skill:// -访问技能定义 - session:// -访问录制会话 - run:// -访问执行日志
- 提示词：内置技能发展指南
- create_skill -循序渐进的技能创造指南 - debug_skill -调试协助 - optimize_skill -性能优化提示 - skill_best_practices -开发最佳实践
内容类型支持
完全支持MCP协议内容类型：
- ✅ 文本内容
- ✅ ImageContent（截图、图表）
- ✅ 音频内容（录音、TTS）
- ✅ 嵌入式资源（文件、数据）
第二阶段：传输层扩展✅
- HTTP+SSE传输：使用服务器发送的事件通过HTTP连接到MCP服务器
- 实时服务器到客户端通知 - RESTful工具调用 - 自动重新连接处理
- WebSocket传输：与MCP服务器进行全双工通信
- JSON-RPC 2.0协议支持 - 乒乓球保活机制 - 双向消息处理
- 灵活的运输选择：为您的用例选择最佳运输方式
- stdio：默认的基于进程的通信 - HTTP+SSE：可扩展的、基于HTTP的架构 - WebSocket：实时、持久连接
第4和第5阶段：Web UI和监控✅
- Web控制面板：用于管理SkillFlow的现代web界面
- 可在以下网址访问 http://localhost:8080 当web服务器正在运行时 - 实时WebSocket更新实时指标 - 采用Tailwind CSS的响应式设计
- 可视化DAG编辑器：用于技能的交互式图形编辑器
- 拖放节点创建 - 视觉连接生成器 - 基于Dagre算法的自动布局 - 使用Cytoscape.js进行实时图形可视化 - 节点属性编辑
- 执行监控仪表板：实时执行跟踪
- 实时指标：主动执行、吞吐量、成功率 - 绩效图表：执行时间表、分布 - 最近执行历史 - 每3秒自动刷新一次
- 技能调试工具：交互式调试界面
- 技能定义检查员 - 执行图可视化工具 - 干运行模拟 - 测试输入/输出检查器 - 执行跟踪查看器
- 交互式技能构建器：分步向导
- 4步流程：信息→ 节点→ 连接→ 审查 - 可视化节点管理 - 连接生成器 - JSON预览 - 一键技能创造
- 审计日志：全面的事件跟踪
- 记录所有技能操作 - 工具调用跟踪 - 服务器事件监控 - 严重级别（调试、信息、警告、错误、严重） - 按日期存储时间序列 - 可查询的审计跟踪
- 高级指标：性能监控
- 执行时间百分比（P50、P95、P99） - 吞吐量跟踪（每分钟执行次数） - 错误率计算 - 内存使用监控 - 并发执行跟踪 - Prometheus导出格式
阶段3：高级控制流✅
- 条件节点：技能中的动态分支逻辑
- 如果/其他：简单的条件执行 - 开关：多分支条件逻辑 - 使用JSONPath、Jinja2或Python表达式进行条件求值 - 默认分支回退支持
- 循环节点：迭代数据并重复操作
- for循环：使用JSONPath选择对集合进行迭代 - while循环：带动态求值的条件循环 - FOR_RANGE循环：数值范围迭代 - 安全限制 max_iterations 防止无限循环 - 访问循环变量： $loop.item, $loop.index
- 技能嵌套：从简单技能中组合复杂技能
- 模块化设计技能中的调用技能 - 具有适当上下文隔离的递归执行 - 将现有技能作为构建块重用
- 参数变换：动态参数生成
- JSONPath：从以前的输出中提取和转换数据 - Jinja2模板：使用模板生成复杂参数 - 可访问输入、输出和循环变量的上下文感知转换 - 模板变量： $inputs.field, @step_id.outputs.field, $loop.var_name
📦 安装
先决条件
- Python 3.11+
- 紫外线 包管理器
- MCP客户端（如克劳德桌面）
再进行
# Clone the repository
git clone
cd skillflow-mcp
# Install base dependencies
uv sync
# Optional: Install advanced features
uv sync --extra http # HTTP+SSE transport support
uv sync --extra websocket # WebSocket transport support
uv sync --extra transforms # JSONPath & Jinja2 parameter transformations
uv sync --extra web # Web UI and monitoring dashboard
uv sync --extra full # All advanced features可选依赖关系
- 超文本传输协议 (
aiohttp>=3.9.0)：上游MCP服务器的HTTP+SSE传输 - 双向通信 (
websockets>=12.0)：用于实时通信的WebSocket传输 - 变换 (
jsonpath-ng>=1.6.0,jinja2>=3.1.0)：高级参数转换 - 网络 (
fastapi>=0.109.0,uvicorn>=0.27.0,psutil>=5.9.0)：Web UI控制面板、监控仪表板和调试工具 - 满的：所有可选依赖项已合并
⚙️ 配置
Claude桌面设置
编辑您的Claude Desktop配置文件：
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json 视窗: %APPDATA%\Claude\claude_desktop_config.json Linux: ~/.config/Claude/claude_desktop_config.json
添加SkillFlow服务器：
{
"mcpServers": {
"skillflow": {
"command": "uv",
"args": ["run", "skillflow"],
"cwd": "/absolute/path/to/skillflow-mcp"
}
}
}重要：替换 cwd 与技能流mcp的实际绝对路径。
🚀 快速开始
基本工作流程
- 开始录制
Ask Claude: "Please start recording with session name 'my_workflow'"- 执行工具调用
Perform your multi-step operations through Claude- 停止录制
Ask Claude: "Please stop recording"- 创造技能
Ask Claude: "Create a skill from the last session"- 执行技能
Ask Claude: "Execute skill__my_workflow"示例：文件备份自动化
1. Start recording: "Start recording session 'backup_docs'"
2. Execute operations:
- List all .txt files in Documents
- Read the first file
- Copy content to backup directory
3. Stop recording: "Stop recording"
4. Create skill: "Create skill 'backup_first_txt' from last session"
5. Use skill: "Execute skill__backup_first_txt"📖 可用工具
录音控制
start_recording-开始新的录制会话stop_recording-停止当前录制会话
技能管理
create_skill_from_session-将录音转化为技能list_skills-列出所有可用技能get_skill-获取技能详细信息delete_skill-删除技能
执行控制
skill__-执行特定技能（为每个技能自动生成）get_run_status-检查执行状态cancel_run-取消运行执行
上游服务器管理
register_upstream_server-注册MCP服务器list_upstream_servers-列出已注册的服务器disconnect_server-断开与服务器的连接
上游资源和提示
list_upstream_resources-列出来自上游服务器的资源read_upstream_resource-阅读特定资源list_upstream_prompts-列出可用提示get_upstream_prompt-检索提示模板
调试工具
debug_recording_session-分析记录数据debug_skill_definition-检查技能结构debug_skill_execution-跟踪执行流程debug_upstream_tools-测试上游服务器连接
🏗️ 建筑
核心组件
src/skillflow/
├── schemas.py # Pydantic data models (extended for Phase 3)
├── storage.py # JSON storage layer
├── skills.py # Skill management
├── recording.py # Recording manager
├── engine.py # Execution engine (DAG, concurrency, Phase 3 features)
├── mcp_clients.py # Upstream MCP client manager
├── native_mcp_client.py # Native MCP client implementation (stdio)
├── http_sse_client.py # HTTP+SSE transport client (Phase 2)
├── websocket_client.py # WebSocket transport client (Phase 2)
├── parameter_transform.py # JSONPath & Jinja2 transformations (Phase 3)
├── tool_naming.py # Smart tool naming strategy
└── server.py # MCP server implementation数据流
Recording Flow:
start_recording() → RecordingSession → Log tool calls → stop_recording() → Save to data/sessions/
Skill Creation Flow:
create_skill_from_session() → Load session → Generate SkillGraph (nodes + edges)
→ Apply parameter templates → Save to data/skills/ → Register as MCP tool
Skill Execution Flow:
skill__(inputs) → ExecutionEngine.run_skill() → Parse DAG, topological sort
→ Execute nodes (sequential/phased/parallel) → Call upstream tools
→ Record execution to data/runs/ → Return SkillRunResult🎨 用例
1.工作流自动化
记录重复的多步操作，并用一个命令执行它们。
2.工作流编排
将来自多个MCP服务器的工具组合到复杂的工作流程中。
3.批量处理
并行执行多个独立任务（批量下载、处理）。
4.技能库
在团队中分享常见的自动化技能。
5.CI/CD集成
将技能整合到自动化部署管道中。
🔧 高级配置
并发模式
在创建技能时，您可以指定执行策略：
create_skill_from_session({
"session_id": "...",
"skill_id": "parallel_fetch",
"name": "Parallel Data Fetch",
"concurrency_mode": "full_parallel", # sequential | phased | full_parallel
"max_parallel": 5 # Limit concurrent executions
})分阶段执行
定义分组并行的执行阶段：
create_skill_from_session({
"session_id": "...",
"concurrency_mode": "phased",
"concurrency_phases": {
"phase1": ["step_1", "step_2"], # Execute in parallel
"phase2": ["step_3", "step_4"] # Execute after phase1
}
})第3阶段高级示例
条件节点
使用条件逻辑创建技能：
{
"node": {
"id": "check_status",
"kind": "conditional",
"conditional_config": {
"type": "if_else",
"branches": [
{
"condition": "$.status == 'success'",
"nodes": ["success_handler"],
"description": "Handle success case"
}
],
"default_branch": ["error_handler"]
}
}
}循环节点
在集合或范围内迭代：
{
"node": {
"id": "process_items",
"kind": "loop",
"loop_config": {
"type": "for",
"collection_path": "$.items",
"iteration_var": "current_item",
"body_nodes": ["process_single_item"],
"max_iterations": 100
}
}
}参数变换
使用JSONPath提取数据：
{
"parameter_transform": {
"engine": "jsonpath",
"expression": "$.results[*].id"
}
}使用Jinja2模板进行复杂转换：
{
"parameter_transform": {
"engine": "jinja2",
"expression": "{{ value | upper }} - {{ loop.index }}"
}
}模板变量
访问技能参数中的上下文：
$inputs.field_name-访问技能输入参数@step_id.outputs.field-访问前面步骤的输出$loop.item-当前循环项$loop.index-当前循环迭代指数
例子：
{
"args_template": {
"user_id": "$inputs.user_id",
"previous_result": "@fetch_data.outputs.result",
"item_name": "$loop.item.name"
}
}📚 文档
有关繁体中文文档，请参阅 README_ZH.md.
🛠️ 发展
运行测试
uv run pytest tests/ -v代码的风格
- 关注PEP 8
- 使用类型提示
- 为所有公共API编写文档字符串
🗺️ 路线图
第二阶段：传输层✅ 完成
- ✅ HTTP+SSE传输支持
- ✅ WebSocket传输支持
- ✅ 为上游服务器提供灵活的传输选择
第三阶段：高级功能✅ 完成
- ✅ 技能嵌套和组合
- ✅ 条件节点（if/else/switch）
- ✅ 循环节点（for/while/for_range）
- ✅ 参数转换表达式（JSONPath、Jinja2）
- ✅ 增强的模板变量（$输入、@输出、$循环）
第四阶段：审计与监控✅ 完成
- ✅ 带有事件跟踪的审核日志
- ✅ 高级监控和指标
- ✅ 性能统计和分析
- ✅ 实时指标收集
- ✅ Prometheus指标导出
第五阶段：用户体验✅ 完成
- ✅ Web UI控制面板
- ✅ 使用Cytoscape.js的可视化DAG编辑器
- ✅ 执行监控仪表板
- ✅ 技能调试工具
- ✅ 交互式技能构建向导
第6阶段：企业功能（未来）
- \[\]多租户支持
- \[\]权限和访问控制
- \[\]技能市场和分享
- \[\]高级安全功能
🤝 贡献
欢迎投稿！请遵循以下指南：
- 分叉存储库
- 创建要素分支
- 为新功能编写测试
- 提交拉取请求
测试要求
- 新功能必须包括测试
- 保持测试覆盖率>80%
📄 许可证
MIT许可证-请参阅 许可证 文件以获取详细信息。
🙏 致谢
📊 项目状态
✅ MVP完成 -准备进行生产测试
最后更新: 2025-11-16 维护者：SkillFlow团队
______________________________________________________________________
立即开始自动化您的工作流程！ 🚀