Edict (三省六部) Multi-Agent Orchestration

Skill by ara.so — Daily 2026 Skills collection.

Edict implements a 1400-year-old Tang Dynasty governance model as an AI multi-agent architecture. Twelve specialized agents form a checks-and-balances pipeline: Crown Prince (triage) → Zhongshu (planning) → Menxia (review/veto) → Shangshu (dispatch) → Six Ministries (parallel execution). Built on OpenClaw, it provides a real-time React kanban dashboard, full audit trails, and per-agent LLM configuration.

Architecture Overview

You (Emperor) → taizi (triage) → zhongshu (plan) → menxia (review/veto)
             → shangshu (dispatch) → [hubu|libu|bingbu|xingbu|gongbu|libu2] (execute)
             → memorial (result archived)

Key differentiator vs CrewAI/AutoGen: Menxia (门下省) is a mandatory quality gate — it can veto and force rework before tasks reach executors.

Prerequisites

• OpenClaw installed and running
• 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

taizi

zhongshu

menxia

shangshu

hubu

libu

bingbu

xingbu

gongbu

libu2

zaochao

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

#kanban

#monitor

#memorials

#templates

#officials

#news

#models

#skills

#sessions

#court

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 && 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