Flow runtime

The core runtime: the flow orchestrator, execution context, nodes, the model registry, and the built-in pattern helpers for routing, fan-out/fan-in, and subflows.

core

Implements Context, Floe, and PenguiFlow runtime with backpressure-aware queues, cycle detection, and graceful shutdown semantics.

DEFAULT_QUEUE_MAXSIZE module-attribute

DEFAULT_QUEUE_MAXSIZE = 64

PenguiFlow

PenguiFlow(*adjacencies: tuple[Node, Sequence[Node]], queue_maxsize: int = DEFAULT_QUEUE_MAXSIZE, allow_cycles: bool = False, middlewares: Sequence[Middleware] | None = None, emit_errors_to_rookery: bool = False, state_store: StateStore | None = None, message_bus: MessageBus | None = None)

Async runtime that wires nodes into a graph and drives message routing.

A flow is defined by (node, successors) adjacency tuples. On construction the runtime builds the graph, validates it against cycles (unless allowed), and connects a synthetic OpenSea ingress endpoint to every source node and a Rookery egress sink to every terminal node. Call :meth:run to spawn one worker task per node, feed inputs with :meth:emit, and collect terminal results with :meth:fetch; call :meth:stop for graceful shutdown.

Every edge is a bounded queue, so producers block once a downstream consumer falls behind (backpressure). Work is tracked per trace_id, enabling per-run accounting, trace-scoped roundtrips (emit(..., trace_id=...) paired with fetch(trace_id=...)), and single-run cancellation via :meth:cancel without disturbing other in-flight runs.

Parameters:

Name	Type	Description	Default
*adjacencies	tuple[Node, Sequence[Node]]	One or more (node, successors) tuples describing directed edges. A node may appear as a successor of several parents; endpoints are wired automatically.	()
queue_maxsize	int	Bound applied to every edge queue. Controls how much buffering (and therefore backpressure) each edge allows. <= 0 creates unbounded queues. Defaults to :data:DEFAULT_QUEUE_MAXSIZE.	DEFAULT_QUEUE_MAXSIZE
allow_cycles	bool	When True, skip acyclicity validation so controller/loop graphs with back edges are permitted. Defaults to False.	False
middlewares	Sequence[Middleware] | None	Optional sequence of :class:~penguiflow.middlewares.Middleware callables invoked for every emitted :class:~penguiflow.metrics.FlowEvent.	None
emit_errors_to_rookery	bool	When True, terminal :class:~penguiflow.errors.FlowError values (after retries are exhausted) are delivered to the Rookery sink in addition to being logged. Defaults to False.	False
state_store	StateStore | None	Optional :class:~penguiflow.state.StateStore used to persist :class:~penguiflow.metrics.FlowEvent history and remote bindings.	None
message_bus	MessageBus | None	Optional :class:~penguiflow.bus.MessageBus that receives a published envelope for every message crossing an edge.	None

registry property

registry: Any | None

Return the :class:~penguiflow.registry.ModelRegistry passed to :meth:run, if any.

add_middleware

add_middleware(middleware: Middleware) -> None

Append a middleware that will receive every subsequently emitted FlowEvent.

Parameters:

Name	Type	Description	Default
middleware	Middleware	An async callable invoked with each :class:~penguiflow.metrics.FlowEvent.	required

run

run(*, registry: Any | None = None) -> None

Start the flow by spawning one worker task per node on the running event loop.

Must be called from within a running asyncio event loop. When a registry is supplied it is validated up front to ensure every node requiring validation has a matching entry.

Parameters:

Name	Type	Description	Default
registry	Any | None	Optional :class:~penguiflow.registry.ModelRegistry providing typed input/output adapters for the flow's nodes.	None

Raises:

Type	Description
RuntimeError	If the flow is already running, or the registry is missing entries for nodes that require validation.

stop async

stop() -> None

Gracefully shut down the flow, cancelling all worker and auxiliary tasks.

Cancels node workers, in-flight node invocations, registered external tasks, the Rookery fetch dispatcher, and drains pending bus-publish tasks, then clears all per-trace bookkeeping. Idempotent: returns immediately if the flow is not running.

emit async

emit(msg: Any, to: Node | Sequence[Node] | None = None, *, trace_id: str | None = None) -> None

Inject a message into the flow through the OpenSea ingress endpoint.

Awaits queue capacity, so this call honours backpressure. Passing trace_id enables a trace-scoped roundtrip: the message's trace id is (re)stamped and the runtime begins buffering only that trace's terminal results for a subsequent fetch(trace_id=...) call, keeping concurrent runs isolated.

Parameters:

Name	Type	Description	Default
msg	Any	The value to inject, typically a :class:~penguiflow.types.Message.	required
to	Node | Sequence[Node] | None	Target ingress node(s), or None to reach every source node.	None
trace_id	str | None	Optional trace id to bind this emission to a scoped roundtrip. When provided, msg must be a :class:~penguiflow.types.Message (or expose a writable trace_id attribute).	None

Raises:

Type	Description
RuntimeError	If trace_id is given while the flow is not running.
TypeError	If trace_id is given but msg cannot carry a trace id.

emit_nowait

emit_nowait(msg: Any, to: Node | Sequence[Node] | None = None, *, trace_id: str | None = None) -> None

Inject a message through OpenSea without awaiting queue capacity.

Non-blocking counterpart to :meth:emit. Trace-scoped roundtrips are not supported here because they require awaiting dispatcher setup.

Parameters:

Name	Type	Description	Default
msg	Any	The value to inject, typically a :class:~penguiflow.types.Message.	required
to	Node | Sequence[Node] | None	Target ingress node(s), or None to reach every source node.	None
trace_id	str | None	Not supported; must be None.	None

Raises:

Type	Description
RuntimeError	If trace_id is provided.
QueueFull	If a target ingress queue is at capacity.

emit_chunk async

emit_chunk(*, parent: Message, text: str, stream_id: str | None = None, seq: int | None = None, done: bool = False, meta: dict[str, Any] | None = None, to: Node | Sequence[Node] | None = None) -> StreamChunk

Emit a streaming chunk from outside a node via OpenSea context.

fetch async

fetch(from_: Node | Sequence[Node] | None = None, *, trace_id: str | None = None) -> Any

Await the next terminal result delivered to the Rookery egress sink.

Without trace_id this returns the next result from any run. With trace_id it returns only results belonging to that run, activating a background dispatcher that demultiplexes Rookery output into per-trace queues; other traces' results are routed to a shared queue so nothing is dropped. from_ filtering is incompatible with the trace-scoped path.

Parameters:

Name	Type	Description	Default
from_	Node | Sequence[Node] | None	Optional source node filter (only valid before trace-scoped fetching is enabled).	None
trace_id	str | None	Optional trace id to fetch results for a single run.	None

Returns:

Type	Description
Any	The next terminal payload. If it was a :class:~penguiflow.types.Message, the
Any	message itself (or its unwrapped value) is returned.

Raises:

Type	Description
RuntimeError	If from_ is combined with trace_id, or if used after trace-scoped fetching has been enabled.

fetch_any async

fetch_any(from_: Node | Sequence[Node] | None = None, *, trace_id: str | None = None) -> Any

Await the next Rookery result; alias of :meth:fetch at the runtime level.

Unlike :meth:Context.fetch_any, the runtime exposes a single Rookery stream, so there are no multi-queue race semantics here; this simply forwards to :meth:fetch.

Parameters:

Name	Type	Description	Default
from_	Node | Sequence[Node] | None	Optional source node filter (only valid before trace-scoped fetching).	None
trace_id	str | None	Optional trace id to fetch results for a single run.	None

Returns:

Type	Description
Any	The next terminal payload delivered to the Rookery sink.

load_history async

load_history(trace_id: str) -> Sequence[StoredEvent]

Return the persisted history for trace_id from the state store.

ensure_trace_event

ensure_trace_event(trace_id: str) -> asyncio.Event

Return (and create if needed) the cancellation event for trace_id.

register_external_task

register_external_task(trace_id: str, task: Future[Any]) -> None

Track an externally created task for cancellation bookkeeping.

save_remote_binding async

save_remote_binding(binding: RemoteBinding) -> None

Persist a remote binding if a state store is configured.

record_remote_event async

record_remote_event(*, event: str, node: Node, context: Context, trace_id: str | None, latency_ms: float | None, level: int = logging.INFO, extra: Mapping[str, Any] | None = None) -> None

Emit a structured :class:FlowEvent for remote transport activity.

cancel async

cancel(trace_id: str) -> bool

Cancel a single in-flight run, leaving all other traces untouched.

Sets the trace's cancellation event, drains queued messages for that trace from every edge and fetch queue, cancels its in-flight node invocations, and releases any pending trace-scoped roundtrip. Idempotent for an already-cancelling trace. Emits trace_cancel_start (and later trace_cancel_finish) flow events.

Parameters:

Name	Type	Description	Default
trace_id	str	Identifier of the run to cancel.	required

Returns:

Type	Description
bool	True if the trace was active and cancellation was initiated; False if no
bool	work for that trace was in flight.

Raises:

Type	Description
RuntimeError	If the flow is not running.

Context

Context(owner: Node | Endpoint, runtime: PenguiFlow | None = None)

Per-node handle for fetching inputs and emitting outputs within a running flow.

Each node worker receives its own Context bound to that node's incoming and outgoing :class:Floe edges. Node callables use it to pull the next message (:meth:fetch) and forward results (:meth:emit); the runtime also uses it internally to route work. All queue operations respect backpressure via the bounded edge queues.

Outputs are addressed with the to/from_ arguments: pass a single node, a sequence of nodes, or None to fan out to (or gather from) every connected edge. Messages carry a trace_id that is preserved end to end so the runtime can account for, cancel, and finalize all work belonging to a single run.

Attributes:

Name	Type	Description
owner	Node | Endpoint	The node or synthetic endpoint this context belongs to.
runtime	PenguiFlow | None	The owning :class:PenguiFlow, or None if the context is detached.

owner property

owner: Node | Endpoint

Return the node or synthetic endpoint this context is bound to.

runtime property

runtime: PenguiFlow | None

Return the runtime this context is attached to, if any.

add_incoming_floe

add_incoming_floe(floe: Floe) -> None

add_outgoing_floe

add_outgoing_floe(floe: Floe) -> None

emit async

emit(msg: Any, to: Node | Endpoint | Sequence[Node | Endpoint] | None = None) -> None

Forward msg onto one or more outgoing edges, awaiting queue capacity.

This is the backpressure-aware path: if a target edge queue is full, the call suspends until space is available rather than dropping the message.

Parameters:

Name	Type	Description	Default
msg	Any	The value to enqueue. Typically a :class:~penguiflow.types.Message, but any object may be sent on edges whose consumers accept it.	required
to	Node | Endpoint | Sequence[Node | Endpoint] | None	Target node(s) to send to. A single node, a sequence of nodes, or None to fan out to every outgoing edge of this context.	None

Raises:

Type	Description
RuntimeError	If this context is not attached to a running flow.
KeyError	If to names a node that is not a downstream target of this context.

emit_nowait

emit_nowait(msg: Any, to: Node | Endpoint | Sequence[Node | Endpoint] | None = None) -> None

Forward msg onto outgoing edges without awaiting, raising if a queue is full.

Non-blocking counterpart to :meth:emit. Prefer :meth:emit inside async nodes so backpressure is honoured; use this only when a synchronous, fail-fast enqueue is required.

Parameters:

Name	Type	Description	Default
msg	Any	The value to enqueue on each resolved outgoing edge.	required
to	Node | Endpoint | Sequence[Node | Endpoint] | None	Target node(s) to send to, or None to fan out to every outgoing edge.	None

Raises:

Type	Description
RuntimeError	If this context is not attached to a running flow.
KeyError	If to names a node that is not a downstream target of this context.
QueueFull	If any target edge queue is at capacity.

emit_chunk async

emit_chunk(*, parent: Message, text: str, stream_id: str | None = None, seq: int | None = None, done: bool = False, meta: dict[str, Any] | None = None, to: Node | Endpoint | Sequence[Node | Endpoint] | None = None) -> StreamChunk

Emit a streaming chunk that inherits routing metadata from parent.

The helper manages monotonically increasing sequence numbers per stream_id (defaulting to the parent's trace id) unless an explicit seq is provided. It returns the emitted StreamChunk for introspection in tests or downstream logic.

fetch_nowait

fetch_nowait(from_: Node | Endpoint | Sequence[Node | Endpoint] | None = None) -> Any

Pop the next available message without blocking.

Drains the internal buffer first (used to hold extra results from :meth:fetch_any), then tries each resolved incoming edge in turn.

Parameters:

Name	Type	Description	Default
from_	Node | Endpoint | Sequence[Node | Endpoint] | None	Source node(s) to read from, or None to consider every incoming edge.	None

Returns:

Type	Description
Any	The next available message.

Raises:

Type	Description
QueueEmpty	If no buffered item and every incoming edge queue is empty.
KeyError	If from_ names a node that is not an upstream source of this context.

fetch async

fetch(from_: Node | Endpoint | Sequence[Node | Endpoint] | None = None) -> Any

Await and return the next message from this context's incoming edges.

Buffered items are returned first. With a single incoming edge this awaits that edge's queue directly; with several it delegates to :meth:fetch_any and returns the first message to arrive.

Parameters:

Name	Type	Description	Default
from_	Node | Endpoint | Sequence[Node | Endpoint] | None	Source node(s) to read from, or None to consider every incoming edge.	None

Returns:

Type	Description
Any	The next message to become available.

Raises:

Type	Description
RuntimeError	If this context has no incoming edges to fetch from.
KeyError	If from_ names a node that is not an upstream source of this context.

fetch_any async

fetch_any(from_: Node | Endpoint | Sequence[Node | Endpoint] | None = None) -> Any

Await the first message from any incoming edge, buffering the rest.

Races all resolved incoming edges concurrently and returns the first message received. If more than one edge produces simultaneously, the extra messages are stashed in an internal buffer and served by subsequent :meth:fetch/:meth:fetch_any calls, so nothing is lost.

Parameters:

Name	Type	Description	Default
from_	Node | Endpoint | Sequence[Node | Endpoint] | None	Source node(s) to read from, or None to consider every incoming edge.	None

Returns:

Type	Description
Any	The first message to become available across the resolved edges.

Raises:

Type	Description
RuntimeError	If this context has no incoming edges to fetch from.
KeyError	If from_ names a node that is not an upstream source of this context.

outgoing_count

outgoing_count() -> int

Return the number of outgoing edges attached to this context.

queue_depth_in

queue_depth_in() -> int

Return the total number of buffered messages across all incoming edge queues.

queue_depth_out

queue_depth_out() -> int

Return the total number of buffered messages across all outgoing edge queues.

call_playbook async

call_playbook(playbook: PlaybookFactory, parent_msg: Message, *, timeout: float | None = None) -> Any

Launch a subflow playbook using the current runtime for propagation.

create

create(*adjacencies: tuple[Node, Sequence[Node]], **kwargs: Any) -> PenguiFlow

Instantiate a :class:PenguiFlow from adjacency tuples.

Thin convenience wrapper around the :class:PenguiFlow constructor; the preferred entry point for building a flow.

Parameters:

Name	Type	Description	Default
*adjacencies	tuple[Node, Sequence[Node]]	One or more (node, successors) tuples describing the graph edges.	()
**kwargs	Any	Keyword arguments forwarded verbatim to :class:PenguiFlow (e.g. queue_maxsize, allow_cycles, middlewares, state_store).	{}

Returns:

Type	Description
PenguiFlow	A configured, not-yet-running :class:PenguiFlow instance. Call
PenguiFlow	meth:PenguiFlow.run to start it.

call_playbook async

call_playbook(playbook: PlaybookFactory, parent_msg: Message, timeout: float | None = None, *, runtime: PenguiFlow | None = None) -> Any

Execute a subflow playbook and return the first Rookery payload.

node

Node abstractions for PenguiFlow runtime.

Node dataclass

Node(func: Callable[..., Awaitable[Any]], name: str | None = None, policy: NodePolicy = NodePolicy(), allow_cycle: bool = False)

Wraps an async callable with the metadata the PenguiFlow runtime needs.

A Node is the unit of work in a flow graph: it pairs a coroutine function of the form async def handler(message, ctx) -> result with a name (used for routing and registry lookups), a :class:NodePolicy governing validation/timeouts/retries, and a flag indicating whether it may participate in a routing cycle. Nodes are hashable by their generated node_id so they can be used as graph vertices/dict keys.

Attributes:

Name	Type	Description
func	Callable[..., Awaitable[Any]]	The async callable implementing the node's logic. Must be declared with async def and accept exactly two positional parameters: the incoming message and the :class:~penguiflow.core.Context.
name	str | None	Human-readable node name used in routing, logs, and registry lookups. Defaults to func.__name__ when not provided.
policy	NodePolicy	The :class:NodePolicy controlling validation, timeout, and retry behavior for this node. Defaults to a policy with no retries and full validation.
allow_cycle	bool	Whether this node is permitted to be part of a routing cycle (e.g. a controller loop) without the flow raising a :class:~penguiflow.errors.CycleError. Defaults to False.
node_id	str	Unique identifier generated automatically for the node instance; used for hashing and equality.

Raises:

Type	Description
TypeError	If func is not declared with async def.
ValueError	If func does not accept exactly two positional parameters, or if its second parameter (the context) is not positional.

func instance-attribute

func: Callable[..., Awaitable[Any]]

name class-attribute instance-attribute

name: str | None = None

policy class-attribute instance-attribute

policy: NodePolicy = field(default_factory=NodePolicy)

allow_cycle class-attribute instance-attribute

allow_cycle: bool = False

node_id class-attribute instance-attribute

node_id: str = field(init=False)

invoke async

invoke(message: Any, ctx: Context, *, registry: ModelRegistry | None) -> Any

Invoke the underlying coroutine, applying optional validation.

Looks up the node's input/output adapters from registry (unless the policy's validate is "none"), validates the message according to self.policy.validate, calls self.func, and validates the result (if any) on the way out.

Parameters:

Name	Type	Description	Default
message	Any	The inbound payload to pass to self.func.	required
ctx	Context	The runtime :class:~penguiflow.core.Context for this invocation, passed through unchanged to self.func.	required
registry	ModelRegistry | None	The :class:~penguiflow.registry.ModelRegistry used to resolve validation adapters for this node's name, or None to skip adapter lookup entirely.	required

Returns:

Type	Description
Any	The result produced by self.func, validated against the
Any	output adapter when applicable, or None if the node produced
Any	no result.

to

to(*nodes: Node) -> tuple[Node, tuple[Node, ...]]

Declare this node's successors for building the flow graph.

Parameters:

Name	Type	Description	Default
*nodes	Node	Zero or more downstream nodes that this node may emit to.	()

Returns:

Type	Description
Node	A (self, nodes) tuple, the adjacency-list entry format expected
tuple[Node, ...]	when constructing a flow's edge list.

NodePolicy dataclass

NodePolicy(validate: str = 'both', timeout_s: float | None = None, max_retries: int = 0, backoff_base: float = 0.5, backoff_mult: float = 2.0, max_backoff: float | None = None)

Per-node execution policy controlling validation, timeouts, and retries.

A NodePolicy is attached to each :class:Node and consulted by the runtime on every invocation: it decides whether the registry-backed Pydantic adapters validate the inbound message and/or outbound result, how long a single attempt may run before it is treated as a timeout, and how failed attempts are retried with exponential backoff.

Attributes:

Name	Type	Description
validate	str	Which sides of the call to validate against the node's registered adapters. One of "both" (validate message in and result out), "in" (message only), "out" (result only), or "none" (skip validation entirely). Defaults to "both".
timeout_s	float | None	Wall-clock timeout in seconds applied to each invocation attempt. None (the default) disables the timeout, letting the node run to completion.
max_retries	int	Number of retry attempts allowed after the initial invocation fails or times out. 0 (the default) means no retries; the runtime surfaces the first failure as a :class:~penguiflow.errors.FlowError.
backoff_base	float	Base delay, in seconds, used to compute the exponential backoff before each retry. Defaults to 0.5.
backoff_mult	float	Multiplier applied per retry attempt to backoff_base when computing the exponential backoff delay. Defaults to 2.0.
max_backoff	float | None	Optional upper bound, in seconds, clamping the computed backoff delay. None (the default) leaves the delay unbounded.

Raises:

Type	Description
ValueError	If validate is not one of "both", "in", "out", or "none".

validate class-attribute instance-attribute

validate: str = 'both'

timeout_s class-attribute instance-attribute

timeout_s: float | None = None

max_retries class-attribute instance-attribute

max_retries: int = 0

backoff_base class-attribute instance-attribute

backoff_base: float = 0.5

backoff_mult class-attribute instance-attribute

backoff_mult: float = 2.0

max_backoff class-attribute instance-attribute

max_backoff: float | None = None

registry

Model registry for PenguiFlow.

ModelRegistry

ModelRegistry()

Stores per-node type adapters for validation.

register

register(node_name: str, in_model: type[BaseModel], out_model: type[BaseModel]) -> None

has

has(node_name: str) -> bool

Check if a node is already registered.

adapters

adapters(node_name: str) -> tuple[TypeAdapter[Any], TypeAdapter[Any]]

models

models(node_name: str) -> tuple[type[BaseModel], type[BaseModel]]

Return the registered models for node_name.

Raises:

Type	Description
KeyError	If the node has not been registered.

patterns

Common orchestration patterns for PenguiFlow.

predicate_router

predicate_router(name: str, predicate: Callable[[Any], Sequence[Node | str] | Node | str | None], *, policy: PolicyLike | None = None) -> Node

Create a node that routes messages based on predicate outputs.

Builds and returns a :class:~penguiflow.node.Node whose handler calls predicate(msg) to decide which successor(s) to forward the message to. Successors may be named by their node name (resolved against the node's outgoing edges at emit time) or passed directly as :class:Node instances. If a policy is supplied, the predicate's proposed targets are additionally passed through :func:~penguiflow.policies.evaluate_policy before the message is emitted, allowing config-driven overrides of the routing decision.

Parameters:

Name	Type	Description	Default
name	str	Name assigned to the generated router node.	required
predicate	Callable[[Any], Sequence[Node | str] | Node | str | None]	A callable that inspects the incoming message and returns the successor(s) to route to: a single :class:Node, a node name string, a sequence of nodes/names, or None/an empty sequence to drop the message (no emission).	required
policy	PolicyLike | None	Optional routing policy consulted after the predicate; may veto or rewrite the proposed targets. Defaults to None (the predicate's decision is used as-is).	None

Returns:

Name	Type	Description
A	Node	class:Node (with NodePolicy(validate="none")) that performs
	Node	the routing when invoked by the runtime.

union_router

union_router(name: str, union_model: type[BaseModel], *, policy: PolicyLike | None = None) -> Node

Create a node that routes based on a discriminated union Pydantic model.

Builds and returns a :class:~penguiflow.node.Node whose handler validates the incoming message against union_model (a discriminated union), then routes to the successor whose name matches the validated model's kind attribute (falling back to the model's class name if kind is absent). If a policy is supplied, the resolved target is additionally passed through :func:~penguiflow.policies.evaluate_policy before emission.

Parameters:

Name	Type	Description	Default
name	str	Name assigned to the generated router node.	required
union_model	type[BaseModel]	The Pydantic discriminated-union type used to validate and classify incoming messages.	required
policy	PolicyLike | None	Optional routing policy consulted after the union match; may veto or rewrite the proposed target. Defaults to None (the union match is used as-is).	None

Returns:

Name	Type	Description
A	Node	class:Node (with NodePolicy(validate="none")) that performs
	Node	the routing when invoked by the runtime.

Raises:

Type	Description
KeyError	If no successor's name matches the validated message's kind (or class name) and no policy rewrites the selection.

join_k

join_k(name: str, k: int) -> Node

Create a node that aggregates k messages per trace_id.

Builds and returns a :class:~penguiflow.node.Node that buffers incoming messages keyed by their trace_id attribute until k messages have arrived for a given trace, then emits the aggregated result and clears that trace's bucket. If the buffered messages are :class:~penguiflow.types.Message instances, the result is a copy of the first message with its payload replaced by the list of collected payloads; otherwise the result is the raw list of buffered messages.

Parameters:

Name	Type	Description	Default
name	str	Name assigned to the generated aggregator node.	required
k	int	Number of messages to collect per trace_id before emitting. Must be positive.	required

Returns:

Name	Type	Description
A	Node	class:Node (with NodePolicy(validate="none")) that performs
	Node	the aggregation when invoked by the runtime; it returns None until
	Node	the k-th message for a trace arrives.

Raises:

Type	Description
ValueError	If k is not positive, or (when invoked) if an incoming message lacks a trace_id.

map_concurrent async

map_concurrent(items: Iterable[PayloadT], worker: Callable[[PayloadT], Awaitable[ResultT]], *, max_concurrency: int = 8) -> list[ResultT]

Run the async worker across items with bounded concurrency.

Materializes items into a list, then schedules worker for every item behind a semaphore so at most max_concurrency invocations run concurrently. Results are collected in the same order as the input items, regardless of completion order.

Parameters:

Name	Type	Description	Default
items	Iterable[PayloadT]	The iterable of payloads to process. Consumed eagerly into a list before scheduling.	required
worker	Callable[[PayloadT], Awaitable[ResultT]]	An async callable invoked once per item; its return value is collected into the result list at the item's original index.	required
max_concurrency	int	Maximum number of worker calls allowed to run at once. Values less than 1 are clamped up to 1. Defaults to 8.	8

Returns:

Type	Description
list[ResultT]	A list of results from worker, in the same order as items.

catalog

Tool catalog helpers for the planner.

SideEffect module-attribute

SideEffect: TypeAlias = Literal['pure', 'read', 'write', 'external', 'stateful']

NodeSpec dataclass

NodeSpec(node: Node, name: str, desc: str, args_model: type[BaseModel], out_model: type[BaseModel], side_effects: SideEffect = 'pure', tags: Sequence[str] = tuple(), auth_scopes: Sequence[str] = tuple(), cost_hint: str | None = None, latency_hint_ms: int | None = None, safety_notes: str | None = None, extra: Mapping[str, Any] = dict(), loading_mode: ToolLoadingMode = ToolLoadingMode.ALWAYS, examples: Sequence[ToolInputExample] = tuple())

Structured metadata describing a planner-discoverable node.

node instance-attribute

node: Node

name instance-attribute

name: str

desc instance-attribute

desc: str

args_model instance-attribute

args_model: type[BaseModel]

out_model instance-attribute

out_model: type[BaseModel]

side_effects class-attribute instance-attribute

side_effects: SideEffect = 'pure'

tags class-attribute instance-attribute

tags: Sequence[str] = field(default_factory=tuple)

auth_scopes class-attribute instance-attribute

auth_scopes: Sequence[str] = field(default_factory=tuple)

cost_hint class-attribute instance-attribute

cost_hint: str | None = None

latency_hint_ms class-attribute instance-attribute

latency_hint_ms: int | None = None

safety_notes class-attribute instance-attribute

safety_notes: str | None = None

extra class-attribute instance-attribute

extra: Mapping[str, Any] = field(default_factory=dict)

loading_mode class-attribute instance-attribute

loading_mode: ToolLoadingMode = ToolLoadingMode.ALWAYS

examples class-attribute instance-attribute

examples: Sequence[ToolInputExample] = field(default_factory=tuple)

examples_payload

examples_payload() -> list[dict[str, Any]]

to_tool_record

to_tool_record() -> dict[str, Any]

Convert the spec to a serialisable record for prompting.

ToolInputExample

Bases: BaseModel

args instance-attribute

args: dict[str, Any]

description class-attribute instance-attribute

description: str | None = None

tags class-attribute instance-attribute

tags: list[str] = []

ToolLoadingMode

Bases: str, Enum

ALWAYS class-attribute instance-attribute

ALWAYS = 'always'

DEFERRED class-attribute instance-attribute

DEFERRED = 'deferred'

build_catalog

build_catalog(nodes: Sequence[Node], registry: ModelRegistry, *, default_loading_mode: ToolLoadingMode | str | None = None) -> list[NodeSpec]

Derive :class:NodeSpec objects from runtime nodes.

tool

tool(*, desc: str | None = None, side_effects: SideEffect = 'pure', tags: Sequence[str] | None = None, auth_scopes: Sequence[str] | None = None, cost_hint: str | None = None, latency_hint_ms: int | None = None, safety_notes: str | None = None, arg_validation: Mapping[str, Any] | None = None, arg_validator: Callable[..., Any] | None = None, extra: Mapping[str, Any] | None = None, loading_mode: ToolLoadingMode | str | None = None, examples: Sequence[ToolInputExample | Mapping[str, Any]] | Mapping[str, Any] | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Annotate a node function with catalog metadata.

bus

Message bus protocol for distributed PenguiFlow edges.

MessageBus

Bases: Protocol

Protocol for pluggable message bus adapters.

publish async

publish(envelope: BusEnvelope) -> None

Publish an envelope for downstream workers.

BusEnvelope dataclass

BusEnvelope(edge: str, source: str | None, target: str | None, trace_id: str | None, payload: Any, headers: Mapping[str, Any] | None, meta: Mapping[str, Any] | None)

Structured payload published to a :class:MessageBus.

edge instance-attribute

edge: str

source instance-attribute

source: str | None

target instance-attribute

target: str | None

trace_id instance-attribute

trace_id: str | None

payload instance-attribute

payload: Any

headers instance-attribute

headers: Mapping[str, Any] | None

meta instance-attribute

meta: Mapping[str, Any] | None