Multi-Node Pipeline with DAG and Streaming¤

from pydantic import BaseModel, ConfigDict, Field


class BrandSeed(BaseModel):
    """The seed brand identity provided by the user."""
    model_config = ConfigDict(frozen=True)

    name: str = Field(description="Brand name")
    industry: str = Field(description="Industry sector")
    values: list[str] = Field(description="Core brand values")


class TypographyDesign(BaseModel):
    """Typography design output from the TypographyDesigner agent."""
    model_config = ConfigDict(frozen=True)

    primary_font: str = Field(description="Primary typeface name")
    secondary_font: str = Field(description="Secondary typeface name")
    font_weights: list[str] = Field(description="Font weights used")
    rationale: str = Field(description="Design rationale")


class ColorPalette(BaseModel):
    """Color palette output from the Colorist agent."""
    model_config = ConfigDict(frozen=True)

    primary_color: str = Field(description="Primary brand color (hex)")
    secondary_color: str = Field(description="Secondary brand color (hex)")
    accent_color: str = Field(description="Accent color (hex)")
    background_color: str = Field(description="Background color (hex)")
    rationale: str = Field(description="Color selection rationale")


class VisualIdentity(BaseModel):
    """Final visual identity output from the Designer agent."""
    model_config = ConfigDict(frozen=True)

    brand_name: str = Field(description="Brand name")
    typography: TypographyDesign = Field(description="Typography choices")
    colors: ColorPalette = Field(description="Color palette")
    identity_summary: str = Field(description="Synthesis of visual identity")

from typing import Any

from crewmaster.agents.context.projector import ContextProjector


class BrandSeedProjector(ContextProjector):
    """Projects BrandSeed into template context."""

    def __init__(self, brand_name: str, brand_industry: str,
                 brand_values: list[str]) -> None:
        self._brand_name = brand_name
        self._brand_industry = brand_industry
        self._brand_values = brand_values

    @classmethod
    def from_domain(cls, data: BrandSeed) -> "BrandSeedProjector":
        return cls(
            brand_name=data.name,
            brand_industry=data.industry,
            brand_values=list(data.values),
        )

    def to_context(self) -> dict[str, Any]:
        return {
            "brand_name": self._brand_name,
            "brand_industry": self._brand_industry,
            "brand_values": ", ".join(self._brand_values),
        }

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

FONT_DATABASE_TOOL = ToolSchema(
    name="font_database",
    description="Query the font database for typeface recommendations",
    input_schema={
        "type": "object",
        "properties": {
            "industry": {"type": "string"},
            "style": {"type": "string"},
        },
        "required": ["industry"],
    },
)

COLOR_HARMONY_TOOL = ToolSchema(
    name="color_harmony",
    description="Generate a harmonious color palette based on constraints",
    input_schema={
        "type": "object",
        "properties": {
            "base_hue": {"type": "string"},
            "mood": {"type": "string"},
        },
    },
)

DESIGN_VALIDATOR_TOOL = ToolSchema(
    name="design_validator",
    description="Validate visual identity for accessibility and contrast",
    input_schema={
        "type": "object",
        "properties": {
            "primary_color": {"type": "string"},
            "background_color": {"type": "string"},
            "primary_font": {"type": "string"},
        },
    },
)

tool_registry = ToolRegistry()
tool_registry.register("typography", FONT_DATABASE_TOOL)
tool_registry.register("color", COLOR_HARMONY_TOOL)
tool_registry.register("design", DESIGN_VALIDATOR_TOOL)

from crewmaster.agents.agent import AgentConfig

typographer = AgentConfig(
    name="typographer",
    identity_blocks=["blocks://pipeline/identity/typographer"],
    default_tool_scope="typography",
)

colorist = AgentConfig(
    name="colorist",
    identity_blocks=["blocks://pipeline/identity/colorist"],
    default_tool_scope="color",
)

designer = AgentConfig(
    name="designer",
    identity_blocks=["blocks://pipeline/identity/designer"],
    default_tool_scope="design",
)

import tempfile
from pathlib import Path

from crewmaster.agents.prompts.local_disk_store import LocalDiskStore

blocks_dir = Path(tempfile.mkdtemp())


def _write(path_parts, content):
    p = blocks_dir.joinpath(*path_parts)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content)


_write(("pipeline", "identity", "typographer.j2"), """\
---
id: blocks://pipeline/identity/typographer
kind: identity
provides: typographer_identity
---
You are a typography expert. Your role is to select fonts that reflect
the brand's personality.
""")

_write(("pipeline", "identity", "colorist.j2"), """\
---
id: blocks://pipeline/identity/colorist
kind: identity
provides: colorist_identity
---
You are a color expert. Create color palettes that evoke the right mood
and align with brand values.
""")

_write(("pipeline", "identity", "designer.j2"), """\
---
id: blocks://pipeline/identity/designer
kind: identity
provides: designer_identity
---
You are a visual identity designer. Synthesize typography and color into
a cohesive brand identity.
""")

_write(("pipeline", "task", "typographer.j2"), """\
---
id: blocks://pipeline/task/typographer
kind: task
provides: typography_design_output
---
Select a font pairing for brand **{{ ctx.brand_name }}** in the
**{{ ctx.brand_industry }}** sector. The brand values are:
**{{ ctx.brand_values }}**.

Output a TypographyDesign with primary_font, secondary_font,
font_weights, and rationale.
""")

_write(("pipeline", "task", "colorist.j2"), """\
---
id: blocks://pipeline/task/colorist
kind: task
provides: color_palette_output
---
Create a color palette for **{{ ctx.brand_name }}** that complements
the typography from the upstream TypographyDesigner.

Output a ColorPalette with primary_color, secondary_color, accent_color,
background_color (all hex), and rationale.
""")

_write(("pipeline", "task", "designer.j2"), """\
---
id: blocks://pipeline/task/designer
kind: task
provides: visual_identity_output
---
Synthesize the typography and color palette from upstream nodes into a
cohesive VisualIdentity for **{{ ctx.brand_name }}**.

Output a VisualIdentity with brand_name, typography (the TypographyDesign),
colors (the ColorPalette), and identity_summary.
""")

block_store = LocalDiskStore(root=str(blocks_dir))

from crewmaster.operations.operation import Operation

pipeline = Operation(
    name="brand_identity_pipeline",
    produces=VisualIdentity,
    kind="artifact",
    sub_operations=[
        Operation(
            name="TypographyDesigner",
            produces=TypographyDesign,
            kind="artifact",
            agent=typographer,
            task_blocks=["blocks://pipeline/task/typographer"],
            consumes=[BrandSeedProjector],
        ),
        Operation(
            name="Colorist",
            produces=ColorPalette,
            kind="artifact",
            agent=colorist,
            task_blocks=["blocks://pipeline/task/colorist"],
            consumes=[BrandSeedProjector, TypographyDesign],
        ),
        Operation(
            name="Designer",
            produces=VisualIdentity,
            kind="artifact",
            agent=designer,
            task_blocks=["blocks://pipeline/task/designer"],
            consumes=[BrandSeedProjector, TypographyDesign, ColorPalette],
        ),
    ],
)

import asyncio

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

# Register the BrandSeed domain object
brand = BrandSeed(
    name="NovaTech",
    industry="AI & Developer Tools",
    values=["innovation", "clarity", "empowerment"],
)
store = ContextStore()
store.register(BrandSeed, brand)


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

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

    async for chunk in execute_stream(
        operation=pipeline,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        tool_registry=tool_registry,
    ):
        if chunk.kind == "node_start":
            print(f"\n▶ Starting: {chunk.node_name}")
        elif chunk.kind == "text_delta":
            print(chunk.content, end="", flush=True)
        elif chunk.kind == "node_complete":
            print(f"\n✔ Completed: {chunk.node_name}")
            final = chunk.output
        elif chunk.kind == "final":
            print(f"\n🏁 Pipeline complete")
            break

    # final holds the VisualIdentity
    print(f"\nBrand: {final.brand_name}")
    print(f"Font:  {final.typography.primary_font} + {final.typography.secondary_font}")
    print(f"Colors: {final.colors.primary_color} / {final.colors.accent_color}")
    print(f"Summary: {final.identity_summary}")


asyncio.run(main())

from fastapi import FastAPI
from fastapi.responses import StreamingResponse


app = FastAPI()


@app.get("/pipeline/stream")
async def stream_pipeline():
    async def event_generator():
        async for chunk in execute_stream(
            operation=pipeline,
            context_store=store,
            default_runtime=driver,
            block_store=block_store,
            tool_registry=tool_registry,
        ):
            yield f"data: {chunk.model_dump_json()}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
    )