CrewMaster v2.0.0 — The Ultimate Framework for Building AI Teams 🚀

CrewMaster is a framework-agnostic Python library for building, orchestrating, and evaluating type-safe multi-agent systems. Define Operations, wire them into DAGs, and execute with structured, streaming output — without being locked into a single LLM framework.

Why CrewMaster v2.0.0?

v2.0.0 is a complete redesign built on lessons learned from v1. Every concept has been rethought for developer ergonomics, type safety, and framework independence.

Quick Example

import asyncio
from pydantic import BaseModel
from crewmaster import execute
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.store import ContextStore
from crewmaster.operations.operation import Operation
from crewmaster.execution.runtime import FakeDriver


class Greeting(BaseModel):
    message: str


# 1. Define an agent
greeter = AgentConfig(
    name="greeter",
    identity_blocks=["blocks://greeter/identity"],
)

# 2. Define an operation
say_hello = Operation(
    name="say_hello",
    produces=Greeting,
    kind="artifact",
    agent=greeter,
    task_blocks=["blocks://greeter/task"],
)

# 3. Execute with FakeDriver (no API key needed)
async def main():
    result = await execute(
        operation=say_hello,
        context_store=ContextStore(),
        default_runtime=FakeDriver(),
    )
    print(result.message)


asyncio.run(main())

Key Concepts

• Operation — Typed, recursive unit of work that declares produces/consumes
• AgentConfig — Persistent agent identity with prompt blocks, tool scope, context projections
• ExecutionPlan — DAG resolved automatically from Operation trees via resolve_plan()
• ContextStore — Type-keyed container of domain objects
• PromptEngine + BlockStore — Jinja2-based prompt composition from blocks:// URIs
• ToolRegistry — Scoped tool storage with semantic retrieval at runtime (request_capability)
• RuntimeDriver — Async protocol for LLM backends (PydanticAI, LangChain, custom)
• execute / execute_stream — Top-level entry points for blocking and streaming execution

Installation

pip install crewmaster

With extras

# PydanticAI runtime driver (recommended)
pip install "crewmaster[pydantic-ai]"

# Sandbox UI for interactive evaluation
pip install "crewmaster[sandbox-ui]"

Environment

# For PydanticAIDriver
export OPENAI_API_KEY=<your-api-key>

See the installation guide for full details.

Framework-Agnostic

Swap drivers without changing your code:

from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver

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

result = await execute(
    operation=my_op,
    default_runtime=driver,  # just swap the driver
    ...
)

Built-in Collaboration & Conversation

# Sequential multi-agent channel
from crewmaster.conversation.channel import channel_dispatch

# Human-agent dialogue with structured clarification
from crewmaster.conversation.dialogue import dialogue, AgentOutputClarification

# Grill-me: structured critique-and-revision cycles
from crewmaster.collaboration.protocol import GrillMeProtocol

Evaluation

CrewMaster v2.0.0 includes a complete evaluation framework:

• Snapshots — Version your operations with expected inputs/outputs
• Datasets — Define test data in YAML
• Criteria — Write evaluation criteria as Operation nodes producing Score objects
• Reports — Aggregate results with per-criterion averages and global scores

Documentation

Full documentation →

Development

# Install dev dependencies
poetry install

# Run tests (674+ and counting)
poetry run pytest

# Build docs
poetry run mkdocs build --strict

License

MIT — see LICENSE for details.