9. Agent Teams
"When the task is too big for one, delegate to teammates"
New to this?
What makes a teammate different from a subagent?
A subagent is disposable β spawn, work, return summary, die. A teammate has a persistent identity, a lifecycle (IDLE, WORKING, SHUTDOWN), and a mailbox it checks between tasks. Teammates remember their role across invocations.
What is an agent mailbox?
An append-only JSONL file on disk (e.g., .team/inbox/alice.jsonl). Any agent can write a message to Alice's inbox. When Alice's turn comes, she drains the file β reads all messages, then truncates it. Like email, but for agents.
What is a team roster?
A JSON file (.team/config.json) listing all teammates, their names, roles, and current statuses. The lead agent reads this to know who's available and what each teammate specializes in.
The Problem
Subagents are disposable: spawn, work, return summary, die. No identity, no memory between invocations. Background tasks run shell commands but canβt make LLM-guided decisions.
Real teamwork needs: (1) persistent agents that outlive a single prompt, (2) identity and lifecycle management, (3) a communication channel between agents.
The Solution
Teammate lifecycle:
spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN
Communication:
.team/
config.json <- team roster + statuses
inbox/
alice.jsonl <- append-only, drain-on-read
bob.jsonl
lead.jsonl
+--------+ send("alice","bob","...") +--------+
| lead | -----------------------------> | bob |
+--------+ +--------+
^ |
| inbox/lead.jsonl |
+----------------------------------------+
How It Works
- The team config stores roster and statuses.
import json, threading
from pathlib import Path
TEAM_DIR = Path(".team")
INBOX_DIR = TEAM_DIR / "inbox"
TEAM_DIR.mkdir(exist_ok=True)
INBOX_DIR.mkdir(exist_ok=True)
def init_team(teammates: list) -> None:
config = {
"teammates": [
{"name": t["name"], "role": t["role"], "status": "IDLE"}
for t in teammates
]
}
(TEAM_DIR / "config.json").write_text(json.dumps(config, indent=2))
for t in teammates:
(INBOX_DIR / f"{t['name']}.jsonl").touch()
def get_roster() -> list:
config = json.loads((TEAM_DIR / "config.json").read_text())
return config["teammates"]
- Mailboxes are append-only JSONL files. Drain means read-and-truncate.
def send_message(to: str, from_: str, content: str) -> str:
inbox = INBOX_DIR / f"{to}.jsonl"
msg = {"from": from_, "content": content}
with open(inbox, "a") as f:
f.write(json.dumps(msg) + "\n")
return f"Message sent to {to}."
def drain_inbox(name: str) -> list:
inbox = INBOX_DIR / f"{name}.jsonl"
if not inbox.exists() or inbox.stat().st_size == 0:
return []
lines = inbox.read_text().strip().split("\n")
inbox.write_text("") # truncate after reading
return [json.loads(l) for l in lines if l]
- Each teammate runs its own agent loop in a thread.
def teammate_loop(name: str, role: str) -> None:
system = f"You are {name}, a {role}. Check your inbox for tasks."
while True:
inbox_messages = drain_inbox(name)
if not inbox_messages:
# IDLE: wait for work
threading.Event().wait(timeout=2)
continue
update_status(name, "WORKING")
history = [{"role": "user", "content":
"\n".join(m["content"] for m in inbox_messages)}]
# run agent loop with history
result = run_agent(history, system)
send_message("lead", name, result)
update_status(name, "IDLE")
What Changed From Background Tasks
| Component | Before (Background Tasks) | After (Agent Teams) |
|---|---|---|
| Agents | One main + bg threads | Lead + named teammates |
| Communication | Queue (one-way) | Mailboxes (bidirectional) |
| Identity | None | Name, role, lifecycle status |
| Coordination | None | Roster + send/drain pattern |
Key Takeaway
The mailbox pattern is what makes agent teams work. Each teammate has a private inbox on disk β durable, concurrent-safe, and transparent. The lead delegates by sending a message; the teammate drains its inbox and responds. No shared memory, no locks needed β just files.
Interactive Code Walkthrough
1TEAM_DIR = Path(".team")2INBOX_DIR = TEAM_DIR / "inbox"3 4def send_message(to: str, from_: str, content: str) -> str:5 inbox = INBOX_DIR / f"{to}.jsonl"6 msg = {"from": from_, "content": content}7 with open(inbox, "a") as f:8 f.write(json.dumps(msg) + "\n")9 return f"Message sent to {to}."10 11def drain_inbox(name: str) -> list:12 inbox = INBOX_DIR / f"{name}.jsonl"13 if not inbox.exists() or inbox.stat().st_size == 0:14 return []15 lines = inbox.read_text().strip().split("\n")16 inbox.write_text("") # truncate after reading17 return [json.loads(l) for l in lines if l]18 19def teammate_loop(name: str, role: str) -> None:20 while True:21 inbox_messages = drain_inbox(name)22 if not inbox_messages:23 threading.Event().wait(timeout=2)24 continue25 update_status(name, "WORKING")26 result = run_agent(inbox_messages, name, role)27 send_message("lead", name, result)28 update_status(name, "IDLE")29 .team/ directory. Each teammate gets a JSONL file as their inbox β a simple, crash-safe, concurrent-friendly communication channel.Compare [Subagents](/en/s04-subagent) to [Agent Teams](/en/s09-agent-teams) teammates. List 3 specific scenarios where you'd use each. What's the key factor in choosing?
Hint
Subagents for one-off tasks with no memory needed. Teammates for ongoing work with identity and communication.
Spawn two teammates with different specialties and have them collaborate on a task through the mailbox.
Hint
Give them complementary system prompts like 'frontend expert' and 'backend expert'.
Add a health monitoring system: the lead agent pings each teammate every 10 seconds. If a teammate doesn't respond within 5 seconds, mark it as UNRESPONSIVE and reassign its tasks.
Hint
Use a special 'ping' message type and track last response time per teammate.