Skip to content

pyagent-blueprint API Reference

pyagent_blueprint

PyAgent Blueprint — declarative YAML specs for multi-agent LLM systems.

BlueprintCompiler

Compile a BlueprintSpec into a RuntimeGraph.

Steps: 1. Resolve provider bindings → LLMCallable instances 2. Instantiate agents with resolved LLMs 3. Look up pattern class from registry 4. Wire agents into pattern constructor kwargs 5. Return RuntimeGraph

Parameters:

Name Type Description Default
provider_registry Any

Optional ProviderRegistry for real providers. If None, uses MockLLM for all providers.

 None
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/compiler.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class BlueprintCompiler:
    """Compile a BlueprintSpec into a RuntimeGraph.

    Steps:
    1. Resolve provider bindings → LLMCallable instances
    2. Instantiate agents with resolved LLMs
    3. Look up pattern class from registry
    4. Wire agents into pattern constructor kwargs
    5. Return RuntimeGraph

    Args:
        provider_registry: Optional ``ProviderRegistry`` for real providers.
            If ``None``, uses ``MockLLM`` for all providers.
    """

    def __init__(self, provider_registry: Any = None) -> None:
        self._provider_registry = provider_registry

    def compile(self, spec: BlueprintSpec) -> RuntimeGraph:
        """Compile a blueprint spec into a runnable RuntimeGraph.

        Args:
            spec: Validated ``BlueprintSpec``.

        Returns:
            ``RuntimeGraph`` ready to execute.

        Raises:
            CompilationError: If the spec references unknown patterns or agents.
        """
        # Step 1: Resolve providers → LLMCallable
        llm_map = self._resolve_providers(spec)

        # Step 2: Instantiate agents
        agent_map = self._build_agents(spec, llm_map)

        # Step 3 & 4: Wire patterns
        workflows = self._build_workflows(spec, agent_map)

        metadata = spec.metadata.model_dump() if spec.metadata else {}
        graph = RuntimeGraph(workflows=workflows, agents=agent_map, metadata=metadata)

        # Step 5: Emit warnings for declared-but-unwired config
        self._warn_unwired(spec)

        return graph

    def _resolve_providers(self, spec: BlueprintSpec) -> dict[str, Any]:
        """Map provider names to LLMCallable instances."""
        llm_map: dict[str, Any] = {}

        for name, binding in spec.providers.items():
            if self._provider_registry is not None:
                provider = self._provider_registry.get(binding.provider)
                if provider is not None:
                    llm_map[name] = provider
                    continue

            # Fallback: MockLLM
            llm_map[name] = MockLLM(responses=[f"[{binding.model}] Mock response"])
            logger.debug("Using MockLLM for provider '%s' (model=%s)", name, binding.model)

        return llm_map

    def _build_agents(
        self,
        spec: BlueprintSpec,
        llm_map: dict[str, Any],
    ) -> dict[str, Agent]:
        """Instantiate Agent objects from spec."""
        agent_map: dict[str, Agent] = {}

        for name, agent_spec in spec.agents.items():
            llm = llm_map.get(agent_spec.provider, MockLLM(responses=["Mock response"]))
            agent_map[name] = Agent(
                name=name,
                llm=llm,
                system_prompt=agent_spec.prompt,
            )

        return agent_map

    def _build_workflows(
        self,
        spec: BlueprintSpec,
        agent_map: dict[str, Agent],
    ) -> dict[str, Any]:
        """Look up pattern classes and wire agents."""

        workflows: dict[str, Any] = {}

        for wf_name, wf_spec in spec.workflows.items():
            pattern_cls = get_pattern_class(wf_spec.pattern)
            if pattern_cls is None:
                raise CompilationError(
                    f"Unknown pattern '{wf_spec.pattern}' in workflow '{wf_name}'"
                )

            # Resolve agent refs
            wired_agents = self._resolve_agent_refs(wf_spec.agents, agent_map, wf_name)

            # Try to instantiate the pattern
            try:
                pattern = self._instantiate_pattern(
                    pattern_cls, wf_spec.pattern, wired_agents, wf_spec.config
                )
            except Exception as exc:
                raise CompilationError(
                    f"Failed to instantiate pattern '{wf_spec.pattern}' "
                    f"for workflow '{wf_name}': {exc}"
                ) from exc

            workflows[wf_name] = pattern

        return workflows

    def _resolve_agent_refs(
        self,
        agents_config: dict[str, Any],
        agent_map: dict[str, Agent],
        wf_name: str,
    ) -> dict[str, Any]:
        """Resolve agent name references to Agent objects."""
        resolved: dict[str, Any] = {}
        for role, ref in agents_config.items():
            if isinstance(ref, str):
                if ref not in agent_map:
                    raise CompilationError(
                        f"Agent ref '{ref}' in workflow '{wf_name}' not found. "
                        f"Available: {list(agent_map.keys())}"
                    )
                resolved[role] = agent_map[ref]
            elif isinstance(ref, dict):
                # Nested refs (e.g., routes: {billing: billing_agent})
                resolved[role] = {k: agent_map.get(v, v) for k, v in ref.items()}
            else:
                resolved[role] = ref
        return resolved

    @staticmethod
    def _instantiate_pattern(
        pattern_cls: type,
        pattern_name: str,
        agents: dict[str, Any],
        config: dict[str, Any],
    ) -> Any:
        """Instantiate a pattern from resolved agents and config.

        Handles the common pattern constructor signatures.
        """
        # Pipeline: expects `stages` list
        if pattern_name == "pipeline":
            stages = agents.get("stages", list(agents.values()))
            if isinstance(stages, dict):
                stages = list(stages.values())
            return pattern_cls(stages=stages, **config)

        # Supervisor: expects `classifier` + `routes`
        if pattern_name == "supervisor":
            return pattern_cls(
                classifier=agents.get("classifier"),
                routes=agents.get("routes", {}),
                **config,
            )

        # Fan-out/fan-in: expects `agents` list
        if pattern_name in ("fan_out_fan_in", "voting", "debate"):
            agent_list = list(agents.values())
            if len(agent_list) == 1 and isinstance(agent_list[0], dict):
                agent_list = list(agent_list[0].values())
            return pattern_cls(agents=agent_list, **config)

        # Default: pass agents dict as kwargs
        try:
            return pattern_cls(**agents, **config)
        except TypeError:
            # Try with agents as a list
            return pattern_cls(agents=list(agents.values()), **config)

    @staticmethod
    def _warn_unwired(spec: BlueprintSpec) -> None:
        """Emit warnings for blueprint config that requires manual wiring."""
        if spec.observability and spec.observability.tracing.enabled:
            logger.warning(
                "Blueprint declares observability.tracing.enabled=True but no trace_bus "
                "was wired. Call graph.wire_trace(bus) on the compiled RuntimeGraph."
            )
        if spec.observability and spec.observability.cost_budget is not None:
            logger.warning(
                "Blueprint declares observability.cost_budget but cost tracking "
                "must be manually wired via graph.wire_cost_tracker(tracker)."
            )
        if spec.context and spec.context.compression.policy != "none":
            logger.warning(
                "Blueprint declares context.compression.policy='%s' but compression "
                "must be manually wired via graph.wire_compressor(compressor).",
                spec.context.compression.policy,
            )
        if spec.context and spec.context.memory.semantic_enabled:
            logger.warning(
                "Blueprint declares context.memory.semantic_enabled=True but memory "
                "must be manually wired via graph.wire_context(ledger)."
            )
        if spec.context and spec.context.redaction is not None:
            logger.warning(
                "Blueprint declares context.redaction but redaction "
                "must be applied manually before sending messages to agents."
            )

compile(spec)

Compile a blueprint spec into a runnable RuntimeGraph.

Parameters:

Name Type Description Default
spec BlueprintSpec

Validated BlueprintSpec.

 required

Returns:

Type Description
RuntimeGraph

RuntimeGraph ready to execute.

Raises:

Type Description
CompilationError

If the spec references unknown patterns or agents.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/compiler.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def compile(self, spec: BlueprintSpec) -> RuntimeGraph:
    """Compile a blueprint spec into a runnable RuntimeGraph.

    Args:
        spec: Validated ``BlueprintSpec``.

    Returns:
        ``RuntimeGraph`` ready to execute.

    Raises:
        CompilationError: If the spec references unknown patterns or agents.
    """
    # Step 1: Resolve providers → LLMCallable
    llm_map = self._resolve_providers(spec)

    # Step 2: Instantiate agents
    agent_map = self._build_agents(spec, llm_map)

    # Step 3 & 4: Wire patterns
    workflows = self._build_workflows(spec, agent_map)

    metadata = spec.metadata.model_dump() if spec.metadata else {}
    graph = RuntimeGraph(workflows=workflows, agents=agent_map, metadata=metadata)

    # Step 5: Emit warnings for declared-but-unwired config
    self._warn_unwired(spec)

    return graph

CompilationError

Bases: Exception

Raised when a blueprint cannot be compiled.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/compiler.py 
19
20
class CompilationError(Exception):
    """Raised when a blueprint cannot be compiled."""

BlueprintDiffer

Compute semantic diff between two blueprint specs.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/differ.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class BlueprintDiffer:
    """Compute semantic diff between two blueprint specs."""

    def diff(self, old: BlueprintSpec, new: BlueprintSpec) -> list[Change]:
        """Diff two blueprint specs.

        Args:
            old: Previous version.
            new: Updated version.

        Returns:
            List of ``Change`` objects.
        """
        changes: list[Change] = []

        old_dict = old.model_dump()
        new_dict = new.model_dump()

        self._diff_dicts(old_dict, new_dict, "", changes)
        return changes

    def summary(self, changes: list[Change]) -> str:
        """Human-readable summary of changes.

        Args:
            changes: List from ``diff()``.

        Returns:
            Multi-line summary string.
        """
        if not changes:
            return "No changes detected."

        lines: list[str] = []
        for severity in (ChangeSeverity.BREAKING, ChangeSeverity.WARNING, ChangeSeverity.INFO):
            filtered = [c for c in changes if c.severity == severity]
            if filtered:
                lines.append(f"\n{severity.upper()} ({len(filtered)}):")
                for c in filtered:
                    lines.append(f"  [{c.change_type}] {c.path}")
        return "\n".join(lines)

    def _diff_dicts(
        self,
        old: dict,
        new: dict,
        prefix: str,
        changes: list[Change],
    ) -> None:
        """Recursively diff two dictionaries."""
        all_keys = set(old.keys()) | set(new.keys())

        for key in sorted(all_keys):
            path = f"{prefix}.{key}" if prefix else key
            old_val = old.get(key)
            new_val = new.get(key)

            if key not in old:
                changes.append(
                    Change(
                        path=path,
                        change_type=ChangeType.ADDED,
                        old_value=None,
                        new_value=new_val,
                        severity=self._classify_severity(key, ChangeType.ADDED),
                    )
                )
            elif key not in new:
                changes.append(
                    Change(
                        path=path,
                        change_type=ChangeType.REMOVED,
                        old_value=old_val,
                        new_value=None,
                        severity=self._classify_severity(key, ChangeType.REMOVED),
                    )
                )
            elif isinstance(old_val, dict) and isinstance(new_val, dict):
                self._diff_dicts(old_val, new_val, path, changes)
            elif old_val != new_val:
                changes.append(
                    Change(
                        path=path,
                        change_type=ChangeType.MODIFIED,
                        old_value=old_val,
                        new_value=new_val,
                        severity=self._classify_severity(key, ChangeType.MODIFIED),
                    )
                )

    @staticmethod
    def _classify_severity(field_name: str, change_type: ChangeType) -> ChangeSeverity:
        """Determine severity based on field name and change type."""
        if field_name in _BREAKING_FIELDS:
            return ChangeSeverity.BREAKING
        if change_type == ChangeType.REMOVED:
            return ChangeSeverity.WARNING
        if field_name in _WARNING_FIELDS:
            return ChangeSeverity.WARNING
        return ChangeSeverity.INFO

diff(old, new)

Diff two blueprint specs.

Parameters:

Name Type Description Default
old BlueprintSpec

Previous version.

 required
new BlueprintSpec

Updated version.

 required

Returns:

Type Description
list[Change]

List of Change objects.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/differ.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def diff(self, old: BlueprintSpec, new: BlueprintSpec) -> list[Change]:
    """Diff two blueprint specs.

    Args:
        old: Previous version.
        new: Updated version.

    Returns:
        List of ``Change`` objects.
    """
    changes: list[Change] = []

    old_dict = old.model_dump()
    new_dict = new.model_dump()

    self._diff_dicts(old_dict, new_dict, "", changes)
    return changes

summary(changes)

Human-readable summary of changes.

Parameters:

Name Type Description Default
changes list[Change]

List from diff().

 required

Returns:

Type Description
str

Multi-line summary string.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/differ.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def summary(self, changes: list[Change]) -> str:
    """Human-readable summary of changes.

    Args:
        changes: List from ``diff()``.

    Returns:
        Multi-line summary string.
    """
    if not changes:
        return "No changes detected."

    lines: list[str] = []
    for severity in (ChangeSeverity.BREAKING, ChangeSeverity.WARNING, ChangeSeverity.INFO):
        filtered = [c for c in changes if c.severity == severity]
        if filtered:
            lines.append(f"\n{severity.upper()} ({len(filtered)}):")
            for c in filtered:
                lines.append(f"  [{c.change_type}] {c.path}")
    return "\n".join(lines)

Change dataclass

A single semantic change between two blueprint versions.

Attributes:

Name Type Description
path str

Dotted path to the changed field.
change_type ChangeType

Added, removed, or modified.
old_value Any

Previous value (None for added).
new_value Any

New value (None for removed).
severity ChangeSeverity

Impact level.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/differ.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass(frozen=True)
class Change:
    """A single semantic change between two blueprint versions.

    Attributes:
        path: Dotted path to the changed field.
        change_type: Added, removed, or modified.
        old_value: Previous value (None for added).
        new_value: New value (None for removed).
        severity: Impact level.
    """

    path: str
    change_type: ChangeType
    old_value: Any
    new_value: Any
    severity: ChangeSeverity

BlueprintGenerator

Generate scaffold blueprint YAML from a pattern name and agent list.

Parameters:

Name Type Description Default
default_provider_model str

Model to use in the default provider binding.

 'gpt-4.1-mini'
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/generator.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class BlueprintGenerator:
    """Generate scaffold blueprint YAML from a pattern name and agent list.

    Args:
        default_provider_model: Model to use in the default provider binding.
    """

    def __init__(self, default_provider_model: str = "gpt-4.1-mini") -> None:
        self._default_model = default_provider_model

    def generate(
        self,
        pattern: str,
        agents: list[str],
        *,
        name: str = "my-blueprint",
        version: str = "0.1.0",
        description: str = "",
    ) -> str:
        """Generate a blueprint YAML string.

        Args:
            pattern: Pattern registry name (e.g., ``"supervisor"``, ``"pipeline"``).
            agents: List of agent names.
            name: Blueprint name.
            version: Blueprint version.
            description: Blueprint description.

        Returns:
            YAML string.

        Raises:
            ValueError: If the pattern is not registered.
        """
        if get_pattern_class(pattern) is None:
            available = list_patterns()
            raise ValueError(f"Unknown pattern '{pattern}'. Available: {available}")

        spec: dict = {
            "api_version": "pyagent/v1",
            "metadata": {
                "name": name,
                "version": version,
                "description": description or f"A {pattern} blueprint",
            },
            "providers": {
                "primary": {"model": self._default_model},
            },
            "agents": {},
            "workflows": {},
        }

        # Generate agent specs
        for agent_name in agents:
            spec["agents"][agent_name] = {
                "prompt": f"You are the {agent_name} agent. TODO: add your prompt here.",
                "provider": "primary",
            }

        # Generate workflow spec
        wf_agents = self._wire_pattern_agents(pattern, agents)
        spec["workflows"]["main"] = {
            "pattern": pattern,
            "agents": wf_agents,
        }

        return yaml.dump(spec, default_flow_style=False, sort_keys=False)

    @staticmethod
    def _wire_pattern_agents(pattern: str, agents: list[str]) -> dict:
        """Create the agents mapping for a workflow based on pattern type."""
        if pattern == "supervisor" and len(agents) >= 2:
            return {
                "classifier": agents[0],
                "routes": {name: name for name in agents[1:]},
            }

        if pattern == "pipeline":
            return {"stages": {name: name for name in agents}}

        if pattern in ("fan_out_fan_in", "voting", "debate"):
            return {"agents": {name: name for name in agents}}

        if pattern in ("self_reflection", "evaluator_optimizer") and len(agents) >= 2:
            return {
                "generator": agents[0],
                "evaluator": agents[1],
            }

        if pattern == "cross_reflection" and len(agents) >= 2:
            return {
                "agent_a": agents[0],
                "agent_b": agents[1],
            }

        # Default: pass all agents
        return {name: name for name in agents}

generate(pattern, agents, *, name='my-blueprint', version='0.1.0', description='')

Generate a blueprint YAML string.

Parameters:

Name Type Description Default
pattern str

Pattern registry name (e.g., "supervisor", "pipeline").

 required
agents list[str]

List of agent names.

 required
name str

Blueprint name.

 'my-blueprint'
version str

Blueprint version.

 '0.1.0'
description str

Blueprint description.

 ''

Returns:

Type Description
str

YAML string.

Raises:

Type Description
ValueError

If the pattern is not registered.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/generator.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def generate(
    self,
    pattern: str,
    agents: list[str],
    *,
    name: str = "my-blueprint",
    version: str = "0.1.0",
    description: str = "",
) -> str:
    """Generate a blueprint YAML string.

    Args:
        pattern: Pattern registry name (e.g., ``"supervisor"``, ``"pipeline"``).
        agents: List of agent names.
        name: Blueprint name.
        version: Blueprint version.
        description: Blueprint description.

    Returns:
        YAML string.

    Raises:
        ValueError: If the pattern is not registered.
    """
    if get_pattern_class(pattern) is None:
        available = list_patterns()
        raise ValueError(f"Unknown pattern '{pattern}'. Available: {available}")

    spec: dict = {
        "api_version": "pyagent/v1",
        "metadata": {
            "name": name,
            "version": version,
            "description": description or f"A {pattern} blueprint",
        },
        "providers": {
            "primary": {"model": self._default_model},
        },
        "agents": {},
        "workflows": {},
    }

    # Generate agent specs
    for agent_name in agents:
        spec["agents"][agent_name] = {
            "prompt": f"You are the {agent_name} agent. TODO: add your prompt here.",
            "provider": "primary",
        }

    # Generate workflow spec
    wf_agents = self._wire_pattern_agents(pattern, agents)
    spec["workflows"]["main"] = {
        "pattern": pattern,
        "agents": wf_agents,
    }

    return yaml.dump(spec, default_flow_style=False, sort_keys=False)

BlueprintLoadError

Bases: Exception

Raised when a blueprint cannot be loaded or validated.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/loader.py 
14
15
class BlueprintLoadError(Exception):
    """Raised when a blueprint cannot be loaded or validated."""

BlueprintRenderer

Render a BlueprintSpec as Mermaid diagrams or Markdown docs.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/renderer.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class BlueprintRenderer:
    """Render a BlueprintSpec as Mermaid diagrams or Markdown docs."""

    def to_mermaid(self, spec: BlueprintSpec) -> str:
        """Render the blueprint as a Mermaid flowchart.

        Args:
            spec: Blueprint specification.

        Returns:
            Mermaid diagram string.
        """
        lines = ["graph TD"]

        # Agent nodes
        for name, agent_spec in spec.agents.items():
            label = agent_spec.description or name
            lines.append(f"    {name}[{label}]")

        # Workflow edges
        for _, wf_spec in spec.workflows.items():
            agent_refs = self._extract_agent_refs(wf_spec.agents)
            if len(agent_refs) > 1:
                for i in range(len(agent_refs) - 1):
                    lines.append(f"    {agent_refs[i]} -->|{wf_spec.pattern}| {agent_refs[i + 1]}")

            # For supervisor-like patterns, show classifier → routes
            if "classifier" in wf_spec.agents and "routes" in wf_spec.agents:
                classifier = wf_spec.agents["classifier"]
                routes = wf_spec.agents.get("routes", {})
                if isinstance(routes, dict):
                    for route_name, route_ref in routes.items():
                        lines.append(f"    {classifier} -->|{route_name}| {route_ref}")

        return "\n".join(lines)

    def to_markdown(self, spec: BlueprintSpec) -> str:
        """Render the blueprint as Markdown documentation.

        Args:
            spec: Blueprint specification.

        Returns:
            Markdown string.
        """
        sections: list[str] = []

        # Title
        sections.append(f"# {spec.metadata.name}")
        if spec.metadata.description:
            sections.append(f"\n{spec.metadata.description}")
        sections.append(f"\n**Version:** {spec.metadata.version}")
        if spec.metadata.owner:
            sections.append(f"**Owner:** {spec.metadata.owner}")

        # Providers
        if spec.providers:
            sections.append("\n## Providers\n")
            for name, p in spec.providers.items():
                sections.append(f"- **{name}**: `{p.model}` ({p.provider})")

        # Agents
        sections.append("\n## Agents\n")
        for name, a in spec.agents.items():
            desc = a.description or "No description"
            sections.append(f"### {name}\n")
            sections.append(f"- **Description:** {desc}")
            sections.append(f"- **Prompt:** {a.prompt[:100]}{'...' if len(a.prompt) > 100 else ''}")
            if a.provider:
                sections.append(f"- **Provider:** {a.provider}")
            if a.guardrails:
                sections.append(f"- **Guardrails:** {', '.join(a.guardrails)}")

        # Workflows
        sections.append("\n## Workflows\n")
        for name, w in spec.workflows.items():
            sections.append(f"### {name}\n")
            sections.append(f"- **Pattern:** {w.pattern}")
            if w.recovery:
                sections.append(
                    f"- **Recovery:** max_retries={w.recovery.max_retries}, "
                    f"timeout={w.recovery.timeout_seconds}s"
                )

        # Contracts
        if spec.contracts:
            sections.append("\n## Contracts\n")
            for name, c in spec.contracts.items():
                sections.append(f"### {name}\n")
                sections.append(
                    f"- **SLA:** p95 latency ≤ {c.sla.latency_p95_ms}ms, "
                    f"cost ≤ ${c.sla.cost_max_usd}"
                )

        # Diagram
        sections.append("\n## Architecture Diagram\n")
        sections.append("```mermaid")
        sections.append(self.to_mermaid(spec))
        sections.append("```")

        return "\n".join(sections)

    @staticmethod
    def _extract_agent_refs(agents: dict) -> list[str]:
        """Flatten agent refs from a workflow's agents dict."""
        refs: list[str] = []
        for _, ref in agents.items():
            if isinstance(ref, str):
                refs.append(ref)
            elif isinstance(ref, dict):
                refs.extend(ref.values())
        return refs

to_mermaid(spec)

Render the blueprint as a Mermaid flowchart.

Parameters:

Name Type Description Default
spec BlueprintSpec

Blueprint specification.

 required

Returns:

Type Description
str

Mermaid diagram string.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/renderer.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def to_mermaid(self, spec: BlueprintSpec) -> str:
    """Render the blueprint as a Mermaid flowchart.

    Args:
        spec: Blueprint specification.

    Returns:
        Mermaid diagram string.
    """
    lines = ["graph TD"]

    # Agent nodes
    for name, agent_spec in spec.agents.items():
        label = agent_spec.description or name
        lines.append(f"    {name}[{label}]")

    # Workflow edges
    for _, wf_spec in spec.workflows.items():
        agent_refs = self._extract_agent_refs(wf_spec.agents)
        if len(agent_refs) > 1:
            for i in range(len(agent_refs) - 1):
                lines.append(f"    {agent_refs[i]} -->|{wf_spec.pattern}| {agent_refs[i + 1]}")

        # For supervisor-like patterns, show classifier → routes
        if "classifier" in wf_spec.agents and "routes" in wf_spec.agents:
            classifier = wf_spec.agents["classifier"]
            routes = wf_spec.agents.get("routes", {})
            if isinstance(routes, dict):
                for route_name, route_ref in routes.items():
                    lines.append(f"    {classifier} -->|{route_name}| {route_ref}")

    return "\n".join(lines)

to_markdown(spec)

Render the blueprint as Markdown documentation.

Parameters:

Name Type Description Default
spec BlueprintSpec

Blueprint specification.

 required

Returns:

Type Description
str

Markdown string.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/renderer.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def to_markdown(self, spec: BlueprintSpec) -> str:
    """Render the blueprint as Markdown documentation.

    Args:
        spec: Blueprint specification.

    Returns:
        Markdown string.
    """
    sections: list[str] = []

    # Title
    sections.append(f"# {spec.metadata.name}")
    if spec.metadata.description:
        sections.append(f"\n{spec.metadata.description}")
    sections.append(f"\n**Version:** {spec.metadata.version}")
    if spec.metadata.owner:
        sections.append(f"**Owner:** {spec.metadata.owner}")

    # Providers
    if spec.providers:
        sections.append("\n## Providers\n")
        for name, p in spec.providers.items():
            sections.append(f"- **{name}**: `{p.model}` ({p.provider})")

    # Agents
    sections.append("\n## Agents\n")
    for name, a in spec.agents.items():
        desc = a.description or "No description"
        sections.append(f"### {name}\n")
        sections.append(f"- **Description:** {desc}")
        sections.append(f"- **Prompt:** {a.prompt[:100]}{'...' if len(a.prompt) > 100 else ''}")
        if a.provider:
            sections.append(f"- **Provider:** {a.provider}")
        if a.guardrails:
            sections.append(f"- **Guardrails:** {', '.join(a.guardrails)}")

    # Workflows
    sections.append("\n## Workflows\n")
    for name, w in spec.workflows.items():
        sections.append(f"### {name}\n")
        sections.append(f"- **Pattern:** {w.pattern}")
        if w.recovery:
            sections.append(
                f"- **Recovery:** max_retries={w.recovery.max_retries}, "
                f"timeout={w.recovery.timeout_seconds}s"
            )

    # Contracts
    if spec.contracts:
        sections.append("\n## Contracts\n")
        for name, c in spec.contracts.items():
            sections.append(f"### {name}\n")
            sections.append(
                f"- **SLA:** p95 latency ≤ {c.sla.latency_p95_ms}ms, "
                f"cost ≤ ${c.sla.cost_max_usd}"
            )

    # Diagram
    sections.append("\n## Architecture Diagram\n")
    sections.append("```mermaid")
    sections.append(self.to_mermaid(spec))
    sections.append("```")

    return "\n".join(sections)

RuntimeGraph

Executable graph of compiled workflows.

Each workflow is a fully wired Pattern instance ready to run.

Parameters:

Name Type Description Default
workflows dict[str, Pattern]

Mapping of workflow name → compiled Pattern.

 required
agents dict[str, Agent] | None

Mapping of agent name → Agent instance.

 None
metadata dict[str, Any] | None

Blueprint metadata dict.

 None
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class RuntimeGraph:
    """Executable graph of compiled workflows.

    Each workflow is a fully wired ``Pattern`` instance ready to run.

    Args:
        workflows: Mapping of workflow name → compiled Pattern.
        agents: Mapping of agent name → Agent instance.
        metadata: Blueprint metadata dict.
    """

    def __init__(
        self,
        workflows: dict[str, Pattern],
        agents: dict[str, Agent] | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        self._workflows = workflows
        self._agents = agents or {}
        self._metadata = metadata or {}

    # -- Hook-wiring convenience methods --

    def wire_trace(self, bus: Any) -> None:
        """Set trace_bus on all patterns and their agents.

        Args:
            bus: A ``TraceEventBus`` instance.
        """
        for pattern in self._workflows.values():
            pattern.set_trace_bus(bus)
        for agent in self._agents.values():
            agent.set_trace_bus(bus)

    def wire_context(self, ledger: Any) -> None:
        """Set context ledger on all agents in all workflows.

        Args:
            ledger: A ``ContextLedger`` instance.
        """
        for agent in self._agents.values():
            agent.set_context(ledger)

    def wire_compressor(self, compressor: Any) -> None:
        """Set compressor on all agents.

        Args:
            compressor: A ``MessageCompressor`` instance.
        """
        for agent in self._agents.values():
            agent.set_compressor(compressor)

    def wire_cost_tracker(self, tracker: Any) -> None:
        """Set cost tracker on all agents.

        Args:
            tracker: A ``CostTracker`` instance.
        """
        for agent in self._agents.values():
            agent.set_cost_tracker(tracker)

    # -- Execution --

    async def run(self, workflow: str, task: str) -> Result:
        """Run a workflow by name.

        Args:
            workflow: Workflow name from the blueprint.
            task: Input task string.

        Returns:
            Pattern ``Result``.

        Raises:
            KeyError: If workflow name doesn't exist.
        """
        if workflow not in self._workflows:
            available = list(self._workflows.keys())
            raise KeyError(f"Unknown workflow '{workflow}'. Available: {available}")

        pattern = self._workflows[workflow]
        return await pattern.run(task)

    async def stream(self, workflow: str, task: str) -> AsyncIterator[str]:
        """Stream results from a workflow.

        Falls back to ``run()`` and yields the full output if the pattern
        doesn't support native streaming.
        """
        result = await self.run(workflow, task)
        yield result.output

    def describe(self) -> dict[str, Any]:
        """Introspect the runtime graph.

        Returns:
            Dict with metadata and workflow descriptions.
        """
        return {
            "metadata": self._metadata,
            "workflows": {
                name: {
                    "pattern_type": type(pattern).__name__,
                }
                for name, pattern in self._workflows.items()
            },
            "agents": list(self._agents.keys()),
        }

    @property
    def workflow_names(self) -> list[str]:
        return list(self._workflows.keys())

    @property
    def agents(self) -> dict[str, Agent]:
        return dict(self._agents)

    def __contains__(self, workflow: str) -> bool:
        return workflow in self._workflows

wire_trace(bus)

Set trace_bus on all patterns and their agents.

Parameters:

Name Type Description Default
bus Any

A TraceEventBus instance.

 required
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
36
37
38
39
40
41
42
43
44
45
def wire_trace(self, bus: Any) -> None:
    """Set trace_bus on all patterns and their agents.

    Args:
        bus: A ``TraceEventBus`` instance.
    """
    for pattern in self._workflows.values():
        pattern.set_trace_bus(bus)
    for agent in self._agents.values():
        agent.set_trace_bus(bus)

wire_context(ledger)

Set context ledger on all agents in all workflows.

Parameters:

Name Type Description Default
ledger Any

A ContextLedger instance.

 required
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
47
48
49
50
51
52
53
54
def wire_context(self, ledger: Any) -> None:
    """Set context ledger on all agents in all workflows.

    Args:
        ledger: A ``ContextLedger`` instance.
    """
    for agent in self._agents.values():
        agent.set_context(ledger)

wire_compressor(compressor)

Set compressor on all agents.

Parameters:

Name Type Description Default
compressor Any

A MessageCompressor instance.

 required
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
56
57
58
59
60
61
62
63
def wire_compressor(self, compressor: Any) -> None:
    """Set compressor on all agents.

    Args:
        compressor: A ``MessageCompressor`` instance.
    """
    for agent in self._agents.values():
        agent.set_compressor(compressor)

wire_cost_tracker(tracker)

Set cost tracker on all agents.

Parameters:

Name Type Description Default
tracker Any

A CostTracker instance.

 required
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
65
66
67
68
69
70
71
72
def wire_cost_tracker(self, tracker: Any) -> None:
    """Set cost tracker on all agents.

    Args:
        tracker: A ``CostTracker`` instance.
    """
    for agent in self._agents.values():
        agent.set_cost_tracker(tracker)

run(workflow, task) async

Run a workflow by name.

Parameters:

Name Type Description Default
workflow str

Workflow name from the blueprint.

 required
task str

Input task string.

 required

Returns:

Type Description
Result

Pattern Result.

Raises:

Type Description
KeyError

If workflow name doesn't exist.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def run(self, workflow: str, task: str) -> Result:
    """Run a workflow by name.

    Args:
        workflow: Workflow name from the blueprint.
        task: Input task string.

    Returns:
        Pattern ``Result``.

    Raises:
        KeyError: If workflow name doesn't exist.
    """
    if workflow not in self._workflows:
        available = list(self._workflows.keys())
        raise KeyError(f"Unknown workflow '{workflow}'. Available: {available}")

    pattern = self._workflows[workflow]
    return await pattern.run(task)

stream(workflow, task) async

Stream results from a workflow.

Falls back to run() and yields the full output if the pattern doesn't support native streaming.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
 96
 97
 98
 99
100
101
102
103
async def stream(self, workflow: str, task: str) -> AsyncIterator[str]:
    """Stream results from a workflow.

    Falls back to ``run()`` and yields the full output if the pattern
    doesn't support native streaming.
    """
    result = await self.run(workflow, task)
    yield result.output

describe()

Introspect the runtime graph.

Returns:

Type Description
dict[str, Any]

Dict with metadata and workflow descriptions.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/runtime.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def describe(self) -> dict[str, Any]:
    """Introspect the runtime graph.

    Returns:
        Dict with metadata and workflow descriptions.
    """
    return {
        "metadata": self._metadata,
        "workflows": {
            name: {
                "pattern_type": type(pattern).__name__,
            }
            for name, pattern in self._workflows.items()
        },
        "agents": list(self._agents.keys()),
    }

AgentSpec

Bases: BaseModel

Specification of a single agent.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/agents.py 
 8
 9
10
11
12
13
14
15
class AgentSpec(BaseModel):
    """Specification of a single agent."""

    prompt: str = Field(..., description="System prompt for this agent")
    provider: str = Field(default="", description="Provider ref from providers dict")
    tools: list[str] = Field(default_factory=list, description="Tool names available to this agent")
    description: str = Field(default="", description="What this agent does")
    guardrails: list[str] = Field(default_factory=list, description="Guardrail refs")

BlueprintSpec

Bases: BaseModel

Root specification for a declarative agent system.

This is the top-level Pydantic model that represents a complete blueprint YAML/JSON document.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/spec.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class BlueprintSpec(BaseModel):
    """Root specification for a declarative agent system.

    This is the top-level Pydantic model that represents a complete
    blueprint YAML/JSON document.
    """

    api_version: str = Field(default="pyagent/v1", description="Schema version")
    metadata: MetadataSpec
    providers: dict[str, ProviderBindingSpec] = Field(default_factory=dict)
    context: ContextConfigSpec | None = Field(default=None)
    agents: dict[str, AgentSpec]
    workflows: dict[str, WorkflowSpec]
    contracts: dict[str, ContractSpec] = Field(default_factory=dict)
    observability: ObservabilitySpec | None = Field(default=None)

ContextConfigSpec

Bases: BaseModel

Context configuration for a blueprint.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/context.py 
34
35
36
37
38
39
class ContextConfigSpec(BaseModel):
    """Context configuration for a blueprint."""

    memory: MemoryConfig = Field(default_factory=MemoryConfig)
    compression: CompressionConfig = Field(default_factory=CompressionConfig)
    redaction: RedactionConfig | None = Field(default=None)

ContractSpec

Bases: BaseModel

Input/output contract for a workflow.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/contracts.py 
18
19
20
21
22
23
24
25
26
27
class ContractSpec(BaseModel):
    """Input/output contract for a workflow."""

    input: dict[str, Any] = Field(
        default_factory=dict, description="Input schema (JSON Schema-like)"
    )
    output: dict[str, Any] = Field(
        default_factory=dict, description="Output schema (JSON Schema-like)"
    )
    sla: SLASpec = Field(default_factory=SLASpec, description="SLA constraints")

MetadataSpec

Bases: BaseModel

Blueprint metadata.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/metadata.py 
 8
 9
10
11
12
13
14
15
class MetadataSpec(BaseModel):
    """Blueprint metadata."""

    name: str = Field(..., description="Human-readable blueprint name")
    version: str = Field(default="0.1.0", description="Semantic version")
    description: str = Field(default="", description="What this blueprint does")
    tags: list[str] = Field(default_factory=list, description="Categorization tags")
    owner: str = Field(default="", description="Team or individual owner")

ObservabilitySpec

Bases: BaseModel

Observability configuration for a blueprint.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/observability.py 
23
24
25
26
27
class ObservabilitySpec(BaseModel):
    """Observability configuration for a blueprint."""

    tracing: TracingConfig = Field(default_factory=TracingConfig)
    cost_budget: CostBudgetConfig | None = Field(default=None)

ProviderBindingSpec

Bases: BaseModel

Binding of a logical provider name to a model + backend.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/providers.py 
 8
 9
10
11
12
13
14
15
class ProviderBindingSpec(BaseModel):
    """Binding of a logical provider name to a model + backend."""

    model: str = Field(..., description="Model identifier (e.g., 'gpt-4.1-mini')")
    provider: str = Field(
        default="mock", description="Provider backend (mock, openai, anthropic, litellm)"
    )
    fallback_ref: str = Field(default="", description="Fallback provider ref name")

WorkflowSpec

Bases: BaseModel

Specification of a workflow: pattern + agent wiring.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/workflows.py 
18
19
20
21
22
23
24
25
class WorkflowSpec(BaseModel):
    """Specification of a workflow: pattern + agent wiring."""

    pattern: str = Field(..., description="Pattern registry name (e.g., 'supervisor', 'pipeline')")
    agents: dict[str, Any] = Field(default_factory=dict, description="Role → agent ref mapping")
    config: dict[str, Any] = Field(default_factory=dict, description="Pattern-specific config")
    recovery: RecoverySpec | None = Field(default=None, description="Recovery configuration")
    guardrails: list[str] = Field(default_factory=list, description="Guardrail refs")

BlueprintTester

Run conformance tests for blueprint contracts using MockLLM.

Compiles the blueprint with mock providers and verifies that: - Workflows produce output - Output type matches contract expectations - SLA constraints are within bounds (basic checks)

Parameters:

Name Type Description Default
compiler BlueprintCompiler | None

Optional compiler instance. Creates one if not provided.

 None
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/tester.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class BlueprintTester:
    """Run conformance tests for blueprint contracts using MockLLM.

    Compiles the blueprint with mock providers and verifies that:
    - Workflows produce output
    - Output type matches contract expectations
    - SLA constraints are within bounds (basic checks)

    Args:
        compiler: Optional compiler instance. Creates one if not provided.
    """

    def __init__(self, compiler: BlueprintCompiler | None = None) -> None:
        self._compiler = compiler or BlueprintCompiler()

    async def test(
        self,
        spec: BlueprintSpec,
        test_inputs: dict[str, str] | None = None,
    ) -> list[TestResult]:
        """Run contract conformance tests for all workflows with contracts.

        Args:
            spec: Blueprint spec to test.
            test_inputs: Optional mapping of workflow name → test input.
                If not provided, uses a default test string.

        Returns:
            List of ``TestResult`` for each contract.
        """
        graph = self._compiler.compile(spec)
        results: list[TestResult] = []

        for contract_name, contract in spec.contracts.items():
            if contract_name not in spec.workflows:
                results.append(
                    TestResult(
                        workflow=contract_name,
                        passed=False,
                        error=f"Contract references non-existent workflow '{contract_name}'",
                    )
                )
                continue

            test_input = (test_inputs or {}).get(
                contract_name, "Test input for contract validation"
            )

            try:
                result = await graph.run(contract_name, test_input)
                checks: dict[str, bool] = {}

                # Check: output is non-empty
                checks["output_non_empty"] = bool(result.output)

                # Check: output type matches
                expected_type = contract.output.get("type", "string")
                if expected_type == "string":
                    checks["output_is_string"] = isinstance(result.output, str)

                # Check: output length within bounds
                max_tokens = contract.input.get("max_tokens")
                if max_tokens is not None:
                    estimated_tokens = len(test_input) // 4
                    checks["input_within_token_limit"] = estimated_tokens <= max_tokens

                all_passed = all(checks.values())
                results.append(
                    TestResult(
                        workflow=contract_name,
                        passed=all_passed,
                        output=result.output,
                        checks=checks,
                    )
                )

            except Exception as exc:
                results.append(
                    TestResult(
                        workflow=contract_name,
                        passed=False,
                        error=f"{type(exc).__name__}: {exc}",
                    )
                )

        return results

    def summary(self, results: list[TestResult]) -> str:
        """Human-readable summary of test results."""
        lines: list[str] = []
        passed = sum(1 for r in results if r.passed)
        total = len(results)

        lines.append(f"\nContract Tests: {passed}/{total} passed\n")

        for r in results:
            status = "✓ PASS" if r.passed else "✗ FAIL"
            lines.append(f"  {status}  {r.workflow}")
            if r.error:
                lines.append(f"         Error: {r.error}")
            for check, ok in r.checks.items():
                mark = "✓" if ok else "✗"
                lines.append(f"         {mark} {check}")

        return "\n".join(lines)

test(spec, test_inputs=None) async

Run contract conformance tests for all workflows with contracts.

Parameters:

Name Type Description Default
spec BlueprintSpec

Blueprint spec to test.

 required
test_inputs dict[str, str] | None

Optional mapping of workflow name → test input. If not provided, uses a default test string.

 None

Returns:

Type Description
list[TestResult]

List of TestResult for each contract.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/tester.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def test(
    self,
    spec: BlueprintSpec,
    test_inputs: dict[str, str] | None = None,
) -> list[TestResult]:
    """Run contract conformance tests for all workflows with contracts.

    Args:
        spec: Blueprint spec to test.
        test_inputs: Optional mapping of workflow name → test input.
            If not provided, uses a default test string.

    Returns:
        List of ``TestResult`` for each contract.
    """
    graph = self._compiler.compile(spec)
    results: list[TestResult] = []

    for contract_name, contract in spec.contracts.items():
        if contract_name not in spec.workflows:
            results.append(
                TestResult(
                    workflow=contract_name,
                    passed=False,
                    error=f"Contract references non-existent workflow '{contract_name}'",
                )
            )
            continue

        test_input = (test_inputs or {}).get(
            contract_name, "Test input for contract validation"
        )

        try:
            result = await graph.run(contract_name, test_input)
            checks: dict[str, bool] = {}

            # Check: output is non-empty
            checks["output_non_empty"] = bool(result.output)

            # Check: output type matches
            expected_type = contract.output.get("type", "string")
            if expected_type == "string":
                checks["output_is_string"] = isinstance(result.output, str)

            # Check: output length within bounds
            max_tokens = contract.input.get("max_tokens")
            if max_tokens is not None:
                estimated_tokens = len(test_input) // 4
                checks["input_within_token_limit"] = estimated_tokens <= max_tokens

            all_passed = all(checks.values())
            results.append(
                TestResult(
                    workflow=contract_name,
                    passed=all_passed,
                    output=result.output,
                    checks=checks,
                )
            )

        except Exception as exc:
            results.append(
                TestResult(
                    workflow=contract_name,
                    passed=False,
                    error=f"{type(exc).__name__}: {exc}",
                )
            )

    return results

summary(results)

Human-readable summary of test results.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/tester.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def summary(self, results: list[TestResult]) -> str:
    """Human-readable summary of test results."""
    lines: list[str] = []
    passed = sum(1 for r in results if r.passed)
    total = len(results)

    lines.append(f"\nContract Tests: {passed}/{total} passed\n")

    for r in results:
        status = "✓ PASS" if r.passed else "✗ FAIL"
        lines.append(f"  {status}  {r.workflow}")
        if r.error:
            lines.append(f"         Error: {r.error}")
        for check, ok in r.checks.items():
            mark = "✓" if ok else "✗"
            lines.append(f"         {mark} {check}")

    return "\n".join(lines)

TestResult dataclass

Result of a single contract test.

Attributes:

Name Type Description
workflow str

Workflow name tested.
passed bool

Whether the test passed.
output str

The actual output from the workflow.
checks dict[str, bool]

Dict of check name → pass/fail.
error str | None

Error message if the test failed with an exception.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/tester.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@dataclass
class TestResult:
    """Result of a single contract test.

    Attributes:
        workflow: Workflow name tested.
        passed: Whether the test passed.
        output: The actual output from the workflow.
        checks: Dict of check name → pass/fail.
        error: Error message if the test failed with an exception.
    """

    workflow: str
    passed: bool
    output: str = ""
    checks: dict[str, bool] = field(default_factory=dict)
    error: str | None = None

BlueprintValidator

Run static checks on a BlueprintSpec.

Checks: - All agent refs in workflows exist in agents dict - All provider refs in agents exist in providers dict - Pattern names are registered - No cyclic workflow dependencies - SLA values are realistic - Security: no hardcoded API keys in prompts

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/validator.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class BlueprintValidator:
    """Run static checks on a BlueprintSpec.

    Checks:
    - All agent refs in workflows exist in agents dict
    - All provider refs in agents exist in providers dict
    - Pattern names are registered
    - No cyclic workflow dependencies
    - SLA values are realistic
    - Security: no hardcoded API keys in prompts
    """

    def validate(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Run all validation checks.

        Args:
            spec: The blueprint spec to validate.

        Returns:
            List of issues found (may be empty).
        """
        issues: list[ValidationIssue] = []
        issues.extend(self._check_agent_refs(spec))
        issues.extend(self._check_provider_refs(spec))
        issues.extend(self._check_pattern_names(spec))
        issues.extend(self._check_contract_refs(spec))
        issues.extend(self._check_sla_values(spec))
        issues.extend(self._check_security(spec))
        return issues

    def _check_agent_refs(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Ensure all agent refs in workflows point to defined agents."""
        issues: list[ValidationIssue] = []
        agent_names = set(spec.agents.keys())

        for wf_name, wf_spec in spec.workflows.items():
            for role, ref in wf_spec.agents.items():
                if isinstance(ref, str) and ref not in agent_names:
                    issues.append(
                        ValidationIssue(
                            path=f"workflows.{wf_name}.agents.{role}",
                            message=f"Agent ref '{ref}' not found in agents. Available: {sorted(agent_names)}",
                            severity=IssueSeverity.ERROR,
                        )
                    )
                elif isinstance(ref, dict):
                    for sub_role, sub_ref in ref.items():
                        if isinstance(sub_ref, str) and sub_ref not in agent_names:
                            issues.append(
                                ValidationIssue(
                                    path=f"workflows.{wf_name}.agents.{role}.{sub_role}",
                                    message=f"Agent ref '{sub_ref}' not found in agents.",
                                    severity=IssueSeverity.ERROR,
                                )
                            )
        return issues

    def _check_provider_refs(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Ensure all provider refs in agents point to defined providers."""
        issues: list[ValidationIssue] = []
        provider_names = set(spec.providers.keys())

        for agent_name, agent_spec in spec.agents.items():
            if agent_spec.provider and agent_spec.provider not in provider_names:
                issues.append(
                    ValidationIssue(
                        path=f"agents.{agent_name}.provider",
                        message=f"Provider ref '{agent_spec.provider}' not found. Available: {sorted(provider_names)}",
                        severity=IssueSeverity.ERROR,
                    )
                )
        return issues

    def _check_pattern_names(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Ensure all pattern names are registered."""
        issues: list[ValidationIssue] = []
        known = set(list_patterns())

        for wf_name, wf_spec in spec.workflows.items():
            if wf_spec.pattern not in known:
                issues.append(
                    ValidationIssue(
                        path=f"workflows.{wf_name}.pattern",
                        message=f"Unknown pattern '{wf_spec.pattern}'. Known: {sorted(known)}",
                        severity=IssueSeverity.ERROR,
                    )
                )
        return issues

    def _check_contract_refs(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Ensure contracts reference existing workflows."""
        issues: list[ValidationIssue] = []
        wf_names = set(spec.workflows.keys())

        for contract_name in spec.contracts:
            if contract_name not in wf_names:
                issues.append(
                    ValidationIssue(
                        path=f"contracts.{contract_name}",
                        message=f"Contract '{contract_name}' references non-existent workflow.",
                        severity=IssueSeverity.WARNING,
                    )
                )
        return issues

    def _check_sla_values(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Warn about unrealistic SLA values."""
        issues: list[ValidationIssue] = []
        for name, contract in spec.contracts.items():
            if contract.sla.latency_p95_ms < 100:
                issues.append(
                    ValidationIssue(
                        path=f"contracts.{name}.sla.latency_p95_ms",
                        message=f"Latency SLA {contract.sla.latency_p95_ms}ms is unrealistically low for LLM calls.",
                        severity=IssueSeverity.WARNING,
                    )
                )
        return issues

    def _check_security(self, spec: BlueprintSpec) -> list[ValidationIssue]:
        """Check for hardcoded API keys in prompts."""
        issues: list[ValidationIssue] = []
        key_patterns = ["sk-", "sk-ant-", "api_key=", "API_KEY", "Bearer "]

        for agent_name, agent_spec in spec.agents.items():
            for pattern in key_patterns:
                if pattern in agent_spec.prompt:
                    issues.append(
                        ValidationIssue(
                            path=f"agents.{agent_name}.prompt",
                            message=f"Possible hardcoded API key detected (contains '{pattern}').",
                            severity=IssueSeverity.ERROR,
                        )
                    )
                    break
        return issues

validate(spec)

Run all validation checks.

Parameters:

Name Type Description Default
spec BlueprintSpec

The blueprint spec to validate.

 required

Returns:

Type Description
list[ValidationIssue]

List of issues found (may be empty).
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/validator.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def validate(self, spec: BlueprintSpec) -> list[ValidationIssue]:
    """Run all validation checks.

    Args:
        spec: The blueprint spec to validate.

    Returns:
        List of issues found (may be empty).
    """
    issues: list[ValidationIssue] = []
    issues.extend(self._check_agent_refs(spec))
    issues.extend(self._check_provider_refs(spec))
    issues.extend(self._check_pattern_names(spec))
    issues.extend(self._check_contract_refs(spec))
    issues.extend(self._check_sla_values(spec))
    issues.extend(self._check_security(spec))
    return issues

ValidationIssue dataclass

A single validation finding.

Attributes:

Name Type Description
path str

Dotted path to the problematic field.
message str

Human-readable description.
severity IssueSeverity

Error, warning, or info.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/validator.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
@dataclass(frozen=True)
class ValidationIssue:
    """A single validation finding.

    Attributes:
        path: Dotted path to the problematic field.
        message: Human-readable description.
        severity: Error, warning, or info.
    """

    path: str
    message: str
    severity: IssueSeverity

load_blueprint(path)

Load a blueprint from a YAML or JSON file.

Parameters:

Name Type Description Default
path str | Path

Path to a .yaml, .yml, or .json file.

 required

Returns:

Type Description
BlueprintSpec

Validated BlueprintSpec.

Raises:

Type Description
BlueprintLoadError

If the file is missing, unreadable, or invalid.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/loader.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def load_blueprint(path: str | Path) -> BlueprintSpec:
    """Load a blueprint from a YAML or JSON file.

    Args:
        path: Path to a ``.yaml``, ``.yml``, or ``.json`` file.

    Returns:
        Validated ``BlueprintSpec``.

    Raises:
        BlueprintLoadError: If the file is missing, unreadable, or invalid.
    """
    path = Path(path)

    if not path.exists():
        raise BlueprintLoadError(f"Blueprint file not found: {path}")

    text = path.read_text(encoding="utf-8")

    try:
        if path.suffix in (".yaml", ".yml"):
            data = yaml.safe_load(text)
        elif path.suffix == ".json":
            data = json.loads(text)
        else:
            raise BlueprintLoadError(f"Unsupported file extension: {path.suffix}")
    except (yaml.YAMLError, json.JSONDecodeError) as exc:
        raise BlueprintLoadError(f"Parse error in {path}: {exc}") from exc

    if not isinstance(data, dict):
        raise BlueprintLoadError(f"Blueprint must be a mapping, got {type(data).__name__}")

    try:
        return BlueprintSpec(**data)
    except ValidationError as exc:
        raise BlueprintLoadError(f"Schema validation failed for {path}:\n{exc}") from exc

load_blueprint_from_str(text, fmt='yaml')

Load a blueprint from a string.

Parameters:

Name Type Description Default
text str

YAML or JSON text.

 required
fmt str

"yaml" or "json".

 'yaml'

Returns:

Type Description
BlueprintSpec

Validated BlueprintSpec.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/loader.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
def load_blueprint_from_str(text: str, fmt: str = "yaml") -> BlueprintSpec:
    """Load a blueprint from a string.

    Args:
        text: YAML or JSON text.
        fmt: ``"yaml"`` or ``"json"``.

    Returns:
        Validated ``BlueprintSpec``.
    """
    data = json.loads(text) if fmt == "json" else yaml.safe_load(text)

    return BlueprintSpec(**data)