langgraph-agent-teams 0.2.0

A multi-agent teams implementation using LangGraph (Blackboard & Mailbox architecture)

🤝 LangGraph Agent Teams

A Python library for building agent teams with LangGraph, inspired by the Claude Code Agent Teams feature. One agent acts as the team lead — coordinating work, assigning tasks, and synthesizing results — while teammates work independently, each in its own context window, claiming tasks from a shared task list and messaging each other through mailboxes.

Unlike supervisor/subagent patterns where workers only report back to a caller, agent teams coordinate through two shared primitives:

• Task board — a shared list of work items with dependencies (blocks / blocked_by). The lead creates tasks; teammates claim and complete them. Claiming uses file locking, so concurrent teammates never grab the same task.
• Mailbox — asynchronous, direct agent-to-agent messaging. Any member can message any other member by name — teammate-to-teammate debate is a first-class capability, not just teammate-to-lead reporting. Idle notifications flow to the lead automatically, and the team shuts down gracefully via a shutdown request/response protocol.

As in Claude Code, the coordination tools (SendMessage plus the task tools) are always available to every team member — define members with TeamMemberSpec and the package attaches them automatically.

All coordination state (task board, mailboxes, roster) is persisted as JSON files on disk — you can cat a mailbox or ls the task list at any time while the team is running, and state survives crashes.

Features

• 👑 Lead / teammate architecture — a coordinator agent delegates to independent worker agents
• 📋 Shared task board — dependency-aware tasks with atomic, lock-protected claiming
• 📬 Inter-agent mailboxes — every member can message every other member (or broadcast), with idle notifications and graceful shutdown
• 🧰 Coordination tools always on — TeamMemberSpec members automatically get SendMessage + task tools, like Claude Code teammates
• ⚡ True parallelism — teammates run concurrently as async LangGraph nodes
• 🗂️ File-system-as-a-bus — inspect and debug the team's state live on disk
• 🛠️ LangChain tools included — SendMessage, TaskCreate, TaskUpdate, TaskList, TaskGet
• 🪝 Lifecycle hooks — validate task creation, gate task completion behind quality checks, and keep idle teammates working (like Claude Code's TaskCreated / TaskCompleted / TeammateIdle hooks)
• 📝 Plan approval — require a teammate to get its plan approved by the lead before executing each task
• 🔁 Session resumption — run with fresh=False to resume an interrupted team: completed tasks stay done, interrupted tasks are requeued

When to Use Agent Teams

Agent teams shine when parallel exploration adds real value:

• Research and review — teammates investigate different aspects of a problem simultaneously
• New modules or features — each teammate owns a separate piece without stepping on the others
• Competing hypotheses — teammates test different theories in parallel and converge faster
• Cross-layer work — frontend, backend, and tests each owned by a different teammate

For sequential tasks or work with many dependencies, a single agent or a supervisor/subagent pattern is more effective — teams add coordination overhead and use more tokens.

Installation

pip install langgraph-agent-teams

Quickstart

pip install langgraph-agent-teams langchain-openai

export OPENAI_API_KEY=<your_api_key>

import asyncio

from langgraph_agent_teams import TeamMemberSpec, create_agent_team

# Every member automatically gets the coordination tools:
# SendMessage, TaskCreate, TaskUpdate, TaskList, TaskGet
lead = TeamMemberSpec(
    name="team-lead",
    model="openai:gpt-5.5",
    system_prompt="You are a team lead. Delegate work to your teammates.",
)
researcher = TeamMemberSpec(
    name="researcher",
    model="openai:gpt-5.5",
    system_prompt="You are a thorough researcher.",
    agent_type="researcher",
)
critic = TeamMemberSpec(
    name="critic",
    model="openai:gpt-5.5",
    system_prompt="You are a devil's advocate who challenges findings.",
    agent_type="critic",
)

app = create_agent_team(
    lead, [researcher, critic], team_name="research-team"
).compile()

result = asyncio.run(
    app.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "Evaluate whether we should migrate from REST to GraphQL. "
                    "Have one teammate research the benefits and another challenge them.",
                }
            ]
        }
    )
)
print(result["messages"][-1].content)

[!IMPORTANT] Agent teams rely on asyncio concurrency — the lead and all teammates run at the same time. Always invoke the compiled graph with await app.ainvoke(...) (or asyncio.run(...)); synchronous invoke is not supported.

How It Works

              ┌─────────────┐
   user ────► │  team lead  │ ◄──── idle notifications, reports, replies
              └──────┬──────┘
        TaskCreate / │ SendMessage
         TaskUpdate  ▼
   ┌───────────────────────────────┐
   │      shared task board        │   {base_dir}/tasks/{team}/
   │  1.json  2.json  .lock  ...   │
   └──────┬─────────────────┬──────┘
          │ claim (locked)  │ claim (locked)
   ┌──────▼──────┐   ┌──────▼──────┐
   │ teammate A  │   │ teammate B  │   each with its own context window
   └─────────────┘   └─────────────┘
          ▲                 ▲
          └───── mailboxes ─┘           {base_dir}/teams/{team}/inboxes/

1. A setup node resets the team's on-disk state and registers every member in the roster.
2. The lead node runs the lead agent with a coordination briefing. The lead breaks the request into tasks (TaskCreate), optionally assigns them (TaskUpdate with owner), and messages teammates (SendMessage).
3. Each teammate node runs a poll loop: read unread mail → claim the next available task → run its own agent turn → mark the task complete → report the result and an idle notification back to the lead.
4. Teammate messages are delivered to the lead automatically; the lead responds, reassigns, or creates follow-up work.
5. When every task is complete and all teammates are idle, the lead broadcasts a shutdown request; each teammate acknowledges with a shutdown response and exits. The lead's final message synthesizes the team's results.

Task dependencies

Tasks can depend on each other. A pending task with unresolved dependencies cannot be claimed until those dependencies are completed — blocked tasks unblock automatically:

# Inside the lead's reasoning, via the TaskCreate / TaskUpdate tools:
#   TaskCreate(subject="Research API options", description="...")            -> task 1
#   TaskCreate(subject="Write recommendation", description="...", blocked_by=["1"])  -> task 2

Lead-assigned vs. self-claimed tasks

• Lead assigns: the lead sets owner on a task with TaskUpdate — only that teammate can claim it.
• Self-claim: unassigned tasks are claimed by idle teammates in ID order.

Teammate-to-teammate communication

Every TeamMemberSpec member can message any other member by name with SendMessage — the classic "teammates debate each other" pattern from the Claude Code docs works out of the box:

result = await app.ainvoke({"messages": [{
    "role": "user",
    "content": "Investigate why the app disconnects after one message. "
    "Have teammates pursue different hypotheses and message each other "
    "to try to disprove each other's theories, like a scientific debate.",
}]})

Communication is agent-driven with a harness fallback: if a teammate finishes a turn without calling SendMessage, its final reply is forwarded automatically — task reports go to the lead, message replies go back to the sender — so no work is silently lost even with models that under-use tools. Fallback replies are marked internally and never trigger another automatic reply, so two teammates can't ping-pong forever.

Lifecycle hooks

Mirror Claude Code's team hooks with plain Python callables. Each hook returns None to allow the event or a feedback string to reject it — the feedback is delivered to the acting agent so it can adjust:

from langgraph_agent_teams import TeamHooks, create_agent_team

def quality_gate(task):
    # e.g. run tests, check the task's metadata, inspect files...
    if "test" not in task.description.lower():
        return "Completion rejected: describe how you verified the work."
    return None  # allow completion

hooks = TeamHooks(
    on_task_created=lambda task: None,        # validate new tasks
    on_task_completed=quality_gate,           # gate completion (quality gate)
    on_teammate_idle=lambda name: None,       # keep an idle teammate working
)

app = create_agent_team(lead, teammates, team_name="research-team", hooks=hooks).compile()

• on_task_created(task) — rejecting deletes the task and returns the feedback to the creating agent.
• on_task_completed(task) — rejecting keeps the task in_progress and sends the feedback to the working teammate, which keeps going until the gate passes.
• on_teammate_idle(name) — returning feedback gives the teammate another turn instead of letting it go idle.

Plan approval

Require a teammate to plan before acting — the equivalent of Claude Code's plan approval for teammates. Set require_plan_approval=True on the spec; before executing each task the teammate submits a plan to the lead, and the lead approves or rejects it (with feedback) using the automatically attached ApprovePlan tool:

coder = TeamMemberSpec(
    name="coder",
    model="openai:gpt-5.5",
    system_prompt="You write production code.",
    require_plan_approval=True,  # plan first, execute after lead approval
)

Rejected plans are revised and resubmitted until approved. If the lead doesn't respond within plan_approval_timeout seconds (default 120), the teammate proceeds so the team never deadlocks.

Resuming an interrupted team

Because the task board lives on disk, a crashed or interrupted run can be resumed. Pass fresh=False and the board is kept: completed tasks stay completed, tasks that were in_progress are requeued, and teammates pick up where the previous run stopped:

app = create_agent_team(
    lead, teammates, team_name="research-team", fresh=False
).compile()
result = await app.ainvoke({"messages": [{"role": "user", "content": "Continue the work."}]})

The default (fresh=True) wipes the team's state at the start of every run.

Using prebuilt agents (advanced)

You can also pass prebuilt agents (e.g. from langchain.agents.create_agent) instead of specs — useful when you need middleware or custom agent construction. In that case, attach the coordination tools yourself so the agent can communicate:

from langchain.agents import create_agent
from langgraph_agent_teams import create_team_tools

coder = create_agent(
    model="openai:gpt-5.5",
    tools=[*my_tools, *create_team_tools("coder", team_name="research-team")],
    name="coder",
)

Prebuilt agents without team tools still work — the harness claims tasks, completes them, and forwards their replies for them.

Inspecting a Running Team

Team state lives under ~/.langgraph-agent-teams by default (override with base_dir):

~/.langgraph-agent-teams/
├── teams/{team_name}/
│   ├── config.json          # Roster: members, roles, live status
│   └── inboxes/
│       ├── team-lead.json   # The lead's mailbox
│       └── researcher.json  # A teammate's mailbox
└── tasks/{team_name}/
    ├── .lock                # Claim lock
    ├── .highwatermark       # Highest task ID ever created
    └── 1.json               # Task #1

# Watch the task board while the team works
watch -n1 'cat ~/.langgraph-agent-teams/tasks/research-team/*.json | jq -r "\(.id) [\(.status)] \(.owner // \"unassigned\") — \(.subject)"'

You can also use the store classes directly:

from langgraph_agent_teams import get_team_resources

resources = get_team_resources("research-team")
for task in resources.task_board().list_tasks():
    print(task.id, task.status, task.owner, task.subject)
for message in resources.mailbox().read_all("team-lead"):
    print(message.from_agent, "->", message.summary or message.text[:80])

API Reference

TeamMemberSpec(name, model, system_prompt=None, tools=[], agent_type=None, require_plan_approval=False)

Declarative definition of one team member. The package builds the agent with langchain.agents.create_agent and automatically attaches the five coordination tools bound to the member's identity. tools adds extra domain tools on top; agent_type is a role label shown in the roster and briefings; require_plan_approval makes the teammate submit a plan to the lead before executing each task.

create_agent_team(lead, teammates, team_name, base_dir=None, *, fresh=True, hooks=None, poll_interval=0.5, shutdown_timeout=60.0, plan_approval_timeout=120.0, max_agent_turns=100)

Builds the team StateGraph. The lead and every teammate run as concurrent async nodes; compile and ainvoke it like any LangGraph graph.

Parameter	Description
lead	TeamMemberSpec (recommended) or a prebuilt agent equipped with create_team_tools.
teammates	TeamMemberSpecs (recommended) or prebuilt agents, each with a unique name.
team_name	Unique team name; also names the on-disk state directories.
base_dir	Root directory for team state (default ~/.langgraph-agent-teams).
fresh	True (default) wipes team state at the start; False resumes the existing task board, requeuing interrupted tasks.
hooks	Optional TeamHooks lifecycle callbacks (see above).
poll_interval	Seconds between mailbox/task-board polls.
shutdown_timeout	Max seconds the lead waits for shutdown acknowledgements.
plan_approval_timeout	Max seconds a teammate waits for a plan verdict before proceeding.
max_agent_turns	Safety cap on agent invocations per member per run.

TeamHooks(on_task_created=None, on_task_completed=None, on_teammate_idle=None)

Lifecycle callbacks, mirroring Claude Code's team hooks. Each is a plain callable returning None to allow the event or a feedback string to reject it (see Lifecycle hooks).

create_team_tools(agent_name, team_name, base_dir=None, hooks=None)

Returns the five coordination tools bound to one agent's identity:

Tool	Purpose
SendMessage	DM another agent by name, or "*" to broadcast
TaskCreate	Add a task to the shared board (optionally blocked_by)
TaskUpdate	Change status/owner/description, add dependency edges, or set status deleted to remove a task
TaskList	List every task with status, owner, and blockers
TaskGet	Fetch one task's full details

create_plan_approval_tool(agent_name, team_name, base_dir=None)

Returns the ApprovePlan tool the lead uses to answer plan approval requests. Attached automatically to a spec-built lead when any teammate has require_plan_approval=True; attach it manually to a prebuilt lead.

Store classes

SwarmTaskBoard, AgentMailbox, and TeamRoster are exported for direct use — every operation is atomic and protected by portalocker file locks, so they're safe across threads, asyncio tasks, and even separate processes.

Comparing Agent Teams with Other Multi-Agent Patterns

Pattern	Coordination	Communication	Best for
Agent Teams	Shared task list + lead	Teammates message each other directly	Parallel work requiring discussion and collaboration
Group Chat	Centralized selector	Broadcast (shared history)	Moderated turn-taking discussions
Swarm	Decentralized handoffs	Control transfer via tools	Fluid handoffs, direct user interaction
Subagents	Main agent delegates	Results report back to caller only	Focused tasks where only the result matters

Use subagents when you need quick, focused workers that report back. Use agent teams when workers benefit from a shared work queue, independent context windows, and inter-agent communication.

Best Practices

1. Give teammates enough context — teammates don't see the lead's conversation history, so the lead's task descriptions must be self-contained (the built-in briefing tells it so)
2. Size tasks appropriately — self-contained units with a clear deliverable; 5–6 tasks per teammate keeps everyone productive
3. Start with 3–5 teammates — token costs scale linearly with team size, and coordination overhead grows with it
4. Avoid file conflicts — if teammates edit files, break work so each owns a different set
5. Use dependencies for ordering — blocked_by edges let the board sequence work without the lead micromanaging

Examples

See the examples/ directory:

• software_team — a lead coordinating researcher/coder/reviewer teammates

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

MIT License — see LICENSE for details.

Acknowledgments

• Inspired by the Claude Code Agent Teams feature and its blackboard/mailbox architecture
• Built on LangGraph and LangChain agents