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)。

不变量

1. stdout = JSON-RPC，stderr = logs，任何 stdout 写入会破坏 MCP 客户端
2. DAO 是 SQL 边界，orchestration / fsm / spec / tools / pagination / export 永不 import pipeflow_mcp.storage.dao
3. 单写者假设，SQLite WAL 在单 host + 单 writer 下安全，多进程并发写超出设计范围
4. user_version 降级拒绝，永不静默 forward-migrate，启动时抛 RuntimeError 是正确行为
5. Annotation 列表不含 data，list 返回 AnnotationSummary，data 只能通过 get 取得
6. reset append-only，保留 annotations + reports + 历史 events，只追加 type='reset' event
7. 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。