AI导哈

自动化技能转换服务器

SkillFlow MCP Server 是一个基于模型上下文协议（MCP）的服务器，能够将工具调用链转换为可重用的自动化技能，适用于工作流自动化和批处理场景。

  • by win10ogod
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 win10ogod/skillflow-mcp

使用说明

SkillFlow MCP服务器

将MCP工具调用链转化为可重用的自动化技能。

概述

SkillFlow是模型上下文协议（MCP）服务器的完整实现，它能够将工具调用链作为可重用的“技能”进行记录、管理和重放。将复杂的多步骤操作转化为单命令自动化工作流程。

核心价值观：将重复的、多步骤的操作转化为可重用的自动化技能。

✨ 主要特点

🎯 核心能力

  • 📹 记录模式：自动捕获工具调用序列
  • 🔄 技能创造：将录制转换为参数化技能
  • ⚡ 一键执行：使用单个命令运行复杂的工作流
  • 🌐 DAG执行：支持并行执行和依赖关系管理
  • 💾 零数据库：基于JSON的存储（不需要数据库）
  • 🔌 完整的MCP协议：完全支持工具、资源和提示

🚀 高级功能（最新）

并发控制

  • 顺序模式：逐一执行步骤（默认）
  • 分阶段模式：并行执行的分组步骤
  • 全并行模式：最大限度地提高依赖关系中的并行性

上游MCP集成

  • 代理工具：自动公开上游服务器工具
  • 资源访问：列出并读取连接的MCP服务器上的资源
  • 提示访问：从上游服务器检索和使用提示模板
  • 本地MCP客户端：具有完全协议控制的自定义实现

SkillFlow自己的MCP端点

  • 资源：通过自定义URI方案公开技能、会话和运行日志
    • skill://<skill_id> -访问技能定义
    • session://<session_id> -访问录制会话
    • run://<run_id> -访问执行日志
  • 提示词：内置技能发展指南
    • 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 <repository-url>
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的实际绝对路径。

🚀 快速开始

基本工作流程

  1. 开始录制

    Ask Claude: "Please start recording with session name 'my_workflow'"

  2. 执行工具调用

    Perform your multi-step operations through Claude

  3. 停止录制

    Ask Claude: "Please stop recording"

  4. 创造技能

    Ask Claude: "Create a skill from the last session"

  5. 执行技能

    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__<id> -执行特定技能（为每个技能自动生成）
  • 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__<id>(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阶段：企业功能（未来）

  • []多租户支持
  • []权限和访问控制
  • []技能市场和分享
  • []高级安全功能

🤝 贡献

欢迎投稿！请遵循以下指南：

  1. 分叉存储库
  2. 创建要素分支
  3. 为新功能编写测试
  4. 提交拉取请求

测试要求

  • 新功能必须包括测试
  • 保持测试覆盖率>80%

📄 许可证

MIT许可证-请参阅 许可证 文件以获取详细信息。

🙏 致谢

📊 项目状态

✅ MVP完成 -准备进行生产测试

最后更新: 2025-11-16 维护者：SkillFlow团队

立即开始自动化您的工作流程！ 🚀

相关 MCP