edict-multi-agent-orchestration

Install and use the Edict (三省六部) multi-agent orchestration system with 12 specialized AI agents, real-time kanban dashboard, and audit trails

INSTALLATION
npx skills add https://github.com/aradotso/trending-skills --skill edict-multi-agent-orchestration
Run in your project or agent environment. Adjust flags if your CLI version differs.

SKILL.md

$27

Prerequisites

  • Python 3.9+
  • Node.js 18+ (for React dashboard build)
  • macOS or Linux

Installation

Quick Demo (Docker — no OpenClaw needed)

# x86/amd64 (Ubuntu, WSL2)

docker run --platform linux/amd64 -p 7891:7891 cft0808/sansheng-demo

# Apple Silicon / ARM

docker run -p 7891:7891 cft0808/sansheng-demo

# Or with docker-compose (platform already set)

docker compose up

Open http://localhost:7891

Full Installation

git clone https://github.com/cft0808/edict.git

cd edict

chmod +x install.sh && ./install.sh

The install script automatically:

  • Creates all 12 agent workspaces (taizi, zhongshu, menxia, shangshu, hubu, libu, bingbu, xingbu, gongbu, libu2, zaochao, legacy-compat)
  • Writes SOUL.md role definitions to each agent workspace
  • Registers agents and permission matrix in openclaw.json
  • Symlinks shared data directories across all agent workspaces
  • Sets sessions.visibility all for inter-agent message routing
  • Syncs API keys across all agents
  • Builds React frontend
  • Initializes data directory and syncs official stats

First-time API Key Setup

# Configure API key on first agent

openclaw agents add taizi

# Then re-run install to propagate to all agents

./install.sh

Running the System

# Terminal 1: Data refresh loop (keeps kanban data current)

bash scripts/run_loop.sh

# Terminal 2: Dashboard server

python3 dashboard/server.py

# Open dashboard

open http://127.0.0.1:7891

Key Commands

OpenClaw Agent Management

# List all registered agents

openclaw agents list

# Add/configure an agent

openclaw agents add <agent-name>

# Check agent status

openclaw agents status

# Restart gateway (required after config changes)

openclaw gateway restart

# Send a message/edict to the system

openclaw send taizi "帮我分析一下竞争对手的产品策略"

Dashboard Server

# dashboard/server.py — serves on port 7891

# Built-in: React frontend + REST API + WebSocket updates

python3 dashboard/server.py

# Custom port

PORT=8080 python3 dashboard/server.py

Data Scripts

# Sync official (agent) statistics

python3 scripts/sync_officials.py

# Update kanban task states

python3 scripts/kanban_update.py

# Run news aggregation

python3 scripts/fetch_news.py

# Full refresh loop (runs all scripts in sequence)

bash scripts/run_loop.sh

Configuration

Agent Model Configuration ( openclaw.json )

{

  "agents": {

    "taizi": {

      "model": "claude-3-5-sonnet-20241022",

      "workspace": "~/.openclaw/workspaces/taizi"

    },

    "zhongshu": {

      "model": "gpt-4o",

      "workspace": "~/.openclaw/workspaces/zhongshu"

    },

    "menxia": {

      "model": "claude-3-5-sonnet-20241022",

      "workspace": "~/.openclaw/workspaces/menxia"

    },

    "shangshu": {

      "model": "gpt-4o-mini",

      "workspace": "~/.openclaw/workspaces/shangshu"

    }

  },

  "gateway": {

    "port": 7891,

    "sessions": {

      "visibility": "all"

    }

  }

}

Per-Agent Model Hot-Switching (via Dashboard)

Navigate to ⚙️ Models panel → select agent → choose LLM → Apply. Gateway restarts automatically (~5 seconds).

Environment Variables

# API keys (set before running install.sh or openclaw)

export ANTHROPIC_API_KEY="sk-ant-..."

export OPENAI_API_KEY="sk-..."

# Optional: Feishu/Lark webhook for notifications

export FEISHU_WEBHOOK_URL="https://open.feishu.cn/open-apis/bot/v2/hook/..."

# Optional: news aggregation

export NEWS_API_KEY="..."

# Dashboard port override

export DASHBOARD_PORT=7891

Agent Roles Reference

Agent

Role

Responsibility

taizi

太子 Crown Prince

Triage: chat → auto-reply, edicts → create task

zhongshu

中书省

Planning: decompose edict into subtasks

menxia

门下省

Review/Veto: quality gate, can reject and force rework

shangshu

尚书省

Dispatch: assign subtasks to ministries

hubu

户部 Ministry of Revenue

Finance, data analysis tasks

libu

礼部 Ministry of Rites

Communication, documentation tasks

bingbu

兵部 Ministry of War

Strategy, security tasks

xingbu

刑部 Ministry of Justice

Review, compliance tasks

gongbu

工部 Ministry of Works

Engineering, technical tasks

libu2

吏部 Ministry of Personnel

HR, agent management tasks

zaochao

早朝官

Morning briefing aggregator

Permission Matrix (who can message whom)

# Defined in openclaw.json — enforced by gateway

PERMISSIONS = {

    "taizi":    ["zhongshu"],

    "zhongshu": ["menxia"],

    "menxia":   ["zhongshu", "shangshu"],  # can veto back to zhongshu

    "shangshu": ["hubu", "libu", "bingbu", "xingbu", "gongbu", "libu2"],

    # ministries report back up the chain

    "hubu":     ["shangshu"],

    "libu":     ["shangshu"],

    "bingbu":   ["shangshu"],

    "xingbu":   ["shangshu"],

    "gongbu":   ["shangshu"],

    "libu2":    ["shangshu"],

}

Task State Machine

# scripts/kanban_update.py enforces valid transitions

VALID_TRANSITIONS = {

    "pending":     ["planning"],

    "planning":    ["reviewing", "pending"],      # zhongshu → menxia

    "reviewing":   ["dispatching", "planning"],   # menxia approve or veto

    "dispatching": ["executing"],

    "executing":   ["completed", "failed"],

    "completed":   [],

    "failed":      ["pending"],  # retry

}

# Invalid transitions are rejected — no silent state corruption

Real Code Examples

Send an Edict Programmatically

import subprocess

import json

def send_edict(message: str, agent: str = "taizi") -> dict:

    """Send an edict to the Crown Prince for triage."""

    result = subprocess.run(

        ["openclaw", "send", agent, message],

        capture_output=True,

        text=True

    )

    return {"stdout": result.stdout, "returncode": result.returncode}

# Example edicts

send_edict("分析本季度用户增长数据,找出关键驱动因素")

send_edict("起草一份关于产品路线图的对外公告")

send_edict("审查现有代码库的安全漏洞")

Read Kanban State

import json

from pathlib import Path

def get_kanban_tasks(data_dir: str = "data") -> list[dict]:

    """Read current kanban task state."""

    tasks_file = Path(data_dir) / "tasks.json"

    if not tasks_file.exists():

        return []

    with open(tasks_file) as f:

        return json.load(f)

def get_tasks_by_status(status: str) -> list[dict]:

    tasks = get_kanban_tasks()

    return [t for t in tasks if t.get("status") == status]

# Usage

executing = get_tasks_by_status("executing")

completed = get_tasks_by_status("completed")

print(f"In progress: {len(executing)}, Done: {len(completed)}")

Update Task Status (with validation)

import json

from pathlib import Path

from datetime import datetime, timezone

VALID_TRANSITIONS = {

    "pending":     ["planning"],

    "planning":    ["reviewing", "pending"],

    "reviewing":   ["dispatching", "planning"],

    "dispatching": ["executing"],

    "executing":   ["completed", "failed"],

    "completed":   [],

    "failed":      ["pending"],

}

def update_task_status(task_id: str, new_status: str, data_dir: str = "data") -> bool:

    """Update task status with state machine validation."""

    tasks_file = Path(data_dir) / "tasks.json"

    tasks = json.loads(tasks_file.read_text())

    task = next((t for t in tasks if t["id"] == task_id), None)

    if not task:

        raise ValueError(f"Task {task_id} not found")

    current = task["status"]

    allowed = VALID_TRANSITIONS.get(current, [])

    if new_status not in allowed:

        raise ValueError(

            f"Invalid transition: {current} → {new_status}. "

            f"Allowed: {allowed}"

        )

    task["status"] = new_status

    task["updated_at"] = datetime.now(timezone.utc).isoformat()

    task.setdefault("history", []).append({

        "from": current,

        "to": new_status,

        "timestamp": task["updated_at"]

    })

    tasks_file.write_text(json.dumps(tasks, ensure_ascii=False, indent=2))

    return True

Dashboard REST API Client

import urllib.request

import json

BASE_URL = "http://127.0.0.1:7891/api"

def api_get(endpoint: str) -> dict:

    with urllib.request.urlopen(f"{BASE_URL}{endpoint}") as resp:

        return json.loads(resp.read())

def api_post(endpoint: str, data: dict) -> dict:

    payload = json.dumps(data).encode()

    req = urllib.request.Request(

        f"{BASE_URL}{endpoint}",

        data=payload,

        headers={"Content-Type": "application/json"},

        method="POST"

    )

    with urllib.request.urlopen(req) as resp:

        return json.loads(resp.read())

# Read dashboard data

tasks    = api_get("/tasks")

agents   = api_get("/agents")

sessions = api_get("/sessions")

news     = api_get("/news")

# Trigger task action

api_post("/tasks/pause",  {"task_id": "task-123"})

api_post("/tasks/cancel", {"task_id": "task-123"})

api_post("/tasks/resume", {"task_id": "task-123"})

# Switch model for an agent

api_post("/agents/model", {

    "agent": "zhongshu",

    "model": "gpt-4o-2024-11-20"

})

Agent Health Check

import json

from pathlib import Path

from datetime import datetime, timezone, timedelta

def check_agent_health(data_dir: str = "data") -> dict[str, str]:

    """

    Returns health status for each agent.

    🟢 active   = heartbeat within 2 min

    🟡 stale    = heartbeat 2-10 min ago

    🔴 offline  = heartbeat >10 min ago or missing

    """

    heartbeats_file = Path(data_dir) / "heartbeats.json"

    if not heartbeats_file.exists():

        return {}

    heartbeats = json.loads(heartbeats_file.read_text())

    now = datetime.now(timezone.utc)

    status = {}

    for agent, last_beat in heartbeats.items():

        last = datetime.fromisoformat(last_beat)

        delta = now - last

        if delta < timedelta(minutes=2):

            status[agent] = "🟢 active"

        elif delta < timedelta(minutes=10):

            status[agent] = "🟡 stale"

        else:

            status[agent] = "🔴 offline"

    return status

# Usage

health = check_agent_health()

for agent, s in health.items():

    print(f"{agent:12} {s}")

Custom SOUL.md (Agent Personality)

<!-- ~/.openclaw/workspaces/gongbu/SOUL.md -->

# 工部尚书 · Minister of Works

## Role

You are the Minister of Works (工部). You handle all technical,

engineering, and infrastructure tasks assigned by Shangshu Province.

## Rules

1. Always break technical tasks into concrete, verifiable steps

2. Return structured results: { "status": "...", "output": "...", "artifacts": [] }

3. Flag blockers immediately — do not silently fail

4. Estimate complexity: S/M/L/XL before starting

## Output Format

Always respond with valid JSON. Include a `summary` field ≤ 50 chars

for kanban display.

Dashboard Panels

Panel

URL Fragment

Key Features

Kanban

#kanban

Task columns, heartbeat badges, filter/search, pause/cancel/resume

Monitor

#monitor

Agent health cards, task distribution charts

Memorials

#memorials

Completed task archive, 5-stage timeline, Markdown export

Templates

#templates

9 preset edict templates with parameter forms

Officials

#officials

Token usage ranking, activity stats

News

#news

Daily tech/finance briefing, Feishu push

Models

#models

Per-agent LLM switcher (hot reload ~5s)

Skills

#skills

View/add agent skills

Sessions

#sessions

Live OC-* session monitor

Court

#court

Multi-agent discussion around a topic

Common Patterns

Pattern 1: Parallel Ministry Execution

# Shangshu dispatches to multiple ministries simultaneously

# Each ministry works independently; shangshu aggregates results

edict = "竞品分析:研究TOP3竞争对手的产品、定价、市场策略"

# Zhongshu splits into subtasks:

# hubu  → pricing analysis

# libu  → market communication analysis

# bingbu → competitive strategy analysis

# gongbu → technical feature comparison

# All execute in parallel; shangshu waits for all 4, then aggregates

Pattern 2: Menxia Veto Loop

# If menxia rejects zhongshu's plan:

# menxia → zhongshu: "子任务拆解不完整,缺少风险评估维度,请补充"

# zhongshu revises and resubmits to menxia

# Loop continues until menxia approves

# Max iterations configurable in openclaw.json: "max_review_cycles": 3

Pattern 3: News Aggregation + Push

# scripts/fetch_news.py → data/news.json → dashboard #news panel

# Optional Feishu push:

import os, json, urllib.request

def push_to_feishu(summary: str):

    webhook = os.environ["FEISHU_WEBHOOK_URL"]

    payload = json.dumps({

        "msg_type": "text",

        "content": {"text": f"📰 天下要闻\n{summary}"}

    }).encode()

    req = urllib.request.Request(

        webhook, data=payload,

        headers={"Content-Type": "application/json"}

    )

    urllib.request.urlopen(req)

Troubleshooting

exec format error in Docker

# Force platform on x86/amd64

docker run --platform linux/amd64 -p 7891:7891 cft0808/sansheng-demo

Agents not receiving messages

# Ensure sessions visibility is set to "all"

openclaw config set sessions.visibility all

openclaw gateway restart

# Or re-run install.sh — it sets this automatically

./install.sh

API key not propagated to all agents

# Re-run install after configuring key on first agent

openclaw agents add taizi  # configure key here

./install.sh               # propagates to all agents

Dashboard shows stale data

# Ensure run_loop.sh is running

bash scripts/run_loop.sh

# Or trigger manual refresh

python3 scripts/sync_officials.py

python3 scripts/kanban_update.py

React frontend not built

# Requires Node.js 18+

cd dashboard/frontend

npm install &#x26;&#x26; npm run build

# server.py will then serve the built assets

Invalid state transition error

# kanban_update.py enforces the state machine

# Check current status before updating:

tasks = get_kanban_tasks()

task = next(t for t in tasks if t["id"] == "your-task-id")

print(f"Current: {task['status']}")

print(f"Allowed next: {VALID_TRANSITIONS[task['status']]}")

Gateway restart after model change

# After editing openclaw.json models section

openclaw gateway restart

# Wait ~5 seconds for agents to reconnect

Project Structure

edict/

├── install.sh              # One-command setup

├── openclaw.json           # Agent registry + permissions + model config

├── scripts/

│   ├── run_loop.sh         # Continuous data refresh daemon

│   ├── kanban_update.py    # State machine enforcement

│   ├── sync_officials.py   # Agent stats aggregation

│   └── fetch_news.py       # News aggregation

├── dashboard/

│   ├── server.py           # stdlib-only HTTP + WebSocket server (port 7891)

│   ├── dashboard.html      # Fallback single-file dashboard

│   └── frontend/           # React 18 source (builds to server.py assets)

├── data/                   # Shared data (symlinked into all workspaces)

│   ├── tasks.json

│   ├── heartbeats.json

│   ├── news.json

│   └── officials.json

├── workspaces/             # Per-agent workspace roots

│   ├── taizi/SOUL.md

│   ├── zhongshu/SOUL.md

│   └── ...

└── docs/

    ├── task-dispatch-architecture.md

    └── getting-started.md
BrowserAct

Let your agent run on any real-world website

Bypass CAPTCHA & anti-bot for free. Start local, scale to cloud.

Explore BrowserAct Skills →

Stop writing automation&scrapers

Install the CLI. Run your first Skill in 30 seconds. Scale when you're ready.

Start free
free · no credit card