agents

Agent models for CrewMaster v2.0.0.

MODULE	DESCRIPTION
agent	Agent configuration model for CrewMaster v2.0.0.
agent_test	Tests for AgentConfig model.
context	Agent context models for CrewMaster v2.0.0.
prompts	Prompt blocks, block storage, and prompt engine for CrewMaster v2.0.0.

CLASS	DESCRIPTION
AgentConfig	Persistent configuration for an agent identity.

__all__ module-attribute

__all__ = ['AgentConfig']

AgentConfig

Persistent configuration for an agent identity.

AgentConfig defines who an agent is — its identity prompts, default tool scope, context projections, and optionally a specific runtime driver. If no runtime is specified, the default_runtime from execute() is used.

ATTRIBUTE	DESCRIPTION
name	Unique agent name (used as node identifier in plans). TYPE: str
identity_blocks	List of blocks:// URIs for identity prompt blocks. TYPE: list[str]
runtime	Optional driver instance. If None, uses the default runtime. TYPE: Any | None
default_tool_scope	Optional scope string for tool registry queries. TYPE: str | None
default_context	List of context projector types to resolve at runtime. TYPE: list[type[Any]]

METHOD	DESCRIPTION
validate_block_uris	Validate that identity blocks are blocks:// URIs.

name instance-attribute

name: str

identity_blocks class-attribute instance-attribute

identity_blocks: list[str] = Field(default_factory=list)

runtime class-attribute instance-attribute

runtime: Any | None = Field(default=None, exclude=True)

default_tool_scope class-attribute instance-attribute

default_tool_scope: str | None = None

default_context class-attribute instance-attribute

default_context: list[type[Any]] = Field(default_factory=list)

model_config class-attribute instance-attribute

model_config = {'arbitrary_types_allowed': True}

validate_block_uris classmethod

validate_block_uris(v: list[str]) -> list[str]

Validate that identity blocks are blocks:// URIs.

context

Agent context models for CrewMaster v2.0.0.

MODULE	DESCRIPTION
projector	Context projection protocol for CrewMaster v2.0.0.
projector_test	Tests for ContextProjector protocol.
store	ContextStore and operation context resolution for CrewMaster v2.0.0.
store_test	Tests for ContextStore and resolve_operation_context.

CLASS	DESCRIPTION
ContextProjector	Protocol for projecting domain objects into LLM context.
ContextStore	Registry of domain objects for context projection.
ContextStoreError	Raised when context resolution fails.

FUNCTION	DESCRIPTION
resolve_operation_context	Resolve all context for an Operation's consumes list.

__all__ module-attribute

__all__ = ['ContextProjector', 'ContextStore', 'ContextStoreError', 'resolve_operation_context']

ContextProjector

Protocol for projecting domain objects into LLM context.

Implementations must provide: - from_domain: A classmethod that accepts a domain object and returns a configured ContextProjector instance. - to_context: Returns a flat dict of context variables for the PromptEngine's template rendering.

Usage::

class UserProjector:
    @classmethod
    def from_domain(cls, user: User) -> "UserProjector":
        return cls(name=user.name, role=user.role)

    def to_context(self) -> dict[str, Any]:
        return {"name": self.name, "role": self.role}

METHOD	DESCRIPTION
from_domain	Construct a projector from a domain object.
to_context	Project the domain data into a flat context dictionary.

from_domain classmethod

from_domain(data: TDomain) -> 'ContextProjector[TDomain]'

Construct a projector from a domain object.

PARAMETER	DESCRIPTION
data	The domain object to project. TYPE: TDomain

RETURNS	DESCRIPTION
'ContextProjector[TDomain]'	A configured ContextProjector instance.

to_context

to_context() -> dict[str, Any]

Project the domain data into a flat context dictionary.

RETURNS	DESCRIPTION
dict[str, Any]	A dictionary of key-value pairs suitable for Jinja2 rendering.

ContextStore

ContextStore()

Registry of domain objects for context projection.

The ContextStore is owned and populated by the application. CrewMaster reads from it at plan-build time to resolve Operation.consumes into flat context dictionaries suitable for template rendering.

Usage::

store = ContextStore()
store.register(User, User(name="Alice", role="analyst"))
ctx = store.resolve(UserProjector)  # -> {"user_name": "Alice", ...}

METHOD	DESCRIPTION
register	Register a domain object for context resolution.
resolve	Resolve context by projecting a registered domain object.

register

register(domain_type: Type[Any], instance: Any) -> None

Register a domain object for context resolution.

PARAMETER	DESCRIPTION
domain_type	The Python type serving as the registration key. TYPE: Type[Any]
instance	The domain object instance. TYPE: Any

resolve

resolve(projector_type: Type[Any]) -> dict[str, Any]

Resolve context by projecting a registered domain object.

Infers TDomain from the projector type, looks up the registered domain instance, and projects it via from_domain → to_context.

Domain type inference tries (in order):

1. Generic base ContextProjector[TDomain] via __orig_bases__.
2. Type annotation of the data parameter of from_domain.

PARAMETER	DESCRIPTION
projector_type	A type implementing the ContextProjector protocol. TYPE: Type[Any]

RETURNS	DESCRIPTION
dict[str, Any]	A flat dictionary of key-value pairs for the PromptEngine.

RAISES	DESCRIPTION
ContextStoreError	If the projector is not a ContextProjector, if the domain type cannot be inferred, or if no instance is registered for the inferred domain type.

ContextStoreError

Raised when context resolution fails.

resolve_operation_context

resolve_operation_context(operation: Operation, store: ContextStore) -> dict[str, dict[str, Any]]

Resolve all context for an Operation's consumes list.

Iterates over the projector types declared in operation.consumes and resolves each one against the store. The returned dictionary is keyed by the projector type's class name.

PARAMETER	DESCRIPTION
operation	The Operation whose consumes need resolving. TYPE: Operation
store	The ContextStore holding registered domain instances. TYPE: ContextStore

RETURNS	DESCRIPTION
dict[str, dict[str, Any]]	A dictionary mapping projector class names to their projected
dict[str, dict[str, Any]]	context dicts. Example: {"UserProjector": {"name": "Bob"}}.

RAISES	DESCRIPTION
ContextStoreError	If any projector in consumes cannot be resolved, either because it is not a ContextProjector or because its domain type is not registered.

prompts

Prompt blocks, block storage, and prompt engine for CrewMaster v2.0.0.

MODULE	DESCRIPTION
block	Prompt blocks and block storage protocol for CrewMaster v2.0.0.
block_test	Tests for PromptBlock and BlockStore protocol.
engine	PromptEngine for block composition, rendering, and validation.
engine_test	Tests for PromptEngine.
local_disk_store	Local disk implementation of BlockStore protocol.
local_disk_store_test	Tests for LocalDiskStore.

CLASS	DESCRIPTION
BlockStore	Protocol for resolving block URIs to PromptBlock instances.
PromptBlock	A discrete prompt block with YAML frontmatter metadata.
PromptEngine	Composes, renders, and validates prompt blocks.
LocalDiskStore	BlockStore implementation that reads .j2 files from a local directory.

__all__ module-attribute

__all__ = ['PromptBlock', 'BlockStore', 'LocalDiskStore', 'PromptEngine']

BlockStore

Protocol for resolving block URIs to PromptBlock instances.

Implementations are responsible for loading template files from their storage medium (local disk, remote, database) and returning parsed PromptBlock instances with frontmatter extracted.

The default implementation is LocalDiskStore, which reads .j2 files from a configured directory.

METHOD	DESCRIPTION
resolve	Resolve a block URI to a fully populated PromptBlock.

resolve

resolve(uri: str) -> PromptBlock

Resolve a block URI to a fully populated PromptBlock.

PARAMETER	DESCRIPTION
uri	A block URI in the form blocks://path/to/block. TYPE: str

RETURNS	DESCRIPTION
PromptBlock	A PromptBlock with frontmatter parsed and content loaded.

RAISES	DESCRIPTION
FileNotFoundError	If the URI cannot be resolved.
ValueError	If the frontmatter is malformed.

PromptBlock

A discrete prompt block with YAML frontmatter metadata.

PromptBlocks are the atomic units of the PromptEngine composition system. Each block declares what context variables it requires and what it provides, enabling compile-time validation of prompt assembly.

ATTRIBUTE	DESCRIPTION
id	Unique block identifier (e.g., "blocks://agent/identity"). TYPE: str
kind	Block kind for semantic categorization (e.g., "identity", "task", "tool"). TYPE: str
requires	List of context variable names this block needs. TYPE: list[str]
provides	The context variable name this block produces (None if terminal). TYPE: str | None
content	The raw Jinja2 template content of the block. TYPE: str

id instance-attribute

id: str

kind class-attribute instance-attribute

kind: str = ''

requires class-attribute instance-attribute

requires: list[str] = Field(default_factory=list)

provides class-attribute instance-attribute

provides: str | None = None

content class-attribute instance-attribute

content: str = ''

PromptEngine

PromptEngine(block_store: BlockStore)

Composes, renders, and validates prompt blocks.

The PromptEngine resolves block URIs through a :class:BlockStore, renders their Jinja2 content with the provided variables, and validates frontmatter constraints at composition time.

PARAMETER	DESCRIPTION
block_store	The BlockStore implementation to use for resolving URIs. TYPE: BlockStore

METHOD	DESCRIPTION
compose	Resolve blocks and render them into a single prompt string.
validate_blocks	Validate that all requires declared in blocks are satisfied.
validate_slots	Validate that all required slots are covered by block provides.

compose

compose(block_uris: list[str], variables: dict[str, Any]) -> str

Resolve blocks and render them into a single prompt string.

Blocks are resolved via the configured BlockStore, then each block's Jinja2 content is rendered with variables. Rendered blocks are joined with double newlines.

PARAMETER	DESCRIPTION
block_uris	List of blocks:// URIs to resolve. TYPE: list[str]
variables	Dictionary of template variables with namespaces (ctx, deps, cfg). TYPE: dict[str, Any]

RETURNS	DESCRIPTION
str	The fully rendered prompt string.

RAISES	DESCRIPTION
FileNotFoundError	If a URI cannot be resolved.
ValueError	If a template references an undefined variable.

validate_blocks

validate_blocks(block_uris: list[str], variables: dict[str, Any]) -> None

Validate that all requires declared in blocks are satisfied.

Checks every block's requires against the available variables. A require like ctx.name is satisfied if variables["ctx"]["name"] exists. Namespaces other than ctx, deps, cfg are rejected.

PARAMETER	DESCRIPTION
block_uris	Block URIs to validate. TYPE: list[str]
variables	Available variables by namespace. TYPE: dict[str, Any]

RAISES	DESCRIPTION
ValueError	If any require is unsatisfied or uses an unknown namespace. The message lists all missing variables.

validate_slots

validate_slots(block_uris: list[str], required_slots: list[str]) -> None

Validate that all required slots are covered by block provides.

Checks that every entry in required_slots appears as provides in at least one resolved block. If required_slots is empty, no validation is performed. Duplicate provides across blocks emits a warning but is not an error.

PARAMETER	DESCRIPTION
block_uris	Block URIs to validate. TYPE: list[str]
required_slots	Slots that must be provided. TYPE: list[str]

RAISES	DESCRIPTION
ValueError	If any required slot has no providing block.

LocalDiskStore

LocalDiskStore(root: str = '.')

BlockStore implementation that reads .j2 files from a local directory.

Resolves blocks:// URIs by stripping the blocks:// prefix and mapping the remainder into the configured root directory. For example, blocks://agent/colors/methodology resolves to {root}/agent/colors/methodology.j2.

Each .j2 file may contain optional YAML frontmatter delimited by --- at the start of the file, followed by another --- line. The frontmatter block is parsed as YAML to extract id, kind, requires, provides, and version. The remainder of the file is treated as the Jinja2 template content.

PARAMETER	DESCRIPTION
root	Root directory from which block files are resolved. TYPE: str DEFAULT: '.'

METHOD	DESCRIPTION
resolve	Resolve a blocks:// URI to a PromptBlock.

resolve

resolve(uri: str) -> PromptBlock

Resolve a blocks:// URI to a PromptBlock.

PARAMETER	DESCRIPTION
uri	URI in the form blocks://path/to/block. TYPE: str

RETURNS	DESCRIPTION
PromptBlock	PromptBlock with parsed frontmatter and content.

RAISES	DESCRIPTION
FileNotFoundError	If the file does not exist.