Multi-Agent Sequential Channel

This guide shows how to use channel_dispatch() to route a message through a sequence of agents, where each agent sees the accumulated responses of all agents that came before it. Agents that have nothing to add emit a Silent event and are filtered out, keeping the output clean.

Concepts covered

• channel_dispatch() — sequential multi-agent dispatch
• AgentConfig — agent identity and behavior
• ChannelAgentMessage — an agent's response
• Silent — an agent chose not to respond
• RuntimeDriver — the LLM backend

---

What you'll build

A contract review channel where three agents review a contract sequentially:

1. Legal Reviewer — checks legal compliance, liability, IP, and dispute resolution
2. Technical Reviewer — checks technical feasibility and integration risks
3. Business Reviewer — checks business value, budget, and timeline

Each agent sees the original contract plus every prior agent's observations. If an agent finds nothing to flag, it stays silent.

Channel vs. Operation tree

channel_dispatch() is a standalone function — it does not use the Operation tree. It calls the runtime directly with assembled instructions. Use it for conversational patterns where each agent build

s on prior responses.

---

1 Define the domain model

The shared data that every agent sees:

from pydantic import BaseModel, Field


class ContractReviewItem(BaseModel):
    """A contract that multiple agents review sequentially."""
    title: str = Field(description="Contract title")
    parties: str = Field(description="Parties involved")
    description: str = Field(description="Contract purpose")
    value: str = Field(description="Estimated contract value")
    key_terms: str = Field(description="Summary of key terms and conditions")

---

2 Define agent configurations

Each agent gets its own AgentConfig with an identity block. Identity blocks describe the agent's role and guidelines — the runtime includes them in every dispatch:

from crewmaster.agents.agent import AgentConfig

legal_reviewer = AgentConfig(
    name="legal_reviewer",
    identity_blocks=[
        """You are a Legal Reviewer. Focus on:
- Legal compliance, liability, IP ownership, dispute resolution
- Identify potential legal pitfalls and suggest mitigations
- Be concise — flag what matters most
- If the contract is legally sound, acknowledge briefly""",
    ],
)

technical_reviewer = AgentConfig(
    name="technical_reviewer",
    identity_blocks=[
        """You are a Technical Reviewer. Focus on:
- Technical feasibility of the proposed solution
- Integration complexity with existing systems
- Technology choices, scalability, and architecture risks
- If no technical concerns exist, acknowledge briefly""",
    ],
)

business_reviewer = AgentConfig(
    name="business_reviewer",
    identity_blocks=[
        """You are a Business Reviewer. Focus on:
- Business value and ROI of the contract
- Budget appropriateness and payment terms
- Strategic alignment and timeline feasibility
- If the terms are favorable, acknowledge briefly""",
    ],
)

Using prompt blocks instead of inline strings

For production use, store identity blocks as Jinja2 templates in a LocalDiskStore and reference them via blocks:// URIs. Inline strings work well for simple scenarios:

agent = AgentConfig(
    name="reviewer",
    identity_blocks=["blocks://review/identity/legal"],
)

---

3 Set up the agent order and mapping

The application controls the order in which agents receive the message. channel_dispatch() does not rank agents — it follows the explicit agent_order you provide:

agent_order = ["legal_reviewer", "technical_reviewer", "business_reviewer"]

agents = {
    "legal_reviewer": legal_reviewer,
    "technical_reviewer": technical_reviewer,
    "business_reviewer": business_reviewer,
}

---

4 Prepare the shared context

Format the contract data into a readable string that all agents see:

contract = ContractReviewItem(
    title="Software Development Services Agreement",
    parties="NovaTech Inc. (Client) and CodeForge Solutions LLC (Vendor)",
    description=(
        "CodeForge will develop a custom ERP module for NovaTech, "
        "integrating with Salesforce and SAP. The module handles "
        "inventory management, order processing, and supplier management."
    ),
    value="$450,000 (fixed price, milestone-based payments)",
    key_terms=(
        "6-month timeline; 12-month warranty; vendor retains IP for "
        "generic components, client owns custom modules; 30-day payment "
        "terms; liability cap at 100% of contract value; arbitration in "
        "Delaware."
    ),
)

shared_context = (
    f"Contract: {contract.title}\n"
    f"Parties: {contract.parties}\n"
    f"Description: {contract.description}\n"
    f"Value: {contract.value}\n"
    f"Key Terms: {contract.key_terms}"
)

---

5 Dispatch the channel

Call channel_dispatch() with the message, agent list, and runtime. Iterate over the async generator to collect events:

import asyncio

from crewmaster.conversation.channel import (
    ChannelAgentMessage,
    Silent,
    channel_dispatch,
)


async def review_contract(driver, context_str: str):
    message = f"Please review the following contract:\n\n{context_str}"
    events = []

    async for event in channel_dispatch(
        message=message,
        agent_order=agent_order,
        agents=agents,
        shared_context=context_str,
        runtime=driver,
    ):
        events.append(event)

    # Analyze results
    for event in events:
        if isinstance(event, ChannelAgentMessage):
            print(f"\n[{event.agent_name}]:")
            print(f"  {event.content[:200]}...")
        elif isinstance(event, Silent):
            print(f"\n[{event.agent_name}]: (no concerns)")

    return events

---

6 Execute with a real driver

Wire it up with a PydanticAIDriver:

from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver


async def main():
    driver = PydanticAIDriver(model_name="openai:gpt-4o-mini")

    events = await review_contract(driver, shared_context)

    messages = [e for e in events if isinstance(e, ChannelAgentMessage)]
    silent = [e for e in events if isinstance(e, Silent)]

    print(f"\nTotal agents: {len(agent_order)}")
    print(f"Responded:    {len(messages)}")
    print(f"Silent:       {len(silent)}")


asyncio.run(main())

Expected output:

[legal_reviewer]:
  Key concerns: (1) The liability cap at 100% of contract value is
  acceptable but should specify whether it covers direct damages only...

[technical_reviewer]:
  Integration with Salesforce and SAP is complex. The 6-month timeline
  is aggressive. Recommend a phased rollout with technical milestones...

[business_reviewer]:
  The $450K fixed price for an ERP module is within market range. The
  30-day payment terms are standard. Strategic alignment is strong...

Total agents: 3
Responded:    3
Silent:       0

---

7 Handling silent agents

When an agent has nothing to contribute, it returns empty text. The channel emits a Silent event instead of a ChannelAgentMessage:

# If legal_reviewer identity says "only respond if you find issues"
# and the contract is flawless, the event stream looks like:

# [Silent(agent_name="legal_reviewer")]
# [ChannelAgentMessage(agent_name="technical_reviewer", content="...")]
# [ChannelAgentMessage(agent_name="business_reviewer", content="...")]

# Filter out silent agents when building a summary:
messages = [e.content for e in events if isinstance(e, ChannelAgentMessage)]
print("\n".join(messages))

The silent behavior is controlled by the agent's identity block — include instructions like "if you find no issues, respond with an empty message" to enable filtering.

---

8 Complete script

channel_review.py

"""Multi-agent channel: contract review with 3 agents."""
import asyncio

from pydantic import BaseModel, Field

from crewmaster.agents.agent import AgentConfig
from crewmaster.conversation.channel import (
    ChannelAgentMessage,
    Silent,
    channel_dispatch,
)
from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver


# ── Domain model ─────────────────────────────────────────────────
class ContractReviewItem(BaseModel):
    title: str
    parties: str
    description: str
    value: str
    key_terms: str


# ── Agents ───────────────────────────────────────────────────────
legal_reviewer = AgentConfig(
    name="legal_reviewer",
    identity_blocks=[
        """You are a Legal Reviewer. Focus on legal compliance,
liability, IP ownership, and dispute resolution. Be concise."""
    ],
)

technical_reviewer = AgentConfig(
    name="technical_reviewer",
    identity_blocks=[
        """You are a Technical Reviewer. Focus on technical
feasibility, integration complexity, and architecture risks."""
    ],
)

business_reviewer = AgentConfig(
    name="business_reviewer",
    identity_blocks=[
        """You are a Business Reviewer. Focus on business value,
ROI, budget, payment terms, and strategic alignment."""
    ],
)

agent_order = ["legal_reviewer", "technical_reviewer", "business_reviewer"]
agents = {
    "legal_reviewer": legal_reviewer,
    "technical_reviewer": technical_reviewer,
    "business_reviewer": business_reviewer,
}


# ── Contract data ────────────────────────────────────────────────
contract = ContractReviewItem(
    title="Software Development Services Agreement",
    parties="NovaTech Inc. (Client) and CodeForge Solutions LLC (Vendor)",
    description=(
        "CodeForge will develop a custom ERP module integrating "
        "with Salesforce and SAP for inventory management."
    ),
    value="$450,000 (fixed price, milestone-based)",
    key_terms=(
        "6-month timeline; 12-month warranty; vendor retains IP "
        "for generic components, client owns custom modules; "
        "30-day payment terms; liability cap at 100% contract "
        "value; arbitration in Delaware."
    ),
)

shared_context = (
    f"Contract: {contract.title}\n"
    f"Parties: {contract.parties}\n"
    f"Description: {contract.description}\n"
    f"Value: {contract.value}\n"
    f"Key Terms: {contract.key_terms}"
)


# ── Execute ──────────────────────────────────────────────────────
async def main():
    driver = PydanticAIDriver(model_name="openai:gpt-4o-mini")
    message = f"Please review:\n\n{shared_context}"

    events = []
    async for event in channel_dispatch(
        message=message,
        agent_order=agent_order,
        agents=agents,
        shared_context=shared_context,
        runtime=driver,
    ):
        events.append(event)

    messages = [e for e in events if isinstance(e, ChannelAgentMessage)]
    silent = [e for e in events if isinstance(e, Silent)]

    for msg in messages:
        print(f"\n--- {msg.agent_name} ---")
        print(msg.content)

    print(f"\nResponded: {len(messages)} / {len(agent_order)}")
    if silent:
        print(f"Silent: {', '.join(s.agent_name for s in silent)}")


asyncio.run(main())

---

Next steps

• Extend the channel to 1:1 dialogue with the dialogue() function.
• Use grill-me collaboration for deeper critique-and-revision between two agents.
• Read the Conversation API reference for full documentation on channel_dispatch, Message, Role, and event types.