My First Operation

This tutorial walks you through building and running your first CrewMaster v2.0.0 Operation — a simple greeter that produces a structured greeting for a user. You will learn how to define an agent identity, wire up an operation, resolve domain context, render prompt blocks, and execute everything with a FakeDriver that requires no LLM API key.

What you'll build

A greeter operation that takes a user's name from the ContextStore, renders a personalized greeting prompt using a BlockStore template, and returns a typed Greeting response through a FakeDriver.

Concepts covered

• AgentConfig — persistent agent identity
• Operation — typed, recursive unit of work
• ContextStore — domain object registry
• BlockStore / LocalDiskStore — prompt templates on disk
• PromptEngine — block composition and rendering
• FakeDriver — fake runtime for tests
• execute() — the entry point

---

1 Define the output model

Every operation in CrewMaster produces a typed output. We start by defining a Pydantic model for the greeting:

from pydantic import BaseModel


class Greeting(BaseModel):
    """A personalized greeting produced by the greeter agent."""

    message: str

---

2 Create an AgentConfig

An AgentConfig defines who the agent is. It holds identity prompt blocks (references to blocks:// URIs) and can optionally declare a default tool scope, context projectors, or a specific runtime driver.

from crewmaster.agents.agent import AgentConfig

greeter_agent = AgentConfig(
    name="greeter",
    identity_blocks=["blocks://greeter/identity"],
)

The identity_blocks list tells CrewMaster which prompt templates contribute to the agent's persona. The actual template files live in a BlockStore — we will create them in step 5.

---

3 Define the Operation

An Operation is the fundamental unit of work. It declares:

• name — a unique identifier within the plan
• produces — the Pydantic type the operation outputs
• kind — semantic category: artifact, cognition, or communication
• agent — the AgentConfig that executes this operation
• task_blocks — blocks:// URIs for task prompt templates

from crewmaster.operations.operation import Operation

operation = Operation(
    name="greet",
    produces=Greeting,
    kind="artifact",
    agent=greeter_agent,
    task_blocks=["blocks://greeter/task"],
)

---

4 Set up the ContextStore

The ContextStore holds domain objects that your operation may need at runtime. For this tutorial, we register a simple User object that will be projected into greeting templates:

from crewmaster.agents.context.store import ContextStore

class User:
    """Simple domain model for the tutorial."""
    def __init__(self, name: str) -> None:
        self.name = name

store = ContextStore()
store.register(User, User(name="Alice"))

Later, if you want the operation to resolve the user automatically through consumes, you would define a ContextProjector and add it to the operation. For this first tutorial we keep things minimal and pass the context as a variable into the template.

---

5 Set up the BlockStore

A BlockStore resolves blocks:// URIs to actual prompt templates. The default implementation is LocalDiskStore, which reads .j2 (Jinja2) files from a local directory.

Each .j2 file may include optional YAML frontmatter (delimited by ---) to declare metadata like id, kind, requires, and provides.

Create two block files — an identity block and a task block:

import tempfile
from pathlib import Path

from crewmaster.agents.prompts.local_disk_store import LocalDiskStore

# Create a temporary directory for our block files
blocks_dir = Path(tempfile.mkdtemp())

# Identity block — defines who the agent is
(blocks_dir / "greeter" / "identity.j2").mkdir(parents=True)
(blocks_dir / "greeter" / "identity.j2").write_text("""\
---
id: blocks://greeter/identity
kind: identity
provides: greeter_persona
---
You are a friendly greeting assistant.
""")

# Task block — defines what the agent should do
(blocks_dir / "greeter" / "task.j2").write_text("""\
---
id: blocks://greeter/task
kind: task
provides: greeting_output
---
Write a warm greeting for {{ user_name }}.
Include the user's name in the greeting.
""")

block_store = LocalDiskStore(root=str(blocks_dir))

The task template uses {{ user_name }} — a Jinja2 variable that will be resolved from the context dictionary when the PromptEngine renders the prompt.

---

6 Create a FakeDriver

A RuntimeDriver is the SPI that connects CrewMaster to an LLM. For development and testing, you can use a FakeDriver that returns canned responses without calling any LLM API:

from crewmaster.execution.runtime import RuntimeRequest, RuntimeResponse


class FakeDriver:
    """Fake driver that returns a pre-configured response."""

    def __init__(self, greeting: Greeting) -> None:
        self._greeting = greeting
        self.calls: list[RuntimeRequest] = []

    async def execute(self, request: RuntimeRequest) -> RuntimeResponse:
        self.calls.append(request)
        return RuntimeResponse(output=self._greeting)

    async def astream(self, request: RuntimeRequest):
        self.calls.append(request)
        from crewmaster.execution.runtime import RuntimeStreamChunk

        yield RuntimeStreamChunk(kind="text_delta", content="Thinking...")
        yield RuntimeStreamChunk(kind="final", output=self._greeting)

The FakeDriver must implement the RuntimeDriver protocol — an execute() method returning a RuntimeResponse, and an astream() generator yielding RuntimeStreamChunk events.

---

7 Execute the operation

Now wire everything together and call crewmaster.execute(). The PromptEngine renders the blocks, and the FakeDriver returns the canned response:

import asyncio

from crewmaster import execute
from crewmaster.agents.prompts.engine import PromptEngine


async def main() -> None:
    driver = FakeDriver(
        greeting=Greeting(message="Hello, Alice! Welcome aboard!")
    )

    result = await execute(
        operation=operation,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        prompt_engine=PromptEngine(block_store=block_store),
        # Pass user_name directly as a context override
        context_overrides={"user_name": "Alice"},
    )

    print(f"Result: {result.message}")

asyncio.run(main())

Expected output:

Result: Hello, Alice! Welcome aboard!

---

8 Complete script

Putting it all together, here is the full self-contained script:

my_first_operation.py

"""My First Operation — a minimal CrewMaster v2.0.0 tutorial."""
import asyncio
import tempfile
from pathlib import Path

from pydantic import BaseModel

from crewmaster import execute
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.engine import PromptEngine
from crewmaster.agents.prompts.local_disk_store import LocalDiskStore
from crewmaster.execution.runtime import (
    RuntimeRequest,
    RuntimeResponse,
    RuntimeStreamChunk,
)
from crewmaster.operations.operation import Operation


# 1. Output model
class Greeting(BaseModel):
    message: str


# 2. Agent identity
greeter_agent = AgentConfig(
    name="greeter",
    identity_blocks=["blocks://greeter/identity"],
)


# 3. Operation
operation = Operation(
    name="greet",
    produces=Greeting,
    kind="artifact",
    agent=greeter_agent,
    task_blocks=["blocks://greeter/task"],
)


# 4. Domain data
class User:
    def __init__(self, name: str) -> None:
        self.name = name


store = ContextStore()
store.register(User, User(name="Alice"))


# 5. Prompt blocks
blocks_dir = Path(tempfile.mkdtemp())

(blocks_dir / "greeter" / "identity.j2").mkdir(parents=True)
(blocks_dir / "greeter" / "identity.j2").write_text("""\
---
id: blocks://greeter/identity
kind: identity
provides: greeter_persona
---
You are a friendly greeting assistant.
""")

(blocks_dir / "greeter" / "task.j2").write_text("""\
---
id: blocks://greeter/task
kind: task
provides: greeting_output
---
Write a warm greeting for {{ user_name }}.
Include the user's name in the greeting.
""")

block_store = LocalDiskStore(root=str(blocks_dir))


# 6. Fake driver (no API key needed)
class FakeDriver:
    def __init__(self, greeting: Greeting) -> None:
        self._greeting = greeting
        self.calls: list[RuntimeRequest] = []

    async def execute(self, request: RuntimeRequest) -> RuntimeResponse:
        self.calls.append(request)
        return RuntimeResponse(output=self._greeting)

    async def astream(self, request: RuntimeRequest):
        self.calls.append(request)
        yield RuntimeStreamChunk(kind="text_delta", content="Thinking...")
        yield RuntimeStreamChunk(kind="final", output=self._greeting)


# 7. Execute
async def main() -> None:
    driver = FakeDriver(
        greeting=Greeting(message="Hello, Alice! Welcome aboard!")
    )

    result = await execute(
        operation=operation,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        prompt_engine=PromptEngine(block_store=block_store),
        context_overrides={"user_name": "Alice"},
    )

    print(f"Result: {result.message}")


asyncio.run(main())

Save the file and run it:

python my_first_operation.py

Result: Hello, Alice! Welcome aboard!

---

9 Using a real LLM

When you are ready to connect a real language model, swap the FakeDriver for a PydanticAIDriver:

from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver

driver = PydanticAIDriver(
    model="openai:gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)

Then pass it as the default_runtime:

result = await execute(
    operation=operation,
    context_store=store,
    default_runtime=driver,
    block_store=block_store,
    prompt_engine=PromptEngine(block_store=block_store),
    context_overrides={"user_name": "Alice"},
)

Environment variables

The PydanticAIDriver reads model configuration from environment variables:

• OPENAI_API_KEY — your OpenAI API key
• ANTHROPIC_API_KEY — for Claude models
• GOOGLE_API_KEY — for Gemini models

Set these in your shell or .env file before running with a real driver.

---

Next steps

Now that you have built your first operation, see the API Reference to explore the full execute() and execute_stream() API.