Skip to main content
Glama
LUKAWI
by LUKAWI

Super Plumber 🚰 — 工作流拓扑图管理工具

把"任务文档"变成 agent 能原生理解的拓扑图:节点是任务、边是依赖、状态机管生命周期。 一条命令装好,CLI / MCP / Web UI 三层访问,纯 YAML 文件存储(无数据库、无服务端)。

npm version License Tests GitHub

English: README.en.md · npm: @lukawi/super-plumber


为什么需要它?

普通的 Todo 列表只有一行行文字——没有依赖顺序、没有验收标准、没有生命周期。当任务交给 AI agent 执行时,它看不懂你的任务文档,只能靠猜。

Super Plumber 把工作流重构为

todo: "做个注册模块"            →     entry → l1_register → l1_login → exit
                                      ↳ 每个节点 = plan + checkpoints + 验收标准
                                      ↳ 每条边 = 类型化依赖(谁先谁后、谁验证谁)
                                      ↳ 每个状态 = 状态机强制流转(不能跳步)
  • 对人类:一目了然的结构、实时可视化的 Web UI、可提交进 Git 的纯文本文件

  • 对 AI agent:通过 MCP 直接读图、认领任务、上报进度——每个节点是一个"压缩包"(计划 + 检查点 + 交接单),agent 不需要猜


Related MCP server: mmc-mcp

特性一览

能力

说明

🧭 类型化拓扑

7 种边类型:depends_on / validates 参与拓扑排序,shares_context / fan_out / fan_in / fallback / iterates 表达运行时控制流

🔄 状态机强制

7 态 14 个合法转换(pending → ready → running → passed/…),非法跳步直接报错,绝不静默

🤖 MCP 原生接入

9 个 graph_* 工具,zod 参数校验,供 Claude Code / opencode 等 agent 直接调用

🌐 Web 可视化

力导向图 + 边类型着色 + running 节点光点流动 + checkpoint 进度条,WebSocket 增量推送

📁 纯文件存储

每个节点/边一个 YAML 文件,Git 是唯一真相源,人类可直接编辑,无数据库

🧩 agent 协作协议

内置 plumber-flow skill(5 阶段协议)+ 2 个专用 subagent(拆解 / 裁决)


安装(保姆式)

前置要求

依赖

版本

检查方法

Node.js

≥ 20(含 npm)

node --version

平台

Windows / macOS / Linux

没有 Node.js?去 nodejs.org 下载 LTS 版本安装(一路默认下一步即可)。

1. 全局安装

npm install -g @lukawi/super-plumber

公司内网 / 代理环境装不上?先确认 npm registry 可达:

npm config get registry        # 应为 https://registry.npmjs.org/
npm install -g @lukawi/super-plumber --registry=https://registry.npmjs.org

2. 验证安装

graph --version     # 输出 0.1.0 即成功
graph --help        # 查看全部 11 个命令
which graph         # 确认命令位置(Windows: where graph)

3. 找个空目录试一下

mkdir ~/my-first-graph && cd ~/my-first-graph
graph init -l "我的第一个拓扑图"

看到 ✅ 已初始化 .graph/ 目录 就成功了。此时目录里多了一个 .graph/ 文件夹——这就是你的图。


快速开始(2 分钟建一张图)

# 1. 初始化
graph init -l "用户注册模块"

# 2. 创建节点(-i id、-l 标签、--level 层级、--plan-desc 计划、--dod 完成标准可重复)
graph create-node -i l1_register -l "注册功能" -t task --level 1 \
  --plan-desc "实现邮箱+密码注册" --dod "注册接口可用" --dod "密码加密存储"

graph create-node -i l1_login -l "登录功能" -t task --level 1 \
  --plan-desc "实现登录与会话" --dod "登录接口可用"

# 3. 添加依赖边(l1_login 依赖 l1_register)
graph add-edge -i e1 -s l1_register -t l1_login --type depends_on

# 4. 查看状态
graph status

# 5. 校验(引用完整性 + 拓扑排序 + 环检测)
graph validate

# 6. 可视化
graph serve    # 浏览器打开 http://localhost:8934

完整教程:从零到交付一张图

以"用户注册模块"为例,走一遍完整生命周期。

第 1 步:初始化并设计入口/出口

graph init -l "用户注册模块"

一张图有且仅有一个入口(entry,表达需求)和一个出口(exit,表达验收标准),都在 level 0。CLI 目前没有专门的 entry/exit 命令,直接用编辑器打开 .graph/graph.yaml 填写:

entry:
  description: "开发用户注册模块,支持邮箱+密码注册与登录"
  defined_by: human
  level: 0
exit:
  description: "可用的注册/登录功能,全部测试通过"
  acceptance_criteria:
    - "注册接口可用"
    - "登录后能保持会话"
  defined_by: human
  level: 0

graph validate 会提醒 entry/exit 为空——这是提示,不是错误;填上后警告消失。

第 2 步:创建节点(每个节点是一个"压缩包")

一个节点 = id + label + plan(计划)+ checkpoints(检查点)+ definition_of_done(验收标准)

# 主干节点:带计划、完成标准、负责人
graph create-node -i l1_register -l "注册功能" -t task --level 1 \
  --plan-desc "实现邮箱+密码注册:接口、校验、存储" \
  --dod "注册接口返回 200" --dod "密码 bcrypt 加密" --dod "重复邮箱报错" \
  --assigned-to "backend-agent"

# 子节点:细化到可执行粒度
graph create-node -i l2_reg_api -l "注册接口" -t task --level 2 \
  --plan-desc "POST /register 接口" --dod "接口测试通过"

graph create-node -i l2_reg_store -l "用户存储" -t task --level 2 \
  --plan-desc "用户表 + 密码加密" --dod "存储层测试通过"

给节点加检查点(checkpoint,执行时逐步上报的子步骤):

graph update-node -i l1_register \
  --add-checkpoint '{"id":"cp1","label":"接口开发"}' \
  --add-checkpoint '{"id":"cp2","label":"密码加密"}' \
  --add-checkpoint '{"id":"cp3","label":"联调测试"}'

# 查看节点完整内容
graph update-node -i l1_register --show

第 3 步:连接边(7 种类型任选)

graph add-edge -i e1 -s l1_register -t l1_login --type depends_on
graph add-edge -i e2 -s l2_reg_api -t l2_reg_store --type depends_on
graph add-edge -i e3 -s l2_reg_api -t l1_register --type validates   # 验证关系
graph add-edge -i e4 -s l1_register -t l1_login --type shares_context # 共享上下文

边类型

语义

参与拓扑排序

depends_on

顺序依赖:B 依赖 A 完成

validates

验证关系:A 的输出由 B 验证

shares_context

A 的输出作为 B 的输入上下文

fan_out

A 完成后多个下游可并行

fan_in

多个上游都完成后 C 才可执行

fallback

B 失败时回退到 A 重试

iterates

A ⇄ B 反复迭代优化

depends_on / validates 参与拓扑排序;其余边表达运行时控制流,排序自动忽略(如 fallback 的逆向引用不会误报成环)。

第 4 步:校验

graph validate

预期输出:

✅ 节点: 3 个
✅ 边: 4 条
✅ 拓扑排序: 3 节点通过
✅ 循环检测: 无环路
📊 校验结果: 0 错误, 0 警告

故意制造一个环试试graph add-edge -i e_cycle -s l1_login -t l1_register --type depends_on validate 会明确报出 检测到循环依赖: l1_register → l1_login → l1_register——工具不会让你带着环上路。

第 5 步:执行(状态机驱动,agent 或人认领)

# 状态机:pending → ready → running → passed
graph update-status -i l1_register -s ready      # 前置完成,进入待执行
graph update-status -i l1_register -s running    # 认领(claim):记录开始时间(记录执行者需 MCP 传 claim_by)
graph update-status -i l1_register -s passed     # 完成

# 试试非法跳步——会被状态机拦住:
graph update-status -i l1_login -s running
# ❌ Invalid transition: pending → running. Allowed: [ready, cancelled]

完整状态机:pending → ready → running → passed → blockedrunning → failed → pending(重试,自动累加 attempts),任意状态 → cancelled

上报 checkpoint 和交接单(CLI 没有这两个命令,用 MCP 工具或 skill 脚本):

方式一 · MCP 工具(需已接入 agent,见下文 MCP 章节):

// graph_update_checkpoint: { node_id: "l1_register", checkpoint_id: "cp1", status: "passed" }
// graph_update_execution_report: { node_id: "l1_register", summary: "注册功能完成", artifacts: ["dist/register.js"] }

方式二 · skill 脚本(脚本随 plumber-flow skill 提供,pi 用户位于 ~/.pi/agent/skills/plumber-flow/scripts/,项目内为 .pi/skills/plumber-flow/scripts/):

SCRIPTS=~/.pi/agent/skills/plumber-flow/scripts

# 认领 ready 节点(记录 claim_by + started_at;非 ready 节点会被状态机拦截)
node $SCRIPTS/graph-claim.mjs l1_register backend-agent

# 每完成一个检查点就上报一次(报告完的进度不会丢)
node $SCRIPTS/graph-checkpoint.mjs l1_register cp1 passed

# 交付交接单(summary + artifacts + blockers + notes)
node $SCRIPTS/graph-report.mjs l1_register "注册功能完成" "dist/register.js,test/register.test.js" "" "密码加密采用 bcrypt"

第 6 步:可视化与分享

graph export --mermaid -o flow.mmd    # 导出 Mermaid 流程图
graph serve                           # 打开 http://localhost:8934 看力导向图

CLI 命令参考(11 个)

命令

功能

常用参数

graph init

初始化 .graph/ 骨架

-l <label> 图名称;--force 已初始化时强制重置

graph create-node

创建节点

-i <id> -l <label> -t <type>(task/checkpoint/decision/gate)--level <n> --plan-desc <text> --dod <item>(可多次)--assigned-to <agent>

graph add-edge

添加边

-i <id> -s <source> -t <target> --type <7种之一>

graph update-status

状态流转(状态机校验)

-i <id> -s <status>(pending/ready/running/passed/failed/blocked/cancelled)

graph update-node

更新节点详情

-i <id> --plan-desc --add-dod <item>(可多次)--clear-dod --add-checkpoint '<JSON>'(可多次)--set-assigned <agent> --show

graph delete-node

软删除节点

-i <id>(保留 .deleted.yaml 历史)

graph status

状态概览 + 拓扑检查

graph validate

完整性校验(引用 + 拓扑 + 环)

graph rebuild

重建 index/ 派生索引

graph export --mermaid

导出 Mermaid 图

-o <file>

graph serve

启动 Web UI

-p <port>(默认 8934)

参数拿不准?每个命令都有 --helpgraph create-node --help


状态机(7 态 14 转换)

pending ──► ready ──► running ──► passed ──► blocked
                     │   │
                     │   └──► failed ──► pending   (重试,attempts 累加)
                     ▼
              cancelled(终止态)
        blocked ──► ready / failed / cancelled
  • 认领(claim)ready → running 记录 assigned_to + started_at(MCP 传 claim_by 参数)

  • 完成:转 passed / failed 自动记录 completed_at

  • 保护:非法转换(如 pending → running 直跳)返回明确错误,绝不静默


MCP:让 AI agent 直接干活 🤖

Super Plumber 自带 MCP Server(stdio 传输),coding agent 可以像用工具一样读图、建节点、认领任务、上报进度。

启动

graph-mcp

接入配置

Claude Codeclaude.json):

{
  "mcpServers": {
    "super-plumber": {
      "command": "graph-mcp"
    }
  }
}

opencode~/.config/opencode/opencode.json):

{
  "mcp": {
    "super-plumber": {
      "type": "local",
      "command": ["graph-mcp"],
      "enabled": true
    }
  }
}

也可以直接用绝对路径:"command": "node D:/path/to/dist/mcp/server.js",并设 cwd 为你的图所在目录。

9 个工具

工具

作用

必填参数

graph_get_node

读取节点完整内容

id

graph_create_node

创建节点

id, label

graph_update_node_status

状态流转;status=running 时传 claim_by 完成认领

id, status

graph_update_checkpoint

执行中上报检查点进度

node_id, checkpoint_id, status

graph_update_execution_report

填写交接单(供验收方抽查)

node_id, summary

graph_delete_node

软删除

id

graph_get_graph

获取完整拓扑(节点+边+邻接表)

graph_traverse

从某节点出发遍历相邻节点

node_id

graph_search

多条件搜索节点

query/status/type/assigned_to

可靠性设计:所有参数经 zod schema 校验——缺参、非法枚举返回 -32602 协议错误;不存在的节点/边返回 isError=true 和可读的错误消息;非法状态转换明确报错。工具永远不静默失败


Web UI(Svelte 5 + D3.js)

graph serve
# 打开 http://localhost:8934
  • 力导向图:缩放 / 平移 / 自动适配,按节点状态着色

  • 边类型可视化:7 种边类型不同颜色,悬停高亮

  • 光点流动running 节点的下游边有光点沿边流动("血管"隐喻)

  • checkpoint 进度条:节点下方展示子步骤完成进度

  • 执行者标签running 节点旁显示 assigned_to

  • WebSocket 增量推送:节点变更推送 node:updated 增量(非全量重推),断线自动 HTTP 回退


存储结构(Git 友好,人类可读)

.graph/                    # 运行时目录(graph init 生成,已在 .gitignore)
├── graph.yaml             # 图定义:入口/出口/根上下文 + 节点/边引用列表
├── nodes/*.yaml           # 节点文件:plan / checkpoints / expected_outcome / execution_report
├── edges/*.yaml           # 边文件:source / target / type / contract
├── snapshots/             # 版本快照(预留)
└── index/                 # 派生索引(graph.json / meta.json,可删可重建)

设计理念:

  • Git 是唯一真相源 —— 所有数据是文件,可 diff、可回滚、可 review

  • 文件即节点 —— 一个节点一个 YAML,人类可以直接用编辑器修改

  • 结构优先于文本 —— YAML schema 约束,拒绝自由 Markdown 的模糊性

  • 纯文件系统 —— 无数据库;软删除保留 .deleted.yaml 历史

仓库里附带了示例拓扑 .graph-example/(20 节点 36 边 + setup-topology.sh 重建脚本),可以参考它的节点/边写法。


Pi Agent 生态:subagent + skill

项目内置 2 个专用 subagent(.pi/agents/)与 1 个 skill(.pi/skills/plumber-flow/):

Agent

角色

职责

graph-designer

拓扑图设计师

把需求拆解为结构化拓扑,为每个节点制定 plan 和 definition_of_done

super-mario

拓扑主控

节点生命周期裁决(checkpoint 聚合 + 输出抽查)、重试管理、状态监测

plumber-flow skill 定义了 5 阶段执行协议(拆解 → 设计 → 建图 → 执行 → 验证),配套 6 个脚本(read/status/claim/checkpoint/report/traverse),保证 agent 按协议操作拓扑图、不越权、不假报进度。


开发与测试

# 从源码构建
git clone https://github.com/LUKAWI/super-plumber
cd super-plumber
npm install
npm run build && npm --prefix web-ui run build

# 测试(96 个用例:状态机/拓扑/CLI/MCP 协议)
npm test

# 本地链接全局(开发调试用)
npm link
graph --version

常见问题(FAQ)

问题

原因与解决

❌ 未找到 .../.graph/graph.yaml,请先运行 graph init

当前目录还没有图。先 graph init,或 cd 到图所在目录

❌ 端口 8934 已被占用

已有 serve 在跑。graph serve -p 8935 换端口

❌ Node x already exists / Edge x already exists

id 重复。工具拒绝覆盖,换一个新 id

❌ Invalid transition: ...

跳过了状态机允许的路径。按 Allowed: [...] 提示走合法转换

❌ MCP error -32602: ...

调用 MCP 工具缺参数或传了非法枚举,按提示补参数/改枚举

❌ 节点不存在: x

该节点不存在(可能是软删除或 id 写错),用 graph status / graph_search 确认

改完代码全局命令没变化

全局是发布包的快照。npm version patch && npm publish && npm i -g @lukawi/super-plumber

graph serve 后页面是空图

检查 cwd 是否是图所在目录;空图时 UI 会显示空状态引导


项目状态

Tests: 96/96 ✅ | CLI: 11 命令 | MCP: 9 工具 | 状态机: 7 态 14 转换 | 边类型: 7 种 | Web UI: Svelte 5 + D3.js

License

MIT © 2026 Super Plumber contributors

A
license - permissive license
-
quality - not tested
B
maintenance

Maintenance

Maintainers
Response time
Release cycle
Releases (12mo)
Commit activity

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Related MCP Servers

  • A
    license
    -
    quality
    A
    maintenance
    Server-enforced workflow discipline for AI agents. An MCP server providing persistent work items, dependency graphs, quality gates, and actor attribution. Schemas define what agents must produce — the server blocks the call if they don't. Works with any MCP-compatible client.
    199
    MIT
  • F
    license
    -
    quality
    D
    maintenance
    MCP server that lets AI agents execute structured business processes by exposing process steps as tools with a sequenced event bus to prevent skipping steps.
    1

View all related MCP servers

Related MCP Connectors

  • MCP server for AI agents to plan, verify, and deploy Cloudflare-native apps.

  • Control plane for autonomous software labor. Agents claim objectives over MCP with audit trail.

  • Coordinate multiple AI agents over MCP: atomic claims, leases, shared ledger, handoffs, tasks.

View all MCP Connectors

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/LUKAWI/super-plumber'

If you have feedback or need assistance with the MCP directory API, please join our Discord server