pipeflow-mcp

PipeFlow MCP Server —— AI Agent 任务的被动状态跟踪容器。

注意：本项目是实验 / demo 性质的学习作品，作者发布它用于代码评审与学习，真实场景下的实用价值尚未确证，不保证有用。

---

这是什么

pipeflow-mcp 是一个基于 Model Context Protocol (stdio JSON-RPC) 的 MCP 服务器。它把 agent 的工作流执行过程记录到一个由 FSM 驱动的 pipeline 账本里，提供：

FSM 状态机：5 状态 × 5 action，pending / running / succeeded / failed / skipped，由 transitions 库实现
审计完整性：reset 保留所有 annotations + reports，delete 是 soft-delete，events 表 append-only
重试与时间预算：retry 计数器 lifetime，stage-level time_budget_seconds 派生 time_status
*16 个 pipeflow_ 工具**：声明 / 状态推进 / 读取 / 注解 / 报告 / 仪表盘 / 健康检查，其中 pipeflow_stage_advance 是唯一的状态变更入口
Cursor 分页：4 个 list 工具共享 base64-urlsafe(last_id + filter_fingerprint) 不透明 cursor，filter 改变即 cursor 失效
静态 HTML 仪表盘：导出产物含 7 个 section —— Pipeline Overview / Decision Chain / Reflection Trace / Iteration Compare / Metric Analysis / Deliverables / Events Stream，Jinja2 autoescape 防 XSS，导出路径白名单
错误脱敏：客户端消息只含 pipeline_id / stage_id / 错误类名，SQL 片段 / 文件路径 / 堆栈仅走 stderr 结构化日志
Schema 校验：5 个 ReportType —— decision / reflection / metric_report / deliverable / iteration_summary，对应 JSON Schema，启动时注册
类型化响应：8 个 Pydantic v2 view model，FastMCP outputSchema 自动填充字段契约，extra='forbid' 锁定公开 JSON 契约

完整 API 文档见 docs/api.md。

---

状态

| 指标 | 值 | |---|---| | 版本 | v0.1.0 | | MCP 工具 | 16 个 pipeflow_*，2 declare · 1 state_advance · 4 read · 3 annotation · 2 report · 3 dashboard · 1 health | | 测试 | 584 passing，unit + integration + e2e + stdio smoke | | 静态检查 | ruff ✅ · mypy --strict ✅ | | Python | 3.11 - 3.14，运行时最低 3.11，CI 矩阵 3.13 + 3.14 | | 存储 | SQLite WAL，单写者假设，多进程并发写不在设计范围内 |

---

安装与运行

MCP 客户端

从 GitHub 直接安装：

uvx --from git+https://github.com/AstralVoidZ/pipeflow-mcp pipeflow-mcp

或者从本地源码 checkout 启动：

git clone https://github.com/AstralVoidZ/pipeflow-mcp.git
cd pipeflow-mcp
uv venv --python 3.14 .venv
uv pip install -e ".[dev]"
uvx --from . pipeflow-mcp    # dev 模式

Server 通过 stdio 监听 JSON-RPC，stdout 保留给 JSON-RPC，所有日志走 stderr。

客户端配置示例

Claude Code，~/.claude/mcp_servers.json：

{
  "mcpServers": {
    "pipeflow": {
      "command": "uvx",
      "args": ["--from", "git+https://github.com/AstralVoidZ/pipeflow-mcp", "pipeflow-mcp"],
      "env": {
        "PIPEFLOW_DB_PATH": "~/.pipeflow_mcp/state.db",
        "PIPEFLOW_LOG_LEVEL": "INFO"
      }
    }
  }
}

opencode，项目级 .opencode/config.json：

{
  "mcp": {
    "pipeflow": {
      "type": "local",
      "command": ["uvx", "--from", "git+https://github.com/AstralVoidZ/pipeflow-mcp", "pipeflow-mcp"],
      "environment": {
        "PIPEFLOW_DB_PATH": "~/.pipeflow_mcp/state.db",
        "PIPEFLOW_LOG_LEVEL": "INFO"
      },
      "enabled": true
    }
  }
}

---

工具矩阵

| Category | Tool | 说明 | |---|---|---| | Declare | pipeflow_create | 一次性创建 pipeline + stages，支持 yaml_spec，与 name / stages / metadata 互斥 | | | pipeflow_stage_append | 追加 stage 到已存在 pipeline | | State | pipeflow_stage_advance | 唯一状态变更入口，start / complete / fail / skip / retry | | Read | pipeflow_get_state | 单 pipeline 当前状态 | | | pipeflow_get_snapshot | 完整快照，含 reports 聚合 + time_status | | | pipeflow_list | pipelines 列表，metadata_filter + cursor | | | pipeflow_history | events 流，stage_id / event_types / since / until + cursor | | Annotation | pipeflow_annotation_upsert | upsert 注解，决策记录 | | | pipeflow_annotation_list | 注解索引，不含 data，3 层梯度 | | | pipeflow_annotation_get | 注解完整 payload | | Report | pipeflow_report_submit | 提交 5 种类型报告，schema 校验 | | | pipeflow_report_list | 报告列表，type / iteration / cursor | | Dashboard | pipeflow_reset | append-only reset，保留注解 + 报告 | | | pipeflow_delete | soft-delete，list 排除，history 保留 | | | pipeflow_report_export | 7 段 HTML 仪表盘导出，白名单路径 | | Health | pipeflow_health | liveness + readiness + uptime，PRAGMA user_version DB 检查 |

完整参数 / 返回 schema / 错误清单见 docs/api.md。

---

配置

pipeflow-mcp 不接受任何 CLI 参数，无 --db-path 之类的 flag。所有配置通过环境变量传入，客户端在 MCP server 配置的 env 块里设置即可。误传 --db-path 只会在 cwd 留下一个同名空文件。

| 变量 | 默认值 | 说明 | |------|--------|------| | PIPEFLOW_DB_PATH | ~/.pipeflow_mcp/state.db | SQLite 数据库文件路径，首次启动自动 init_db | | PIPEFLOW_EXPORT_DIR | ~/.pipeflow_mcp/exports/ | pipeflow_report_export 写入根，白名单校验 | | PIPEFLOW_LOG_LEVEL | INFO | 日志级别，DEBUG / INFO / WARNING / ERROR | | PIPEFLOW_LOG_FORMAT | text | stderr 日志格式，设为 json 切换为结构化 JSON lines，字段 timestamp / level / logger / message / request_id / pipeline_id / stage_id / tool |

所有日志写入 stderr，stdout 保留给 JSON-RPC 通信。

---

架构

5 层严格单向依赖：

protocol (MCP tools / server.py)
   → view layer (Pydantic v2 view models, extra='forbid')
      → persistence orchestration (orchestration/ — advance, create, append, reset, delete, snapshot, annotate, reports, queries, ...)
         → { FSM layer (fsm/machine.py) , Spec layer (yaml_parser, report_registry, result_validator) }
            → data contract layer (models/*.py — dataclasses only)
               → error layer (errors/*.py)

侧模块，storage/ (DB + DAO + migrations) · pagination/ (cursor codec) · export/ (Jinja2 + 8 模板) · tools/ (16 薄封装) · config.py (env + logging)。

不变量

stdout = JSON-RPC，stderr = logs，任何 stdout 写入会破坏 MCP 客户端
DAO 是 SQL 边界，orchestration / fsm / spec / tools / pagination / export 永不 import pipeflow_mcp.storage.dao
单写者假设，SQLite WAL 在单 host + 单 writer 下安全，多进程并发写超出设计范围
user_version 降级拒绝，永不静默 forward-migrate，启动时抛 RuntimeError 是正确行为
Annotation 列表不含 data，list 返回 AnnotationSummary，data 只能通过 get 取得
reset append-only，保留 annotations + reports + 历史 events，只追加 type='reset' event
View 单向投影，tools → views → models 是唯一允许方向，反向 import 被 test_views_have_no_dao_imports 强制禁止

---

开发

本地验证

ruff check src/pipeflow_mcp/ tests/
mypy src/pipeflow_mcp/
pytest tests/ -v --cov=pipeflow_mcp --cov-fail-under=85

打包

本地构建 wheel + sdist 到 release/ 目录，.gitignore 排除。

./scripts/build.sh            # clean + lint + type + test + build
./scripts/build.sh --wheel    # 只构建 wheel
./scripts/build.sh --check    # 跳过 clean，直接基于当前 venv 构建

产物 release/pipeflow_mcp-<version>-*.whl 和 release/pipeflow_mcp-<version>.tar.gz，用于个人 GitHub Release 上传。

---

许可证

MIT，见 LICENSE。

pipeflow-mcp

pipeflow-mcp

这是什么

状态

安装与运行

MCP 客户端

客户端配置示例

工具矩阵

配置

架构

不变量

开发

本地验证

打包

许可证

Related MCP servers

MCP servers by category