Skip to content

Grill-Me Collaboration Protocol¤

This guide shows how to use GrillMeProtocol to have two agents engage in structured critique-and-revision cycles. A critic rigorously evaluates the respondent's output, the respondent revises based on feedback, and this loop continues until the output converges or a maximum number of turns is reached.

Concepts covered

What you'll build¤

A business initiative review where a respondent drafts a proposal and a critic challenges it across multiple rounds:

sequenceDiagram participant R as Respondent participant C as Critic R->>R: Draft initial proposal R->>C: Submit for critique C->>C: Identify weaknesses & risks C->>R: Return critique R->>R: Revise proposal Note over R,C: Repeat up to max_turns R->>R: Final polished proposal
  • Respondent produces an InitiativeProposal (executive summary, problem, solution, metrics, risks)
  • Critic evaluates each iteration and provides specific feedback
  • The loop terminates at convergence (stalemate_threshold) or max_turns
  • Both agents can use request_capability to dynamically fetch tools

1 Define domain models¤

from pydantic import BaseModel, ConfigDict, Field


class InitiativeSeed(BaseModel):
    """Seed data for a business initiative proposal."""
    model_config = ConfigDict(frozen=True)

    title: str = Field(description="Initiative title")
    industry: str = Field(description="Industry sector")
    description: str = Field(description="What the initiative entails")
    rationale: str = Field(description="Business justification")
    budget: str = Field(description="Estimated budget")
    timeline: str = Field(description="Projected timeline")


class InitiativeProposal(BaseModel):
    """The artifact produced and revised by the respondent."""
    model_config = ConfigDict(frozen=True)

    title: str = Field(description="Initiative title")
    executive_summary: str = Field(description="One-paragraph executive summary")
    problem_statement: str = Field(description="Problem being solved")
    proposed_solution: str = Field(description="Detailed solution approach")
    success_metrics: str = Field(description="How success will be measured")
    risk_assessment: str = Field(description="Key risks and mitigations")

2 Create a context projector¤

Projects the seed data into template variables that the respondent's task blocks can reference:

from typing import Any

from crewmaster.agents.context.projector import ContextProjector


class InitiativeProjector(ContextProjector):
    """Projects InitiativeSeed into template context."""

    def __init__(self, title, industry, description, rationale,
                 budget, timeline):
        self._title = title
        self._industry = industry
        self._description = description
        self._rationale = rationale
        self._budget = budget
        self._timeline = timeline

    @classmethod
    def from_domain(cls, data: InitiativeSeed) -> "InitiativeProjector":
        return cls(
            title=data.title,
            industry=data.industry,
            description=data.description,
            rationale=data.rationale,
            budget=data.budget,
            timeline=data.timeline,
        )

    def to_context(self) -> dict[str, Any]:
        return {
            "initiative_title": self._title,
            "initiative_industry": self._industry,
            "initiative_description": self._description,
            "initiative_rationale": self._rationale,
            "initiative_budget": self._budget,
            "initiative_timeline": self._timeline,
        }

3 Define tools with rich capabilities¤

The grill-me protocol supports request_capability — agents can dynamically request tools at runtime by describing what they need. Register tools as Capability objects with descriptive text to enable this semantic matching:

from crewmaster.tools.models import Capability, ToolSchema
from crewmaster.tools.registry import ToolRegistry

# ── Respondent tools ──────────────────────────────────────────────────
MARKET_RESEARCH = ToolSchema(
    name="market_research",
    description="Access market research data for industry trends and competitors",
    input_schema={
        "type": "object",
        "properties": {
            "industry": {"type": "string"},
            "focus_area": {"type": "string"},
        },
        "required": ["industry"],
    },
)

RISK_ANALYSIS = ToolSchema(
    name="risk_analysis",
    description="Analyze financial, operational, and market risks",
    input_schema={
        "type": "object",
        "properties": {
            "initiative_type": {"type": "string"},
            "budget_range": {"type": "string"},
        },
    },
)

BENCHMARKS = ToolSchema(
    name="industry_benchmarks",
    description="Retrieve ROI and success rate benchmarks",
    input_schema={
        "type": "object",
        "properties": {
            "industry": {"type": "string"},
            "initiative_category": {"type": "string"},
        },
    },
)

# ── Critic tool ──────────────────────────────────────────────────────
EVIDENCE_CHECK = ToolSchema(
    name="evidence_check",
    description="Verify claims and check for supporting evidence",
    input_schema={
        "type": "object",
        "properties": {"claim": {"type": "string"}},
    },
)

# ── Registry ─────────────────────────────────────────────────────────
tool_registry = ToolRegistry()

# Register responder tools with rich descriptions for semantic retrieval
tool_registry.register(
    "proposal_writing",
    Capability(
        name="market_research",
        description=(
            "Market research data for SaaS and enterprise software. "
            "Covers industry trends, competitor analysis, customer "
            "demographics, and market sizing."
        ),
        tool_schema=MARKET_RESEARCH,
    ),
)
tool_registry.register(
    "proposal_writing",
    Capability(
        name="risk_analysis",
        description=(
            "Comprehensive risk analysis for business initiatives. "
            "Evaluates financial risk, operational risk, market risk, "
            "and technical risk with mitigation recommendations."
        ),
        tool_schema=RISK_ANALYSIS,
    ),
)
tool_registry.register(
    "proposal_writing",
    Capability(
        name="industry_benchmarks",
        description=(
            "Industry benchmark database with ROI ranges, success "
            "rates, and adoption curves for technology initiatives."
        ),
        tool_schema=BENCHMARKS,
    ),
)

# Register critic tools in their scope
tool_registry.register("critique", EVIDENCE_CHECK)

4 Configure agents¤

Critic — evaluates proposals rigorously. Respondent — drafts and revises. The respondent declares default_context=[InitiativeProjector] so the seed data is projected into its prompts:

from crewmaster.agents.agent import AgentConfig

critic = AgentConfig(
    name="critic",
    identity_blocks=[
        """You are an expert Critic specializing in rigorous review
of business initiatives.

Your role is to critically examine proposals and identify weaknesses,
risks, blind spots, and areas for improvement. Be constructive but
unflinching.

Structure your critiques: strengths, weaknesses, risks, and specific
recommendations for improvement.""",
    ],
    default_tool_scope="critique",
)

respondent = AgentConfig(
    name="respondent",
    identity_blocks=[
        """You are a business strategist who drafts and refines
initiative proposals.

You produce structured InitiativeProposal outputs with: executive
summary, problem statement, proposed solution, success metrics,
and risk assessment.

When receiving critique, address every point raised. If the critic
identifies a gap, fill it with data or reasoning. If you need
additional data, call request_capability to access research tools.""",
    ],
    default_tool_scope="proposal_writing",
    default_context=[InitiativeProjector],
)

5 Build the operation with GrillMeProtocol¤

The Operation uses the collaboration= field to attach the protocol. Set max_retrieval_rounds to enable request_capability:

from crewmaster.collaboration.protocol import GrillMeConfig, GrillMeProtocol
from crewmaster.operations.operation import Operation

operation = Operation(
    name="initiative_review",
    produces=InitiativeProposal,
    kind="artifact",
    agent=respondent,
    task_blocks=["blocks://grill_me/task/respondent"],
    consumes=[InitiativeProjector],
    collaboration=GrillMeProtocol(
        GrillMeConfig(
            critic=critic,
            respondent=respondent,
            max_turns=3,            # Up to 3 critique-revision cycles
            stalemate_threshold=2,   # Stop if output unchanged 2x in a row
        )
    ),
    max_retrieval_rounds=2,          # Allow 2 request_capability calls
)

6 Set up prompt blocks¤

The respondent's task block uses template variables from the projector:

import tempfile
from pathlib import Path

from crewmaster.agents.prompts.local_disk_store import LocalDiskStore

blocks_dir = Path(tempfile.mkdtemp())
(blocks_dir / "grill_me" / "task" / "respondent.j2").mkdir(parents=True)
(blocks_dir / "grill_me" / "task" / "respondent.j2").write_text("""\
---
id: blocks://grill_me/task/respondent
kind: task
provides: proposal_output
---
Draft a business initiative proposal for:

**Title:** {{ ctx.initiative_title }}
**Industry:** {{ ctx.initiative_industry }}
**Description:** {{ ctx.initiative_description }}
**Rationale:** {{ ctx.initiative_rationale }}
**Budget:** {{ ctx.initiative_budget }}
**Timeline:** {{ ctx.initiative_timeline }}

Include: executive_summary, problem_statement, proposed_solution,
success_metrics, and risk_assessment.
""")

block_store = LocalDiskStore(root=str(blocks_dir))

7 Execute and inspect the result¤

Call execute() as usual. The runtime detects the collaboration protocol and delegates to the GrillMeProtocol executor:

import asyncio

from crewmaster import execute
from crewmaster.agents.context.store import ContextStore


async def main():
    from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver

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

    # Register seed data
    seed = InitiativeSeed(
        title="AI-Powered Customer Support Platform",
        industry="SaaS & Enterprise Software",
        description=(
            "Deploy an AI-powered customer support platform that "
            "handles tier-1 tickets automatically, reducing response "
            "time from 4 hours to under 5 minutes for 60% of inquiries."
        ),
        rationale=(
            "Support costs have grown 35% YoY while CSAT dropped to "
            "3.2/5. Competitors using AI report 40% cost reduction."
        ),
        budget="$1.2M annual (platform + integration + training)",
        timeline="Q3 2026: Pilot, Q4 2026: Full rollout",
    )

    store = ContextStore()
    store.register(InitiativeSeed, seed)

    result = await execute(
        operation=operation,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        tool_registry=tool_registry,
    )

    proposal: InitiativeProposal = result
    print(f"Title: {proposal.title}")
    print(f"Executive Summary: {proposal.executive_summary}")
    print(f"Risk Assessment: {proposal.risk_assessment}")


asyncio.run(main())

8 Inspect collaboration internals¤

If you need per-participant history (e.g., to show the critique trail), access the CollaborationResult. The execute() function unwraps it by default, but you can intercept it by running the collaboration executor directly:

from crewmaster.collaboration.protocol import CollaborationResult


# The execute() function handles CollaborationResult internally.
# If you need raw inspection, access it through the plan introspection:
async def run_with_inspection():
    from crewmaster.operations.plans import resolve_plan

    plan = resolve_plan(
        operation=operation,
        context_store=store,
        block_store=block_store,
        default_runtime=driver,
        tool_registry=tool_registry,
    )

    # Plan nodes for collaboration protocols carry the protocol executor
    for node in plan.nodes:
        if node.collaboration is not None:
            collab_result = await node.collaboration.execute(
                operation=node.operation,
                runtime=driver,
                context_store=store,
            )
            print(f"Protocol: {collab_result.protocol_type}")
            print(f"Turns: {collab_result.turn_count}")
            for agent_name, outputs in collab_result.participant_outputs.items():
                print(f"  {agent_name}: {len(outputs)} contributions")

9 Stalemate detection¤

When the respondent produces the same output across consecutive turns, the protocol terminates early. This saves API costs and prevents infinite loops:

GrillMeConfig(
    critic=critic,
    respondent=respondent,
    max_turns=5,
    stalemate_threshold=2,  # Stop after 2 identical iterations
)

The comparison uses structural equality — for Pydantic models, this means model_dump() must return identical dictionaries. For string outputs, exact text comparison is used.

10 Complete script¤

grill_me_review.py 
"""Grill-me collaboration: initiative proposal review."""
import asyncio
import tempfile
from pathlib import Path
from typing import Any

from pydantic import BaseModel, ConfigDict, Field

from crewmaster import execute
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.projector import ContextProjector
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.local_disk_store import LocalDiskStore
from crewmaster.collaboration.protocol import GrillMeConfig, GrillMeProtocol
from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver
from crewmaster.operations.operation import Operation
from crewmaster.tools.models import Capability, ToolSchema
from crewmaster.tools.registry import ToolRegistry


# ── Domain models ─────────────────────────────────────────────────
class InitiativeSeed(BaseModel):
    model_config = ConfigDict(frozen=True)
    title: str
    industry: str
    description: str
    rationale: str
    budget: str
    timeline: str


class InitiativeProposal(BaseModel):
    model_config = ConfigDict(frozen=True)
    title: str
    executive_summary: str
    problem_statement: str
    proposed_solution: str
    success_metrics: str
    risk_assessment: str


# ── Context projector ─────────────────────────────────────────────
class InitiativeProjector(ContextProjector):
    def __init__(self, title, industry, description, rationale,
                 budget, timeline):
        self._title = title
        self._industry = industry
        self._description = description
        self._rationale = rationale
        self._budget = budget
        self._timeline = timeline

    @classmethod
    def from_domain(cls, data: InitiativeSeed) -> "InitiativeProjector":
        return cls(
            title=data.title, industry=data.industry,
            description=data.description, rationale=data.rationale,
            budget=data.budget, timeline=data.timeline,
        )

    def to_context(self) -> dict[str, Any]:
        return {
            "initiative_title": self._title,
            "initiative_industry": self._industry,
            "initiative_description": self._description,
            "initiative_rationale": self._rationale,
            "initiative_budget": self._budget,
            "initiative_timeline": self._timeline,
        }


# ── Tools ─────────────────────────────────────────────────────────
MARKET_RESEARCH = ToolSchema(
    name="market_research",
    description="Access market research data",
    input_schema={
        "type": "object",
        "properties": {"industry": {"type": "string"}},
        "required": ["industry"],
    },
)

EVIDENCE_CHECK = ToolSchema(
    name="evidence_check",
    description="Verify claims",
    input_schema={
        "type": "object",
        "properties": {"claim": {"type": "string"}},
    },
)

tool_registry = ToolRegistry()
tool_registry.register("proposal_writing", MARKET_RESEARCH)
tool_registry.register("critique", EVIDENCE_CHECK)


# ── Agents ────────────────────────────────────────────────────────
critic = AgentConfig(
    name="critic",
    identity_blocks=[
        """You are an expert Critic. Examine proposals rigorously.
Structure: strengths, weaknesses, risks, recommendations."""
    ],
    default_tool_scope="critique",
)

respondent = AgentConfig(
    name="respondent",
    identity_blocks=[
        """You are a business strategist. Draft and refine
initiative proposals with: executive summary, problem,
solution, metrics, risk assessment. Address all critique."""
    ],
    default_tool_scope="proposal_writing",
    default_context=[InitiativeProjector],
)


# ── Prompt block ─────────────────────────────────────────────────
BLOCKS_DIR = Path(tempfile.mkdtemp())
(BLOCKS_DIR / "grill_me" / "task" / "respondent.j2").mkdir(parents=True)
(BLOCKS_DIR / "grill_me" / "task" / "respondent.j2").write_text("""\
---
id: blocks://grill_me/task/respondent
kind: task
provides: proposal_output
---
Draft a proposal for {{ ctx.initiative_title }}.
Include: executive_summary, problem_statement, proposed_solution,
success_metrics, risk_assessment.
""")

block_store = LocalDiskStore(root=str(BLOCKS_DIR))


# ── Operation ─────────────────────────────────────────────────────
operation = Operation(
    name="initiative_review",
    produces=InitiativeProposal,
    kind="artifact",
    agent=respondent,
    task_blocks=["blocks://grill_me/task/respondent"],
    consumes=[InitiativeProjector],
    collaboration=GrillMeProtocol(
        GrillMeConfig(
            critic=critic,
            respondent=respondent,
            max_turns=3,
            stalemate_threshold=2,
        )
    ),
    max_retrieval_rounds=2,
)


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

    seed = InitiativeSeed(
        title="AI-Powered Customer Support Platform",
        industry="SaaS & Enterprise Software",
        description="Deploy AI-powered support handling tier-1 tickets.",
        rationale="Support costs up 35% YoY, CSAT dropped to 3.2/5.",
        budget="$1.2M annual",
        timeline="Q3-Q4 2026",
    )
    store = ContextStore()
    store.register(InitiativeSeed, seed)

    result = await execute(
        operation=operation,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        tool_registry=tool_registry,
    )

    proposal: InitiativeProposal = result
    print(f"Title: {proposal.title}")
    print(f"Summary: {proposal.executive_summary}")
    print(f"Risk: {proposal.risk_assessment}")


asyncio.run(main())

Next steps¤