Collaborative teams

CollaborativeTeam coordinates teammates through a shared TaskBoard.

Unlike HierarchicalTeam (leader delegates via member tools and synthesizes), collaborative mode:

1. Leader creates tasks and assigns each to the right teammate by role
2. Members complete assigned work — either in phased parallel rounds after seed, or with streaming dispatch (members start as soon as they are assigned, overlapping seed/replan)
3. If the board is still incomplete and max_replans allows it, the leader replans (more add_task / assign_task), then member work continues
4. Leader synthesizes a final answer from the board and peer messages (toolless cycle: board mutation tools are not available; the completed board and message log are passed in the prompt)

Members may message each other directly (send_message / list_messages) without routing through the leader. The leader can observe messages with list_messages during seed/replan. Members cannot create dependency-linked tasks in this version; nested teams are not supported as members. See the Roadmap for planned directions.

Review (opt-in): with require_review=True, new tasks stamp reviewer to the leader; complete moves them to pending_review until approve_task / reject_task. Use assign_reviewer to gate a single task (even when the flag is off) or to delegate review to a member. Rejected work goes to needs_revision with rejection_reason kept on the board; done is terminal.

flowchart TD
  seed[Lead seed add_task and assign_task] --> rounds[Member work]
  rounds --> done{board_complete?}
  done -->|yes| synth[Lead synthesizes]
  done -->|no| budget{replans_left?}
  budget -->|yes| replan[Lead replan]
  replan --> rounds
  budget -->|no| synth

In dispatch_mode='streaming', member ticks can start during seed/replan (not only after those leader turns finish). Phased mode keeps a barrier between seed and member rounds — parallel members alone do not optimize end-to-end latency.

Observing a run

team.run(...) is a thin wrapper around CollaborativeRun. For step-by-step observation (inspired by pydantic-graph iter, without modeling the team as a GraphBuilder):

from pydantic_team import (
    CollaborativeTeam,
    MessagePosted,
    PhaseJoined,
    RunEnded,
    TaskReviewDecided,
    TasksScheduled,
)

async with team.iter('Produce a short report') as run:
    async for event in run:
        if isinstance(event, TasksScheduled):
            print('scheduled', [t.kind for t in event.tasks])
        elif isinstance(event, MessagePosted):
            print('message', event.message.sender, '->', event.message.to)
        elif isinstance(event, TaskReviewDecided):
            print('review', event.decision, event.task.id)
        elif isinstance(event, PhaseJoined):
            print('joined', event.phase, 'incomplete=', event.incomplete)
        elif isinstance(event, RunEnded):
            print('done', event.result.data)
    assert run.result is not None

Events: TasksScheduled, TaskCompleted, PhaseJoined, MessagePosted, TaskReviewDecided, RunEnded (see TeamTask). Use run.board for a live snapshot (including messages_snapshot()).

Construction

from pydantic_ai import Agent
from pydantic_team import CollaborativeTeam

researcher = Agent(
    'openai:gpt-4.1',
    name='researcher',
    instructions='Complete only research tasks assigned to you.',
)
writer = Agent(
    'openai:gpt-4.1',
    name='writer',
    instructions='Complete only writing tasks assigned to you.',
)

team = CollaborativeTeam(
    leader_model='openai:gpt-4.1',
    members=[researcher, writer],
    system_prompt_override=(
        'Assign every task to researcher or writer by role; never leave tasks open.'
    ),
    max_rounds=3,
    max_replans=2,
    dispatch_mode='streaming',  # members start on assign; default is 'phased'
    # require_review=True,  # stamp reviewer=leader; complete → pending_review
)
result = await team.run('Produce a short report on agent teams')
print(result.data)
print(result.usage)

• dispatch_mode: 'phased' (default) = seed barrier then member rounds; 'streaming' = ready-queue dispatch so members overlap with seed/replan
• max_rounds: in phased mode, parallel member ticks per phase; in streaming, max ticks per member per phase (after seed and after each replan)
• max_replans: how many times the leader may replan after an incomplete member phase (default 0 = seed → work → synthesize only)
• max_assignments_per_tick: optional cap on how many incomplete assignments are listed for a member in one tick (forces leftover work into later ticks / replan)
• require_review: when True, new tasks get reviewer=leader so complete goes to pending_review until approve/reject; when False (default), complete → done unless assign_reviewer set a reviewer on that task
• Members must be agents (nested teams are not supported in this slice; see Roadmap)
• Pass usage= as a team-level aggregate: each seed / replan / synthesize / member tick is an isolated agent.run with its own usage budget (so the default request_limit applies per cycle), then folded into the aggregate
• Prefer assign-by-role over free-for-all claim so a writer does not take research work

Board operations

Who	Tools
Lead (seed / replan)	add_task, assign_task, assign_reviewer, approve_task, reject_task, list_tasks, list_messages
Lead (synthesize)	none — final answer only (board + messages in prompt)
Members	list_tasks, claim_task, complete_task, assign_reviewer, approve_task, reject_task, send_message, list_messages

send_message(to, body, task_id='') posts a peer DM (to = teammate id) or broadcast (to='*'). Optional task_id links the message to an existing task. Direct messages wake the recipient in streaming dispatch (same wakeup path as assign/claim).

After assign_task, the task is claimed for that assignee (not stealable via claim). Claim remains for residual open tasks only.

Orchestration phases (collaborative.run / .seed / .round / .replan / .synthesize, plus .dispatch / .member_tick / .review_tick when review is active) emit OpenTelemetry spans when instrument_pydantic_team is enabled. Board tool calls are visible via logfire.instrument_pydantic_ai() — see Observability.

Task model

See Task / TaskStatus:

open → claimed → done when no reviewer is set.

With a reviewer: claimed (or needs_revision) → pending_review → done (approve) or needs_revision (reject). Fields: optional assignee, result, reviewer, rejection_reason.

Peer messages use BoardMessage on the same board (sender, to, body, optional task_id).

Live example with review: examples/collaborative_review.py (require_review=True; default path without a gate remains examples/collaborative_basic.py).

Testing

Use TestModel and agent.override — same as hierarchical teams. Unit tests cover board races without live LLM calls.