Core Concepts¤

CrewMaster v2.0.0 is built around eight fundamental concepts. Understanding them gives you a mental model of how the framework composes, resolves, and executes multi-agent systems.

1. Operation¤

An Operation is the fundamental unit of work in CrewMaster. It is a recursive, typed task that declares what it produces, what it consumes, and how it should be executed.

from pydantic import BaseModel
from crewmaster.operations.operation import Operation

class Greeting(BaseModel):
    message: str

say_hello = Operation(
    name="say_hello",
    produces=Greeting,
    kind="artifact",
    task_blocks=["blocks://tasks/hello.j2"],
    consumes=[],
)

Every Operation has a name, a produces type (its output contract), a kind ("artifact", "cognition", or "communication"), optional task_blocks for prompt URIs, and an optional consumes list of types it needs from upstream nodes. Operations can be nested via sub_operations, forming a tree that is resolved into an ExecutionPlan.

2. AgentConfig¤

An AgentConfig defines the persistent identity of an agent: who it is, what prompts define its personality, which tools it has access to, and which runtime driver to use.

from crewmaster.agents.agent import AgentConfig

analyst = AgentConfig(
    name="analyst",
    identity_blocks=["blocks://identity/analyst.j2"],
    default_tool_scope="data-analysis",
    default_context=[SomeDomainContext],
    runtime=None,  # uses default_runtime from execute()
)

If no runtime is assigned, the default_runtime passed to execute() or execute_stream() is used — making it easy to swap drivers without changing agent definitions.

3. ExecutionPlan¤

The ExecutionPlan is a directed acyclic graph (DAG) of PlanNode instances. It is built automatically from an Operation tree via resolve_plan(). Each node carries a resolved disclosure (agent identity, prompt blocks, tools, context), its upstream dependencies, and the assigned runtime driver.

The plan enforces topological ordering and passes each node's output as deps context to downstream nodes that consume that type.

4. ContextStore / ContextProjector¤

The ContextStore is a type-keyed container of domain objects. It holds all the data that agents need during execution. A ContextProjector defines how to project a slice of the ContextStore into the ctx variable available in prompt templates.

from crewmaster.agents.context.store import ContextStore

ctx = ContextStore()
ctx.add(MyDomainObject(some="data"))

At plan resolution time, each node's AgentConfig.default_context projectors are applied to extract the relevant subset of the store.

5. PromptEngine / BlockStore¤

The BlockStore stores PromptBlock instances keyed by blocks:// URIs. The PromptEngine composes multiple blocks and renders them with Jinja2 using the resolved execution context (ctx, deps, cfg).

from crewmaster.agents.prompts.block import BlockStore
from crewmaster.agents.prompts.engine import PromptEngine

store = BlockStore()
store.add("blocks://identity/analyst.j2", PromptBlock(
    template="You are an analyst. Context: {{ ctx.data }}."
))
engine = PromptEngine(store)

The engine resolves blocks:// URIs in agent identity_blocks and operation task_blocks, merges them, and renders the combined template against the current node's context dictionary.

6. ToolSchema / ToolRegistry¤

A ToolSchema defines a tool's signature: name, description, and JSON parameter schema. A Capability pairs a ToolSchema with an async handler and optional scope. The ToolRegistry indexes capabilities and supports semantic retrieval by description at runtime.

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

calc_schema = ToolSchema(
    name="calculator",
    description="Perform arithmetic calculations",
    parameters={"a": "float", "b": "float", "op": "str"},
)

async def calculate(a: float, b: float, op: str) -> dict:
    return {"result": eval(f"{a} {op} {b}")}

cap = Capability(name="calculator", tool_schema=calc_schema, fn=calculate)
registry = ToolRegistry()
registry.register(cap)

Agents can dynamically request capabilities at execution time through the request_capability system tool — the registry finds matches by semantic similarity and injects them into the agent's toolset.

7. RuntimeDriver¤

The RuntimeDriver is an async protocol that all LLM drivers must implement. It has two methods: execute() for blocking calls and astream() for streaming. CrewMaster ships with two implementations:

from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver

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

Each driver accepts a RuntimeRequest (instructions, tools, context, output_schema) and returns a RuntimeResponse (structured output, token usage, raw messages).

8. execute / execute_stream¤

execute() and execute_stream() are the top-level entry points. They resolve the Operation tree into an ExecutionPlan, then execute each node in topological order through its assigned RuntimeDriver.

from crewmaster import execute, execute_stream

# Blocking: returns the final artifact
result = await execute(
    operation=my_op,
    context_store=ctx,
    default_runtime=driver,
    block_store=blocks,
)

# Streaming: yields RuntimeStreamChunk events
async for chunk in execute_stream(
    operation=my_op,
    context_store=ctx,
    default_runtime=driver,
    block_store=blocks,
):
    print(chunk.kind, chunk)

Both functions support optional tool_registry for dynamic tool retrieval, prompt_engine for runtime prompt rendering, and observer for hooking into the execution lifecycle (evaluation, logging).

Putting It All Together¤

This pipeline is framework-agnostic: swap the driver, and the rest of your code stays the same. That's the core promise of CrewMaster v2.0.0.