项目简介
ClawTeam 是一个 框架无关的多智能体协调 CLI 工具，它让 AI 智能体能够自主组织成团队——分配任务、相互通信、协调工作并合并结果，而无需人工微观管理。
这是 HKUDS/ClawTeam 的一个分支版本，深度集成了 OpenClaw 作为默认代理后端。核心创新点包括：
- 默认 OpenClaw 代理：开箱即用的智能体支持
- 每代理会话隔离：每个代理拥有独立的 git worktree 和 tmux 窗口
- 执行批准自动配置：自动处理 OpenClaw 的权限审批
- 生产级 spawn 后端：经过实战验证的代理启动机制
整体架构
ClawTeam 采用去中心化的文件系统架构，所有状态存储在 ~/.clawteam/ 目录中，无需数据库或中央服务器。
┌─────────────────────────────────────────────────────────────────┐
│ 用户 / Leader 代理 │
│ (通过 clawteam CLI 驱动) │
└────────────────────────────┬────────────────────────────────────┘
│
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Spawn 系统 │ │ Team 管理 │ │ Workspace │
│ (启动代理) │ │ (团队配置) │ │ (git worktree)│
└────────┬────────┘ └────────┬────────┘ └────────┬────────┘
│ │ │
└───────────────────┼───────────────────┘
│
▼
┌──────────────────────────────┐
│ Transport 抽象层 │
│ (FileTransport / P2P) │
└──────────────┬───────────────┘
│
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Task Store │ │ Mailbox │ │ Board │
│ (任务依赖) │ │ (消息队列) │ │ (监控看板) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
▼
┌──────────────────────────────┐
│ ~/.clawteam/ 文件系统 │
│ ├── teams/ (团队配置) │
│ ├── tasks/ (任务状态) │
│ ├── inboxes/ (消息队列) │
│ ├── workspaces/(git worktree)│
│ └── events/ (事件日志) │
└──────────────────────────────┘
---
核心模块详解
1. CLI 命令系统 (clawteam/cli/commands.py)
文件路径: clawteam/cli/commands.py
这是整个项目的入口点，使用 Typer 框架构建命令行界面。核心命令组包括：
# 命令组织结构
clawteam
├── config # 配置管理 (show/set/get/health)
├── team # 团队生命周期 (spawn-team/discover/cleanup/status)
├── inbox # 消息系统 (send/broadcast/receive/peek/log/watch)
├── task # 任务管理 (create/update/list/stats)
├── cost # 成本追踪 (report/show/budget)
├── spawn # 代理启动 (tmux/subprocess 后端)
├── workspace # 工作区管理 (list/checkpoint/merge/cleanup)
├── board # 监控看板 (show/live/attach/serve)
├── session # 会话管理 (save/restore)
├── lifecycle # 生命周期 (request-shutdown/idle/on-exit)
├── plan # 计划审批 (submit/approve/reject)
└── launch # 模板启动 (hedge-fund/code-review 等)
关键设计特点：
- 全局选项支持：
--json输出、--data-dir覆盖、--transport选择 - 统一的输出处理：
_output()函数自动处理 JSON/人类可读格式 - 命令分组：使用
typer.Typer()创建子命令组，通过app.add_typer()注册
2. 配置系统 (clawteam/config.py)
文件路径: clawteam/config.py
采用 三层优先级配置：环境变量 > 配置文件 > 默认值
class ClawTeamConfig(BaseModel):
data_dir: str = "" # 数据目录 (默认 ~/.clawteam)
user: str = "" # 用户标识 (多用户协作)
default_team: str = "" # 默认团队名
transport: str = "" # 传输后端 (file/p2p)
workspace: str = "auto" # 工作区模式 (auto/always/never)
default_backend: str = "tmux" # spawn 后端 (tmux/subprocess)
skip_permissions: bool = True # 跳过权限检查 (Claude Code)
配置优先级实现：
def get_effective(key: str) -> tuple[str, str]:
"""返回 (值，来源)"""
# 1. 检查环境变量
env_val = os.environ.get(env_map[key])
if env_val:
return env_val, "env"
# 2. 检查配置文件
file_val = getattr(cfg, key, "")
if file_val != default_val:
return str(file_val), "file"
# 3. 返回默认值
return str(default_val), "default"
3. 代理身份系统 (clawteam/identity.py)
文件路径: clawteam/identity.py
支持多前缀环境变量，兼容 CLAWTEAM_*、OPENCLAW_*、CLAUDE_CODE_*：
@dataclass
class AgentIdentity:
agent_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
agent_name: str = "agent"
user: str = ""
agent_type: str = "general-purpose"
team_name: str | None = None
is_leader: bool = False
plan_mode_required: bool = False
@classmethod
def from_env(cls) -> AgentIdentity:
"""从环境变量构建身份"""
return cls(
agent_id=_env("CLAWTEAM_AGENT_ID", "CLAUDE_CODE_AGENT_ID", ...),
agent_name=_env("CLAWTEAM_AGENT_NAME", "CLAUDE_CODE_AGENT_NAME", ...),
team_name=_env("CLAWTEAM_TEAM_NAME", "CLAUDE_CODE_TEAM_NAME") or None,
is_leader=_env_bool("CLAWTEAM_AGENT_LEADER", "CLAUDE_CODE_AGENT_LEADER"),
)
4. 团队管理系统 (clawteam/team/)
#### 4.1 团队管理器 (manager.py)
文件路径: clawteam/team/manager.py
负责团队的生命周期操作：
class TeamManager:
@staticmethod
def create_team(name, leader_name, leader_id, description, user):
"""创建团队并注册 leader"""
@staticmethod
def discover_teams() -> list[dict]:
"""发现所有团队"""
@staticmethod
def add_member(team_name, member_name, agent_id, agent_type, user):
"""添加成员（创建 inbox）"""
@staticmethod
def cleanup(team_name) -> bool:
"""删除团队及所有数据"""
团队配置结构 (~/.clawteam/teams/{team}/config.json)：
{
"name": "my-team",
"description": "Build a web app",
"leadAgentId": "abc123",
"createdAt": "2026-03-20T10:00:00Z",
"members": [
{"name": "leader", "user": "alice", "agentId": "abc123", "agentType": "leader"},
{"name": "worker1", "user": "alice", "agentId": "def456", "agentType": "general-purpose"}
],
"budgetCents": 1000
}
#### 4.2 任务系统 (tasks.py)
文件路径: clawteam/team/tasks.py
实现 基于文件锁的任务存储，支持任务依赖链：
class TaskStore:
def create(self, subject, description, owner, blocks, blocked_by, metadata):
"""创建任务，支持依赖关系"""
def update(self, task_id, status, owner, caller, force):
"""更新任务状态，自动解锁依赖"""
def _resolve_dependents_unlocked(self, completed_task_id):
"""当任务完成时，自动解除依赖该任务的其他任务"""
任务状态机：
pending ──► in_progress ──► completed
▲ │
│ ▼
blocked ◄───────┘
文件锁机制：
@contextmanager
def _write_lock(self):
lock_path = _tasks_lock_path(self.team_name)
with lock_path.open("a+", encoding="utf-8") as lock_file:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) # 独占锁
try:
yield
finally:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) # 释放锁
#### 4.3 邮箱系统 (mailbox.py)
文件路径: clawteam/team/mailbox.py
实现可插拔的传输层抽象：
class MailboxManager:
def __init__(self, team_name, transport=None):
self._transport = transport or _default_transport(team_name)
def send(self, from_agent, to, content, msg_type, ...):
"""发送点对点消息"""
def broadcast(self, from_agent, content, exclude):
"""广播消息给所有成员"""
def receive(self, agent_name, limit) -> list[TeamMessage]:
"""接收并消费消息（FIFO）"""
def peek(self, agent_name) -> list[TeamMessage]:
"""查看消息但不消费"""
传输层接口 (transport/base.py)：
class Transport(ABC):
@abstractmethod
def deliver(self, recipient: str, data: bytes) -> None:
"""投递消息到收件箱"""
@abstractmethod
def fetch(self, agent_name: str, limit: int, consume: bool) -> list[bytes]:
"""获取消息"""
@abstractmethod
def count(self, agent_name: str) -> int:
"""返回待处理消息数"""
5. Spawn 系统 (clawteam/spawn/)
#### 5.1 Tmux 后端 (tmux_backend.py)
文件路径: clawteam/spawn/tmux_backend.py
这是核心功能——在 tmux 中启动代理，实现可视化监控：
class TmuxBackend(SpawnBackend):
def spawn(self, command, agent_name, agent_id, agent_type, team_name, prompt, env, cwd):
# 1. 构建环境变量
env_vars = {
"CLAWTEAM_AGENT_ID": agent_id,
"CLAWTEAM_AGENT_NAME": agent_name,
"CLAWTEAM_TEAM_NAME": team_name,
"CLAWTEAM_AGENT_LEADER": "0",
}
# 2. 构建命令（添加 --dangerously-skip-permissions 等）
final_command = self._build_command(command, skip_permissions, prompt)
# 3. 创建 tmux session/window
session_name = f"clawteam-{team_name}"
target = f"{session_name}:{agent_name}"
# 4. 发送 prompt 到交互式会话
self._send_prompt(target, prompt)
# 5. 注册 spawn 信息（用于存活检查）
register_agent(team_name, agent_name, "tmux", tmux_target=target, pid=pane_pid)
关键特性：
- 自动处理 workspace trust 提示：检测 Claude Code/Codex 的目录信任提示并自动确认
- Prompt 注入：使用
load-buffer+paste-buffer避免多行 prompt 的转义问题 - 退出钩子：代理退出时自动触发
clawteam lifecycle on-exit - 存活检查：通过 tmux pane 状态和 PID 双重检查代理是否存活
registry.py)文件路径: clawteam/spawn/registry.py
持久化代理进程信息，用于存活检查：
def register_agent(team_name, agent_name, backend, tmux_target, pid, command):
"""原子写入 spawn 信息"""
def is_agent_alive(team_name, agent_name) -> bool | None:
"""检查代理是否存活"""
# tmux 后端：检查 pane_dead 标志
# subprocess 后端：检查 PID
def list_dead_agents(team_name) -> list[str]:
"""返回已死亡的代理名称"""
#### 5.3 Prompt 构建器 (prompt.py)
文件路径: clawteam/spawn/prompt.py
为 spawned 代理构建初始 prompt：
def build_agent_prompt(agent_name, agent_id, agent_type, team_name, leader_name, task, workspace_dir, workspace_branch):
"""构建包含身份、任务、协调协议的 prompt"""
return f"""
## Identity
- Name: {agent_name}
- ID: {agent_id}
- Team: {team_name}
- Leader: {leader_name}
## Workspace
- Working directory: {workspace_dir}
- Branch: {workspace_branch}
## Task
{task}
## Coordination Protocol
- Use `clawteam task list {team_name} --owner {agent_name}` to see your tasks
- Starting: `clawteam task update {team_name} <task-id> --status in_progress`
- Finishing: `clawteam task update {team_name} <task-id> --status completed`
- Report: `clawteam inbox send {team_name} {leader_name} "All tasks completed..."`
"""
6. Workspace 系统 (clawteam/workspace/)
#### 6.1 工作区管理器 (manager.py)
文件路径: clawteam/workspace/manager.py
基于 git worktree 实现代理工作隔离：
class WorkspaceManager:
def create_workspace(self, team_name, agent_name, agent_id):
"""创建 git worktree 和分支"""
branch = f"clawteam/{team_name}/{agent_name}"
wt_path = _workspaces_root() / team_name / agent_name
git.create_worktree(self.repo_root, wt_path, branch, base_ref=self.base_branch)
def checkpoint(self, team_name, agent_name, message):
"""提交当前工作作为检查点"""
def merge_workspace(self, team_name, agent_name, target_branch, cleanup_after):
"""合并工作区到目标分支"""
def cleanup_workspace(self, team_name, agent_name, auto_checkpoint):
"""清理工作区（可选自动 checkpoint）"""
工作区注册表 (~/.clawteam/workspaces/{team}/workspace-registry.json)：
{
"team_name": "my-team",
"repo_root": "/path/to/repo",
"workspaces": [
{
"agent_name": "worker1",
"agent_id": "abc123",
"branch_name": "clawteam/my-team/worker1",
"worktree_path": "~/.clawteam/workspaces/my-team/worker1",
"created_at": "2026-03-20T10:00:00Z"
}
]
}
#### 6.2 Git 操作封装 (git.py)
封装 git 命令，提供工作区操作：
def repo_root(cwd: Path) -> Path:
"""找到 git 仓库根目录"""
def create_worktree(repo_root, wt_path, branch, base_ref):
"""创建 worktree 和分支"""
def remove_worktree(repo_root, wt_path):
"""删除 worktree"""
def merge_branch(repo_root, source_branch, target_branch):
"""合并分支"""
7. Board 监控系统 (clawteam/board/)
#### 7.1 Web 服务器 (server.py)
文件路径: clawteam/board/server.py
使用 Python 标准库实现轻量级 HTTP 服务器：
def serve(host, port, default_team, interval):
"""启动 Web UI 服务器"""
server = ThreadingHTTPServer((host, port), BoardHandler)
server.serve_forever()
class BoardHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/":
self._serve_static("index.html")
elif self.path == "/api/overview":
self._serve_json(self.collector.collect_overview())
elif self.path.startswith("/api/team/"):
self._serve_team(team_name)
elif self.path.startswith("/api/events/"):
self._serve_sse(team_name) # Server-Sent Events
SSE 实时推送：
def _serve_sse(self, team_name):
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
while True:
data = self.collector.collect_team(team_name)
self.wfile.write(f"data: {json.dumps(data)}\n\n".encode())
time.sleep(self.interval)
#### 7.2 数据收集器 (collector.py)
收集团队状态数据供 Web UI 使用：
class BoardCollector:
def collect_overview(self) -> dict:
"""收集所有团队概览"""
def collect_team(self, team_name) -> dict:
"""收集单个团队的详细状态"""
return {
"config": team_config,
"tasks": task_list,
"members": member_status,
"costs": cost_summary,
}
8. 模板系统 (clawteam/templates/)
文件路径: clawteam/templates/
使用 TOML 文件定义团队模板，一键启动复杂的多代理团队：
示例：对冲基金模板 (hedge-fund.toml)：
[template]
name = "hedge-fund"
description = "AI Hedge Fund - multi-agent investment analysis team"
command = ["openclaw"]
backend = "tmux"
[template.leader]
name = "portfolio-manager"
type = "portfolio-manager"
task = """你是投资组合经理..."""
[[template.agents]]
name = "buffett-analyst"
type = "value-analyst"
task = """你是沃伦·巴菲特，使用价值投资原则分析股票..."""
[[template.agents]]
name = "growth-analyst"
type = "growth-analyst"
task = """你是成长型投资者..."""
[[template.tasks]]
subject = "协调分析并做出最终投资决策"
owner = "portfolio-manager"
[[template.tasks]]
subject = "价值投资分析（巴菲特风格）"
owner = "buffett-analyst"
内置模板：
hedge-fund.toml- AI 对冲基金（7 个代理）code-review.toml- 代码审查团队research-paper.toml- 研究论文团队strategy-room.toml- 战略会议室
设计思想与模式
1. 文件系统即数据库
核心理念：所有状态存储在 ~/.clawteam/ 目录中，无需数据库或中央服务器。
~/.clawteam/
├── config.json # 全局配置
├── teams/
│ └── {team}/
│ ├── config.json # 团队配置
│ ├── inboxes/ # 消息队列
│ │ ├── {agent}/
│ │ │ └── msg-*.json
│ │ └── _pending_*/ # 临时收件箱（加入请求）
│ └── events/ # 事件日志
├── tasks/
│ └── {team}/
│ ├── task-*.json # 任务文件
│ └── .tasks.lock # 文件锁
├── workspaces/
│ └── {team}/
│ ├── {agent}/ # git worktree
│ └── workspace-registry.json
├── costs/
│ └── {team}/
│ └── cost-*.json
├── sessions/
│ └── {team}/
│ └── session-*.json
└── plans/
└── {team}/
└── plan-*.json
优势：
- 零依赖：不需要数据库、Redis 等外部服务
- 可移植：整个状态就是一个目录，可轻松备份/迁移
- 可观察：直接用文本编辑器即可查看状态
- 原子性：使用
write tmp + rename模式确保原子写入
2. 传输层抽象
设计模式：策略模式（Strategy Pattern）
# 抽象接口
class Transport(ABC):
@abstractmethod
def deliver(self, recipient, data): ...
@abstractmethod
def fetch(self, agent_name, limit, consume): ...
# 具体实现
class FileTransport(Transport):
"""基于文件系统的传输（默认）"""
class P2PTransport(Transport):
"""基于 ZeroMQ 的 P2P 传输"""
优势：
- 可插拔：通过
CLAWTEAM_TRANSPORT=file|p2p切换后端 - 可扩展：轻松添加 Redis、NATS 等新传输
- 向后兼容：现有代码无需修改
3. 多用户协作设计
命名空间隔离：使用 (user, agent_name) 复合键区分不同用户的代理。
# 收件箱命名
inbox_name = f"{user}_{agent_name}" if user else agent_name
# 成员唯一性检查
for m in config.members:
if m.name == member_name and m.user == user:
raise ValueError("Agent already in team")
用户标识来源：
1. CLAWTEAM_USER 环境变量
2. clawteam config set user 配置
3. 默认为空（单用户模式）
4. 任务依赖链
设计模式：观察者模式（Observer Pattern）的变体
# 任务创建时指定依赖
task = store.create(
subject="Build frontend",
blocked_by=["task-123", "task-456"] # 被这些任务阻塞
)
# 任务完成时自动解锁依赖
def _resolve_dependents_unlocked(self, completed_task_id):
for task in all_tasks:
if completed_task_id in task.blocked_by:
task.blocked_by.remove(completed_task_id)
if not task.blocked_by and task.status == TaskStatus.blocked:
task.status = TaskStatus.pending # 自动解锁
5. 文件锁机制
设计模式：互斥锁（Mutex）
@contextmanager
def _write_lock(self):
lock_path = _tasks_lock_path(self.team_name)
with lock_path.open("a+", encoding="utf-8") as lock_file:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) # 独占锁
try:
yield
finally:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
任务级锁：
def _acquire_lock(self, task, caller, force):
if task.locked_by and task.locked_by != caller and not force:
# 检查锁持有者是否存活
alive = is_agent_alive(self.team_name, task.locked_by)
if alive is not False:
raise TaskLockError("Task is locked by another agent")
task.locked_by = caller
6. 模板驱动的团队创建
设计模式：模板方法模式（Template Method Pattern）
# TOML 模板定义团队结构
[template]
name = "hedge-fund"
command = ["openclaw"]
[template.leader]
name = "portfolio-manager"
task = "..."
[[template.agents]]
name = "analyst-1"
task = "..."
[[template.tasks]]
subject = "..."
owner = "analyst-1"
变量替换：
{goal}- 团队目标{team_name}- 团队名称{agent_name}- 代理名称
技术亮点
1. 原子写入模式
所有 JSON 文件写入都使用 write-tmp-then-rename 模式：
def _save_config(config: TeamConfig) -> None:
path = _config_path(config.name)
tmp = path.with_suffix(".tmp")
tmp.write_text(config.model_dump_json(indent=2), encoding="utf-8")
tmp.rename(path) # 原子操作
优势：即使进程崩溃，也不会留下损坏的文件。
2. 进程存活检查
双重检查机制确保准确判断代理状态：
def is_agent_alive(team_name, agent_name):
info = registry.get(agent_name)
if backend == "tmux":
# 1. 检查 tmux pane_dead 标志
alive = _tmux_pane_alive(info.get("tmux_target"))
if alive is False:
# 2. 回退到 PID 检查
return _pid_alive(info.get("pid"))
return alive
elif backend == "subprocess":
return _pid_alive(info.get("pid"))
3. 智能 Prompt 注入
针对多行 prompt 的转义问题，使用 tmux buffer 机制：
# 写入临时文件
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt") as f:
f.write(prompt)
tmp_path = f.name
# 使用 load-buffer + paste-buffer
subprocess.run(["tmux", "load-buffer", "-b", f"prompt-{agent_name}", tmp_path])
subprocess.run(["tmux", "paste-buffer", "-b", f"prompt-{agent_name}", "-t", target])
4. 自动 Workspace Trust 确认
检测并自动确认 Claude Code/Codex 的目录信任提示：
def _confirm_workspace_trust_if_prompted(target, command):
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
pane_text = _capture_tmux_pane(target)
if _looks_like_workspace_trust_prompt(command, pane_text):
subprocess.run(["tmux", "send-keys", "-t", target, "Enter"])
return True
time.sleep(poll_interval)
5. SSE 实时推送
Web UI 使用 Server-Sent Events 实现实时更新：
def _serve_sse(self, team_name):
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
while True:
data = self.collector.collect_team(team_name)
self.wfile.write(f"data: {json.dumps(data)}\n\n".encode())
time.sleep(self.interval) # 可配置的更新间隔
---
使用流程
典型工作流
# 1. 创建团队
clawteam team spawn-team my-team -d "Build a web app" -n leader
# 2. 启动 Worker 代理
clawteam spawn --team my-team --agent-name alice --task "Implement auth"
clawteam spawn --team my-team --agent-name bob --task "Write tests"
# 3. 监控进度
clawteam board show my-team # 终端看板
clawteam board live my-team # 自动刷新
clawteam board attach my-team # tmux 平铺视图
clawteam board serve --port 8080 # Web UI
# 4. 任务管理
clawteam task list my-team --owner alice
clawteam task update my-team <id> --status completed
# 5. 消息通信
clawteam inbox send my-team alice "Need help with..."
clawteam inbox broadcast my-team "Meeting in 5 minutes"
# 6. 工作区管理
clawteam workspace list my-team
clawteam workspace checkpoint my-team alice
clawteam workspace merge my-team alice
# 7. 清理
clawteam team cleanup my-team
模板启动
# 一键启动对冲基金团队
clawteam launch hedge-fund --team fund1 --goal "Analyze AAPL, MSFT, NVDA"
# 代码审查团队
clawteam launch code-review --team review1 --goal "Review PR #123"
---
总结
ClawTeam 是一个设计精巧的多智能体协调框架，其核心优势在于：
架构优势
1. 零依赖架构：基于文件系统的状态存储，无需数据库或中央服务器 2. 可插拔传输层：支持 File、P2P（ZeroMQ）、Redis 等多种传输后端 3. 多用户协作：通过命名空间隔离支持多用户共享团队 4. git worktree 隔离：每个代理拥有独立的工作区，避免冲突技术亮点
1. 原子写入：write-tmp-then-rename 模式确保数据一致性 2. 文件锁机制：fcntl 锁确保并发安全 3. 进程存活检查：双重检查机制准确判断代理状态 4. 智能 Prompt 注入：使用 tmux buffer 处理多行 prompt设计模式
- 策略模式：传输层抽象
- 观察者模式：任务依赖自动解锁
- 模板方法模式：团队模板系统
- 互斥锁模式：文件和任务级锁
适用场景
- 自主 ML 研究：多 GPU 并行实验
- 敏捷软件开发：多代理并行开发
- AI 对冲基金：多分析师协作
- 代码审查：多视角代码分析