Skip to content

pyagent-blueprint API Reference

pyagent_blueprint

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

AdapterRegistry

Discovers adapters via Python entry points.

Core never imports any adapter package directly — third parties can ship a backend without touching this repo at all by registering an entry point in the pyagent_blueprint.adapters group.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
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 AdapterRegistry:
    """Discovers adapters via Python entry points.

    Core never imports any adapter package directly — third parties can
    ship a backend without touching this repo at all by registering an
    entry point in the ``pyagent_blueprint.adapters`` group.
    """

    GROUP = "pyagent_blueprint.adapters"

    @staticmethod
    def discover() -> dict[str, type[RuntimeAdapter]]:
        """Return all installed adapters, keyed by entry-point name.

        An entry point that fails to import (e.g. an adapter registered
        by `pyproject.toml` whose own optional dependency — like
        `pyagent-patterns` for the `pyagent` adapter — isn't installed)
        is skipped rather than raised: one adapter's missing dependency
        must never break discovery of every OTHER adapter. This is what
        lets `validate`/`generate` degrade gracefully with zero runtime
        packages installed, while the zero-dependency reference adapters
        remain fully discoverable.
        """
        import logging
        from importlib.metadata import entry_points

        logger = logging.getLogger(__name__)

        found: dict[str, type[RuntimeAdapter]] = {}
        try:
            eps = entry_points(group=AdapterRegistry.GROUP)
        except TypeError:  # pragma: no cover - py<3.10 signature fallback
            eps = entry_points().get(AdapterRegistry.GROUP, [])  # type: ignore[attr-defined]
        for ep in eps:
            try:
                found[ep.name] = ep.load()
            except ImportError as exc:
                logger.debug(
                    "Adapter '%s' could not be loaded (missing dependency?): %s", ep.name, exc
                )
                continue
        return found

    @staticmethod
    def get(name: str) -> type[RuntimeAdapter]:
        """Look up a single adapter class by entry-point name.

        Raises:
            KeyError: If no adapter is registered under `name`.
        """
        adapters = AdapterRegistry.discover()
        if name not in adapters:
            raise KeyError(f"No adapter registered as '{name}'. Installed: {sorted(adapters)}")
        return adapters[name]

discover() staticmethod

Return all installed adapters, keyed by entry-point name.

An entry point that fails to import (e.g. an adapter registered by pyproject.toml whose own optional dependency — like pyagent-patterns for the pyagent adapter — isn't installed) is skipped rather than raised: one adapter's missing dependency must never break discovery of every OTHER adapter. This is what lets validate/generate degrade gracefully with zero runtime packages installed, while the zero-dependency reference adapters remain fully discoverable.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
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
@staticmethod
def discover() -> dict[str, type[RuntimeAdapter]]:
    """Return all installed adapters, keyed by entry-point name.

    An entry point that fails to import (e.g. an adapter registered
    by `pyproject.toml` whose own optional dependency — like
    `pyagent-patterns` for the `pyagent` adapter — isn't installed)
    is skipped rather than raised: one adapter's missing dependency
    must never break discovery of every OTHER adapter. This is what
    lets `validate`/`generate` degrade gracefully with zero runtime
    packages installed, while the zero-dependency reference adapters
    remain fully discoverable.
    """
    import logging
    from importlib.metadata import entry_points

    logger = logging.getLogger(__name__)

    found: dict[str, type[RuntimeAdapter]] = {}
    try:
        eps = entry_points(group=AdapterRegistry.GROUP)
    except TypeError:  # pragma: no cover - py<3.10 signature fallback
        eps = entry_points().get(AdapterRegistry.GROUP, [])  # type: ignore[attr-defined]
    for ep in eps:
        try:
            found[ep.name] = ep.load()
        except ImportError as exc:
            logger.debug(
                "Adapter '%s' could not be loaded (missing dependency?): %s", ep.name, exc
            )
            continue
    return found

get(name) staticmethod

Look up a single adapter class by entry-point name.

Raises:

Type Description
KeyError

If no adapter is registered under name.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
220
221
222
223
224
225
226
227
228
229
230
@staticmethod
def get(name: str) -> type[RuntimeAdapter]:
    """Look up a single adapter class by entry-point name.

    Raises:
        KeyError: If no adapter is registered under `name`.
    """
    adapters = AdapterRegistry.discover()
    if name not in adapters:
        raise KeyError(f"No adapter registered as '{name}'. Installed: {sorted(adapters)}")
    return adapters[name]

AdapterResult

Normalized result envelope.

Every adapter must map its native return shape into this, so callers never branch on adapter identity (e.g. never special-case "if adapter is LangGraph, read .raw['messages'][-1]").

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class AdapterResult:
    """Normalized result envelope.

    Every adapter must map its native return shape into this, so callers
    never branch on adapter identity (e.g. never special-case "if adapter
    is LangGraph, read `.raw['messages'][-1]`").
    """

    def __init__(self, output: Any, raw: Any = None, usage: dict[str, Any] | None = None) -> None:
        self.output = output  # the primary answer, always present
        self.raw = raw  # adapter-native object, for advanced users
        self.usage = usage or {}  # tokens/cost if the adapter can report it

    def __repr__(self) -> str:  # pragma: no cover - debugging aid only
        return f"AdapterResult(output={self.output!r}, usage={self.usage!r})"

Capability

Bases: Flag

Optional features an adapter may or may not support.

Core only ever requires COMPILE + RUN (i.e. the two abstract methods below). Everything else is negotiated at runtime via these flags so the contract never assumes a graph, async streaming, or native tool-calling exists.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class Capability(Flag):
    """Optional features an adapter may or may not support.

    Core only ever requires COMPILE + RUN (i.e. the two abstract methods
    below). Everything else is negotiated at runtime via these flags so
    the contract never assumes a graph, async streaming, or native
    tool-calling exists.
    """

    NONE = 0
    STREAMING = auto()
    NATIVE_TOOL_CALLING = auto()
    SYNC_EXECUTION = auto()  # some SDKs are sync-only
    PARTIAL_WORKFLOW_RUN = auto()  # can run a subset of a workflow (debugging)
    ROUND_TRIP = auto()  # can export back to a BlueprintIR losslessly for shared constructs

CompiledArtifact dataclass

Result of compile(): the opaque native handle plus diagnostics.

Attributes:

Name Type Description
handle Any

Opaque, framework-native compiled object. Core never inspects this — that's the whole point of the abstraction.
diagnostics list[CompileDiagnostic]

Every governance feature the blueprint declared that this adapter could NOT honor, as structured diagnostics. Empty means every declared feature was either honored or not applicable to this blueprint.
intent dict[str, str]

Optional map of workflow name -> original pattern name, preserved so pattern intent survives even when an adapter lowers a named pattern (e.g. "debate") to a generic graph.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass
class CompiledArtifact:
    """Result of `compile()`: the opaque native handle plus diagnostics.

    Attributes:
        handle: Opaque, framework-native compiled object. Core never
            inspects this — that's the whole point of the abstraction.
        diagnostics: Every governance feature the blueprint declared that
            this adapter could NOT honor, as structured diagnostics.
            Empty means every declared feature was either honored or not
            applicable to this blueprint.
        intent: Optional map of workflow name -> original pattern name,
            preserved so pattern intent survives even when an adapter
            lowers a named pattern (e.g. "debate") to a generic graph.
    """

    handle: Any
    diagnostics: list[CompileDiagnostic] = field(default_factory=list)
    intent: dict[str, str] = field(default_factory=dict)

CompileDiagnostic dataclass

A single structured diagnostic emitted during compile().

Attributes:

Name Type Description
code DiagnosticCode

A stable DiagnosticCode from diagnostics.py.
path str

Dotted path into the blueprint that triggered this diagnostic, e.g. "workflows.support.recovery".
detail str

Adapter-specific human-readable detail.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@dataclass(frozen=True)
class CompileDiagnostic:
    """A single structured diagnostic emitted during `compile()`.

    Attributes:
        code: A stable `DiagnosticCode` from `diagnostics.py`.
        path: Dotted path into the blueprint that triggered this
            diagnostic, e.g. ``"workflows.support.recovery"``.
        detail: Adapter-specific human-readable detail.
    """

    code: DiagnosticCode
    path: str
    detail: str = ""

RuntimeAdapter

Bases: ABC

Compiles a framework-agnostic BlueprintIR into a runnable object native to a specific agent framework, and executes it.

Deliberately minimal: only compile and run are required. This is the lowest common denominator across graph-based (LangGraph), turn-based (AutoGen), role-based (CrewAI), handoff-based (OpenAI Agents SDK), event-driven (Semantic Kernel), and hand-rolled loop runtimes.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
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
class RuntimeAdapter(ABC):
    """Compiles a framework-agnostic `BlueprintIR` into a runnable object
    native to a specific agent framework, and executes it.

    Deliberately minimal: only `compile` and `run` are required. This is
    the lowest common denominator across graph-based (LangGraph),
    turn-based (AutoGen), role-based (CrewAI), handoff-based (OpenAI
    Agents SDK), event-driven (Semantic Kernel), and hand-rolled loop
    runtimes.
    """

    name: str
    capabilities: Capability = Capability.NONE

    @abstractmethod
    def compile(self, ir: BlueprintIR) -> CompiledArtifact:
        """Compile a `BlueprintIR` into a `CompiledArtifact`.

        Must report every governance feature it cannot honor via
        `CompiledArtifact.diagnostics` — never drop one silently.
        """

    @abstractmethod
    async def run(
        self, compiled: CompiledArtifact, workflow: str, input_: str, **kwargs: Any
    ) -> AdapterResult:
        """Execute a compiled workflow.

        Adapters that are natively sync (`Capability.SYNC_EXECUTION`) wrap
        their own sync call internally (e.g. via `asyncio.to_thread`) —
        callers of `RuntimeAdapter` always await, even for sync-native SDKs.

        Raises:
            UnknownWorkflowError: If `workflow` doesn't exist in `compiled`.
        """

    # -- Optional capability-gated methods (default: not implemented) --

    async def stream(
        self, compiled: CompiledArtifact, workflow: str, input_: str, **kwargs: Any
    ) -> AsyncIterator[Any]:
        raise NotImplementedError(f"{self.name} does not declare Capability.STREAMING")

    def supported_patterns(self) -> list[str]:
        """Pattern/topology vocabulary this adapter understands, for
        `validator.py`'s optional pattern-existence check. Adapters
        without a fixed pattern vocabulary (e.g. a loop-based adapter)
        return an empty list — validator treats that as "no constraint",
        not an error."""
        return []

    def export(self, compiled: CompiledArtifact) -> Any:
        """Export a compiled artifact back toward a portable form.

        Only meaningful if `Capability.ROUND_TRIP` is declared. Default
        raises — adapters that support round-tripping (e.g. a future
        Agent Spec bridge) override this.
        """
        raise NotImplementedError(f"{self.name} does not declare Capability.ROUND_TRIP")

compile(ir) abstractmethod

Compile a BlueprintIR into a CompiledArtifact.

Must report every governance feature it cannot honor via CompiledArtifact.diagnostics — never drop one silently.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
130
131
132
133
134
135
136
@abstractmethod
def compile(self, ir: BlueprintIR) -> CompiledArtifact:
    """Compile a `BlueprintIR` into a `CompiledArtifact`.

    Must report every governance feature it cannot honor via
    `CompiledArtifact.diagnostics` — never drop one silently.
    """

run(compiled, workflow, input_, **kwargs) abstractmethod async

Execute a compiled workflow.

Adapters that are natively sync (Capability.SYNC_EXECUTION) wrap their own sync call internally (e.g. via asyncio.to_thread) — callers of RuntimeAdapter always await, even for sync-native SDKs.

Raises:

Type Description
UnknownWorkflowError

If workflow doesn't exist in compiled.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
@abstractmethod
async def run(
    self, compiled: CompiledArtifact, workflow: str, input_: str, **kwargs: Any
) -> AdapterResult:
    """Execute a compiled workflow.

    Adapters that are natively sync (`Capability.SYNC_EXECUTION`) wrap
    their own sync call internally (e.g. via `asyncio.to_thread`) —
    callers of `RuntimeAdapter` always await, even for sync-native SDKs.

    Raises:
        UnknownWorkflowError: If `workflow` doesn't exist in `compiled`.
    """

supported_patterns()

Pattern/topology vocabulary this adapter understands, for validator.py's optional pattern-existence check. Adapters without a fixed pattern vocabulary (e.g. a loop-based adapter) return an empty list — validator treats that as "no constraint", not an error.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
159
160
161
162
163
164
165
def supported_patterns(self) -> list[str]:
    """Pattern/topology vocabulary this adapter understands, for
    `validator.py`'s optional pattern-existence check. Adapters
    without a fixed pattern vocabulary (e.g. a loop-based adapter)
    return an empty list — validator treats that as "no constraint",
    not an error."""
    return []

export(compiled)

Export a compiled artifact back toward a portable form.

Only meaningful if Capability.ROUND_TRIP is declared. Default raises — adapters that support round-tripping (e.g. a future Agent Spec bridge) override this.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
167
168
169
170
171
172
173
174
def export(self, compiled: CompiledArtifact) -> Any:
    """Export a compiled artifact back toward a portable form.

    Only meaningful if `Capability.ROUND_TRIP` is declared. Default
    raises — adapters that support round-tripping (e.g. a future
    Agent Spec bridge) override this.
    """
    raise NotImplementedError(f"{self.name} does not declare Capability.ROUND_TRIP")

UnknownWorkflowError

Bases: Exception

Raised by RuntimeAdapter.run()/stream() for an unresolvable workflow name.

Adapters MUST raise this (not let an internal AttributeError/KeyError leak through) so callers get one documented, catchable error type regardless of which adapter is installed.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter.py 
107
108
109
110
111
112
113
class UnknownWorkflowError(Exception):
    """Raised by `RuntimeAdapter.run()`/`stream()` for an unresolvable workflow name.

    Adapters MUST raise this (not let an internal AttributeError/KeyError
    leak through) so callers get one documented, catchable error type
    regardless of which adapter is installed.
    """

BlueprintCompiler

DEPRECATED: use PyAgentAdapter (or any RuntimeAdapter) directly.

Compile a BlueprintSpec into a RuntimeGraph, delegating to pyagent_blueprint.adapters.pyagent_adapter.PyAgentAdapter.

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 
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
class BlueprintCompiler:
    """DEPRECATED: use `PyAgentAdapter` (or any `RuntimeAdapter`) directly.

    Compile a BlueprintSpec into a RuntimeGraph, delegating to
    `pyagent_blueprint.adapters.pyagent_adapter.PyAgentAdapter`.

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

    def __init__(self, provider_registry: Any = None) -> None:
        warnings.warn(
            "BlueprintCompiler is deprecated; use "
            "pyagent_blueprint.adapters.pyagent_adapter.PyAgentAdapter "
            "(or any RuntimeAdapter) directly. This shim will be removed "
            "in a future major version.",
            DeprecationWarning,
            stacklevel=2,
        )
        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.
        """
        from pyagent_blueprint.adapters.pyagent_adapter import (
            PyAgentAdapter,
            PyAgentCompilationError,
        )

        ir = BlueprintIR.from_spec(spec)
        adapter = PyAgentAdapter(provider_registry=self._provider_registry)
        try:
            compiled = adapter.compile(ir)
        except PyAgentCompilationError as exc:
            raise CompilationError(str(exc)) from exc
        return compiled.handle

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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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.
    """
    from pyagent_blueprint.adapters.pyagent_adapter import (
        PyAgentAdapter,
        PyAgentCompilationError,
    )

    ir = BlueprintIR.from_spec(spec)
    adapter = PyAgentAdapter(provider_registry=self._provider_registry)
    try:
        compiled = adapter.compile(ir)
    except PyAgentCompilationError as exc:
        raise CompilationError(str(exc)) from exc
    return compiled.handle

CompilationError

Bases: Exception

Raised when a blueprint cannot be compiled.

Preserved for backward compatibility — wraps adapters.pyagent_adapter.PyAgentCompilationError.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/compiler.py 
25
26
27
28
29
30
class CompilationError(Exception):
    """Raised when a blueprint cannot be compiled.

    Preserved for backward compatibility — wraps
    `adapters.pyagent_adapter.PyAgentCompilationError`.
    """

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 
 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 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 = "",
        adapter: str | None = None,
    ) -> 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.
            adapter: Optional adapter entry-point name to validate `pattern`
                against. If omitted, any installed adapter's vocabulary is
                used; if none is installed, the pattern name isn't checked
                at all — scaffolding still works with zero runtime packages
                installed.

        Returns:
            YAML string.

        Raises:
            ValueError: If a pattern vocabulary is available (from the
                resolved adapter, or any installed adapter) and `pattern`
                isn't in it.
        """
        known, _resolved_adapter = _resolve_pattern_vocabulary(adapter)
        if known and pattern not in known:
            raise ValueError(f"Unknown pattern '{pattern}'. Available: {sorted(known)}")

        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='', adapter=None)

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.

 ''
adapter str | None

Optional adapter entry-point name to validate pattern against. If omitted, any installed adapter's vocabulary is used; if none is installed, the pattern name isn't checked at all — scaffolding still works with zero runtime packages installed.

 None

Returns:

Type Description
str

YAML string.

Raises:

Type Description
ValueError

If a pattern vocabulary is available (from the resolved adapter, or any installed adapter) and pattern isn't in it.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/generator.py 
 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
def generate(
    self,
    pattern: str,
    agents: list[str],
    *,
    name: str = "my-blueprint",
    version: str = "0.1.0",
    description: str = "",
    adapter: str | None = None,
) -> 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.
        adapter: Optional adapter entry-point name to validate `pattern`
            against. If omitted, any installed adapter's vocabulary is
            used; if none is installed, the pattern name isn't checked
            at all — scaffolding still works with zero runtime packages
            installed.

    Returns:
        YAML string.

    Raises:
        ValueError: If a pattern vocabulary is available (from the
            resolved adapter, or any installed adapter) and `pattern`
            isn't in it.
    """
    known, _resolved_adapter = _resolve_pattern_vocabulary(adapter)
    if known and pattern not in known:
        raise ValueError(f"Unknown pattern '{pattern}'. Available: {sorted(known)}")

    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)

BlueprintIR dataclass

Root intermediate representation of a complete blueprint.

This is the single normalized structure that RuntimeAdapter.compile(), BlueprintDiffer, BlueprintRenderer, and any future exporter (e.g. an Agent Spec backend) should consume — never the raw BlueprintSpec.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/ir.py 
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
@dataclass(frozen=True)
class BlueprintIR:
    """Root intermediate representation of a complete blueprint.

    This is the single normalized structure that `RuntimeAdapter.compile()`,
    `BlueprintDiffer`, `BlueprintRenderer`, and any future exporter (e.g. an
    Agent Spec backend) should consume — never the raw `BlueprintSpec`.
    """

    api_version: str
    name: str
    version: str
    description: str = ""
    tags: tuple[str, ...] = ()
    owner: str = ""
    providers: dict[str, ProviderBindingIR] = field(default_factory=dict)
    agents: dict[str, AgentIR] = field(default_factory=dict)
    workflows: dict[str, WorkflowIR] = field(default_factory=dict)
    contracts: dict[str, ContractIR] = field(default_factory=dict)
    memory: MemoryPolicyIR | None = None
    observability: ObservabilityIR | None = None
    extensions: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_spec(cls, spec: BlueprintSpec) -> BlueprintIR:
        """Build a `BlueprintIR` from a validated `BlueprintSpec`.

        This is the ONE place that knows how to read the Pydantic schema.
        Everything downstream (adapters, differ, renderer) should be
        written against this dataclass shape instead.
        """
        providers = {
            name: ProviderBindingIR(
                name=name,
                model=binding.model,
                provider=binding.provider,
                fallback_ref=binding.fallback_ref,
            )
            for name, binding in spec.providers.items()
        }

        agents = {
            name: AgentIR(
                name=name,
                prompt=agent_spec.prompt,
                provider=agent_spec.provider,
                tools=tuple(agent_spec.tools),
                description=agent_spec.description,
                guardrails=tuple(agent_spec.guardrails),
            )
            for name, agent_spec in spec.agents.items()
        }

        workflows = {
            name: WorkflowIR(
                name=name,
                pattern=wf.pattern,
                agents=dict(wf.agents),
                config=dict(wf.config),
                recovery=(
                    RecoveryIR(
                        max_retries=wf.recovery.max_retries,
                        timeout_seconds=wf.recovery.timeout_seconds,
                        fallback_provider=wf.recovery.fallback_provider,
                    )
                    if wf.recovery is not None
                    else None
                ),
                guardrails=tuple(wf.guardrails),
            )
            for name, wf in spec.workflows.items()
        }

        contracts = {
            name: ContractIR(
                workflow=name,
                input_schema=dict(contract.input),
                output_schema=dict(contract.output),
                sla=SLAIR(
                    latency_p95_ms=contract.sla.latency_p95_ms,
                    cost_max_usd=contract.sla.cost_max_usd,
                    quality_min=contract.sla.quality_min,
                ),
            )
            for name, contract in spec.contracts.items()
        }

        memory = None
        if spec.context is not None:
            redaction = spec.context.redaction
            memory = MemoryPolicyIR(
                working_max_tokens=spec.context.memory.working_max_tokens,
                session_backend=spec.context.memory.session_backend,
                semantic_enabled=spec.context.memory.semantic_enabled,
                compression_policy=spec.context.compression.policy,
                compression_target_ratio=spec.context.compression.target_ratio,
                redaction_max_sensitivity=(redaction.max_sensitivity if redaction else None),
            )

        observability = None
        if spec.observability is not None:
            budget = spec.observability.cost_budget
            observability = ObservabilityIR(
                tracing_enabled=spec.observability.tracing.enabled,
                tracing_exporter=spec.observability.tracing.exporter,
                cost_budget_daily_usd=(budget.daily_usd if budget else None),
                cost_budget_alert_threshold=(budget.alert_threshold if budget else 0.8),
            )

        return cls(
            api_version=spec.api_version,
            name=spec.metadata.name,
            version=spec.metadata.version,
            description=spec.metadata.description,
            tags=tuple(spec.metadata.tags),
            owner=spec.metadata.owner,
            providers=providers,
            agents=agents,
            workflows=workflows,
            contracts=contracts,
            memory=memory,
            observability=observability,
            extensions={},
        )

    def governance_features(self) -> dict[str, bool]:
        """Which governance features this blueprint actually declares.

        Used by adapters/conformance checks to know what MUST be honored
        or diagnosed — never silently dropped. Keys map 1:1 to the
        diagnostic codes in `diagnostics.py`.
        """
        return {
            "routing": any(p.fallback_ref for p in self.providers.values()),
            "budget": any(c.sla.cost_max_usd for c in self.contracts.values())
            or (
                self.observability is not None
                and self.observability.cost_budget_daily_usd is not None
            ),
            "sla": bool(self.contracts),
            "memory_tier": self.memory is not None,
            "recovery": any(w.recovery is not None for w in self.workflows.values()),
            "guardrails": any(a.guardrails for a in self.agents.values())
            or any(w.guardrails for w in self.workflows.values()),
            "checkpoint": any(w.config.get("human_in_the_loop") for w in self.workflows.values()),
        }

from_spec(spec) classmethod

Build a BlueprintIR from a validated BlueprintSpec.

This is the ONE place that knows how to read the Pydantic schema. Everything downstream (adapters, differ, renderer) should be written against this dataclass shape instead.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/ir.py 
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
231
232
233
@classmethod
def from_spec(cls, spec: BlueprintSpec) -> BlueprintIR:
    """Build a `BlueprintIR` from a validated `BlueprintSpec`.

    This is the ONE place that knows how to read the Pydantic schema.
    Everything downstream (adapters, differ, renderer) should be
    written against this dataclass shape instead.
    """
    providers = {
        name: ProviderBindingIR(
            name=name,
            model=binding.model,
            provider=binding.provider,
            fallback_ref=binding.fallback_ref,
        )
        for name, binding in spec.providers.items()
    }

    agents = {
        name: AgentIR(
            name=name,
            prompt=agent_spec.prompt,
            provider=agent_spec.provider,
            tools=tuple(agent_spec.tools),
            description=agent_spec.description,
            guardrails=tuple(agent_spec.guardrails),
        )
        for name, agent_spec in spec.agents.items()
    }

    workflows = {
        name: WorkflowIR(
            name=name,
            pattern=wf.pattern,
            agents=dict(wf.agents),
            config=dict(wf.config),
            recovery=(
                RecoveryIR(
                    max_retries=wf.recovery.max_retries,
                    timeout_seconds=wf.recovery.timeout_seconds,
                    fallback_provider=wf.recovery.fallback_provider,
                )
                if wf.recovery is not None
                else None
            ),
            guardrails=tuple(wf.guardrails),
        )
        for name, wf in spec.workflows.items()
    }

    contracts = {
        name: ContractIR(
            workflow=name,
            input_schema=dict(contract.input),
            output_schema=dict(contract.output),
            sla=SLAIR(
                latency_p95_ms=contract.sla.latency_p95_ms,
                cost_max_usd=contract.sla.cost_max_usd,
                quality_min=contract.sla.quality_min,
            ),
        )
        for name, contract in spec.contracts.items()
    }

    memory = None
    if spec.context is not None:
        redaction = spec.context.redaction
        memory = MemoryPolicyIR(
            working_max_tokens=spec.context.memory.working_max_tokens,
            session_backend=spec.context.memory.session_backend,
            semantic_enabled=spec.context.memory.semantic_enabled,
            compression_policy=spec.context.compression.policy,
            compression_target_ratio=spec.context.compression.target_ratio,
            redaction_max_sensitivity=(redaction.max_sensitivity if redaction else None),
        )

    observability = None
    if spec.observability is not None:
        budget = spec.observability.cost_budget
        observability = ObservabilityIR(
            tracing_enabled=spec.observability.tracing.enabled,
            tracing_exporter=spec.observability.tracing.exporter,
            cost_budget_daily_usd=(budget.daily_usd if budget else None),
            cost_budget_alert_threshold=(budget.alert_threshold if budget else 0.8),
        )

    return cls(
        api_version=spec.api_version,
        name=spec.metadata.name,
        version=spec.metadata.version,
        description=spec.metadata.description,
        tags=tuple(spec.metadata.tags),
        owner=spec.metadata.owner,
        providers=providers,
        agents=agents,
        workflows=workflows,
        contracts=contracts,
        memory=memory,
        observability=observability,
        extensions={},
    )

governance_features()

Which governance features this blueprint actually declares.

Used by adapters/conformance checks to know what MUST be honored or diagnosed — never silently dropped. Keys map 1:1 to the diagnostic codes in diagnostics.py.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/ir.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def governance_features(self) -> dict[str, bool]:
    """Which governance features this blueprint actually declares.

    Used by adapters/conformance checks to know what MUST be honored
    or diagnosed — never silently dropped. Keys map 1:1 to the
    diagnostic codes in `diagnostics.py`.
    """
    return {
        "routing": any(p.fallback_ref for p in self.providers.values()),
        "budget": any(c.sla.cost_max_usd for c in self.contracts.values())
        or (
            self.observability is not None
            and self.observability.cost_budget_daily_usd is not None
        ),
        "sla": bool(self.contracts),
        "memory_tier": self.memory is not None,
        "recovery": any(w.recovery is not None for w in self.workflows.values()),
        "guardrails": any(a.guardrails for a in self.agents.values())
        or any(w.guardrails for w in self.workflows.values()),
        "checkpoint": any(w.config.get("human_in_the_loop") for w in self.workflows.values()),
    }

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."""

AgentUnitMetadata dataclass

Resolved, packaging-ready metadata for an Agent Unit.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/packaging.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True)
class AgentUnitMetadata:
    """Resolved, packaging-ready metadata for an Agent Unit."""

    name: str
    version: str
    author: str
    runtime: str
    dependencies: tuple[str, ...] = field(default_factory=tuple)
    spec_sha256: str = ""

    def to_dict(self) -> dict[str, object]:
        return {
            "name": self.name,
            "version": self.version,
            "author": self.author,
            "runtime": self.runtime,
            "dependencies": list(self.dependencies),
            "spec_sha256": self.spec_sha256,
            "unit_schema_version": "1.0",
        }

PackagingError

Bases: Exception

Raised when a blueprint cannot be packaged into an Agent Unit.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/packaging.py 
28
29
class PackagingError(Exception):
    """Raised when a blueprint cannot be packaged into an Agent Unit."""

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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
    package: PackageSpec | None = Field(
        default=None, description="Optional 'Agent Unit' packaging metadata"
    )
    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)

PackageSpec

Bases: BaseModel

Optional packaging metadata for producing an 'Agent Unit' artifact.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/schema/package.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class PackageSpec(BaseModel):
    """Optional packaging metadata for producing an 'Agent Unit' artifact."""

    name: str = Field(..., description="Distribution name for the packaged Agent Unit")
    version: str = Field(default="0.1.0", description="Semantic version of the Agent Unit")
    author: str = Field(default="", description="Author or team publishing this unit")
    runtime: str = Field(
        ...,
        description=(
            "Name of the RuntimeAdapter this unit targets (e.g. 'pyagent', "
            "'simple_loop', 'state_machine', 'sequential_chain', 'single_agent', "
            "or a third-party adapter name). Must match a discoverable adapter "
            "at packaging time."
        ),
    )
    dependencies: list[str] = Field(
        default_factory=list,
        description="Extra distribution dependencies beyond the runtime adapter itself",
    )

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
26
27
28
29
30
31
32
33
34
class WorkflowSpec(BaseModel):
    """Specification of a workflow: pattern + agent wiring."""

    # Agent/pattern wiring (routes, stages, teams, ...) must be nested under
    # `agents:`/`config:` — a misplaced sibling key (e.g. `classifier:` next
    # to `pattern:` instead of inside `agents:`) is silently dropped under
    # the default "ignore extra" behavior, producing an empty `agents={}`
    # that only fails at run time with a confusing AttributeError deep in
    # the pattern's `_execute`. Forbidding extras turns that into a clear,
    # immediate ValidationError at load time.
    model_config = ConfigDict(extra="forbid")

    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 
 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
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, IF at least one
        installed adapter declares a pattern vocabulary. Degrades
        gracefully (no error) when no adapter is installed, or when the
        installed adapter(s) don't have a fixed pattern vocabulary at
        all (e.g. a loop-based adapter) — this is what lets `validate`
        keep working even with zero runtime packages installed."""
        issues: list[ValidationIssue] = []
        known = _known_pattern_names()
        if known is None:
            return issues

        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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 
50
51
52
53
54
55
56
57
58
59
60
61
62
@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

render_adapter_template(framework_name, adapter_name=None, dist_name=None)

Render the file contents for a starter adapter package.

Parameters:

Name Type Description Default
framework_name str

Human-readable SDK name, e.g. "LangGraph".

 required
adapter_name str | None

entry-point / RuntimeAdapter.name value. Defaults to a snake_case derivation of framework_name.

 None
dist_name str | None

PyPI distribution name. Defaults to "pyagent-blueprint-adapter-".

 None

Returns:

Type Description
dict[str, str]

Mapping of relative file path -> file content, ready to be
dict[str, str]

written to disk.
Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter_template.py 
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
def render_adapter_template(
    framework_name: str,
    adapter_name: str | None = None,
    dist_name: str | None = None,
) -> dict[str, str]:
    """Render the file contents for a starter adapter package.

    Args:
        framework_name: Human-readable SDK name, e.g. "LangGraph".
        adapter_name: entry-point / RuntimeAdapter.name value. Defaults
            to a snake_case derivation of framework_name.
        dist_name: PyPI distribution name. Defaults to
            "pyagent-blueprint-adapter-<adapter_name-with-dashes>".

    Returns:
        Mapping of relative file path -> file content, ready to be
        written to disk.
    """
    slug = "".join(c if c.isalnum() else "_" for c in framework_name.lower()).strip("_")
    while "__" in slug:
        slug = slug.replace("__", "_")

    adapter_name = adapter_name or slug
    module_name = f"pyagent_blueprint_adapter_{slug}"
    class_name = "".join(part.capitalize() for part in slug.split("_")) + "Adapter"
    dist_name = dist_name or f"pyagent-blueprint-adapter-{slug.replace('_', '-')}"

    fmt_kwargs = {
        "framework_name": framework_name,
        "adapter_name": adapter_name,
        "module_name": module_name,
        "class_name": class_name,
        "dist_name": dist_name,
    }

    return {
        "pyproject.toml": _PYPROJECT_TEMPLATE.format(**fmt_kwargs),
        f"src/{module_name}/__init__.py": "",
        f"src/{module_name}/adapter.py": _ADAPTER_MODULE_TEMPLATE.format(**fmt_kwargs),
        f"tests/test_{slug}_adapter_conformance.py": _TEST_MODULE_TEMPLATE.format(**fmt_kwargs),
        "README.md": _README_TEMPLATE.format(**fmt_kwargs),
    }

write_adapter_template(framework_name, output_dir, adapter_name=None, dist_name=None)

Render and write a starter adapter package to output_dir.

Returns the output directory path.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/adapter_template.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def write_adapter_template(
    framework_name: str,
    output_dir: str | Path,
    adapter_name: str | None = None,
    dist_name: str | None = None,
) -> Path:
    """Render and write a starter adapter package to `output_dir`.

    Returns the output directory path.
    """
    files = render_adapter_template(framework_name, adapter_name, dist_name)
    out_dir = Path(output_dir)

    for rel_path, content in files.items():
        full_path = out_dir / rel_path
        full_path.parent.mkdir(parents=True, exist_ok=True)
        full_path.write_text(content)

    return out_dir

blueprint_contracts_json_schema(ir)

Render every declared contract in a BlueprintIR as JSON Schema documents, keyed by workflow name.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/contract.py 
61
62
63
64
def blueprint_contracts_json_schema(ir: BlueprintIR) -> dict[str, dict[str, Any]]:
    """Render every declared contract in a `BlueprintIR` as JSON Schema
    documents, keyed by workflow name."""
    return {name: contract_json_schema(contract) for name, contract in ir.contracts.items()}

contract_json_schema(contract)

Render a single workflow's ContractIR as a JSON Schema document.

The document's $id-less shape is intentionally minimal: input/ output are the declared JSON Schemas verbatim (empty dict means "unconstrained"), and SLA constraints are surfaced under the x-pyagent extension namespace since they have no native JSON Schema representation — a non-Python consumer can still read them, it's just not part of core JSON Schema vocabulary.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/contract.py 
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
def contract_json_schema(contract: ContractIR) -> dict[str, Any]:
    """Render a single workflow's `ContractIR` as a JSON Schema document.

    The document's `$id`-less shape is intentionally minimal: `input`/
    `output` are the declared JSON Schemas verbatim (empty dict means
    "unconstrained"), and SLA constraints are surfaced under the
    `x-pyagent` extension namespace since they have no native JSON
    Schema representation — a non-Python consumer can still read them,
    it's just not part of core JSON Schema vocabulary.
    """
    return {
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "title": f"pyagent-blueprint contract: {contract.workflow}",
        "type": "object",
        "properties": {
            "input": contract.input_schema or {"description": "unconstrained"},
            "output": contract.output_schema or {"description": "unconstrained"},
        },
        "x-pyagent": {
            "schema_version": CONTRACT_SCHEMA_VERSION,
            "sla": {
                "latency_p95_ms": contract.sla.latency_p95_ms,
                "cost_max_usd": contract.sla.cost_max_usd,
                "quality_min": contract.sla.quality_min,
            },
        },
    }

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)

package_blueprint(spec, raw_source, source_filename, output_dir)

Build a distributable Agent Unit archive.

Produces <output_dir>/<name>-<version>.agentunit.zip containing: - unit.json — the AgentUnitMetadata manifest - the original blueprint source, preserved under its own filename

Returns the path to the written archive.

Source code in packages/pyagent-blueprint/src/pyagent_blueprint/packaging.py 
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
def package_blueprint(
    spec: BlueprintSpec,
    raw_source: str,
    source_filename: str,
    output_dir: str | Path,
) -> Path:
    """Build a distributable Agent Unit archive.

    Produces ``<output_dir>/<name>-<version>.agentunit.zip`` containing:
      - ``unit.json`` — the AgentUnitMetadata manifest
      - the original blueprint source, preserved under its own filename

    Returns the path to the written archive.
    """
    metadata = build_metadata(spec, raw_source)

    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    archive_path = out_dir / f"{metadata.name}-{metadata.version}.agentunit.zip"

    with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("unit.json", json.dumps(metadata.to_dict(), indent=2))
        zf.writestr(Path(source_filename).name, raw_source)

    return archive_path