Execution helper API

Session

executionkit.kit.Kit

Kit(provider: LLMProvider, *, track_cost: bool = True, messages: Sequence[dict[str, Any]] | None = None, rate_limiter: TokenBucket | None = None)

Session that holds a :class:~executionkit.provider.Provider and tracks cumulative token usage across all pattern calls.

Parameters:

Name	Type	Description	Default
provider	LLMProvider	The LLM provider to use for all calls.	required
track_cost	bool	When True (default), accumulate usage in an internal :class:~executionkit.cost.CostTracker. Set to False to disable tracking (e.g. in hot paths or tests).	True
messages	Sequence[dict[str, Any]] | None	Optional seed conversation (OpenAI-format message dicts) for multi-turn use via :meth:turn. Copied on construction.	None
rate_limiter	TokenBucket | None	Optional :class:~executionkit.engine.rate_bucket.TokenBucket. When provided, every pattern dispatch first awaits one token, pacing calls to a sustained rate (and honouring any Retry-After penalty). Defaults to None (unlimited).	None

Conversation state

:attr:messages holds the running transcript across :meth:turn calls. The single-shot pattern methods (:meth:react, :meth:consensus, …) do not read or write it — use :meth:turn for stateful conversations.

Source code in executionkit/kit.py

def __init__(
    self,
    provider: LLMProvider,
    *,
    track_cost: bool = True,
    messages: Sequence[dict[str, Any]] | None = None,
    rate_limiter: TokenBucket | None = None,
) -> None:
    self.provider = provider
    self._tracker: CostTracker | None = CostTracker() if track_cost else None
    self.messages: list[dict[str, Any]] = list(messages) if messages else []
    self._rate_limiter = rate_limiter

usage property

usage: TokenUsage

Cumulative token usage across all calls made through this Kit.

consensus async

consensus(prompt: str, **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.consensus.consensus pattern.

All keyword arguments are forwarded unchanged to :func:consensus.

Source code in executionkit/kit.py

async def consensus(self, prompt: str, **kwargs: Any) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.consensus.consensus` pattern.

    All keyword arguments are forwarded unchanged to :func:`consensus`.
    """
    return await self._run_tracked(consensus(self.provider, prompt, **kwargs))

map_reduce async

map_reduce(inputs: Sequence[str], **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.map_reduce.map_reduce pattern.

All keyword arguments are forwarded unchanged to :func:map_reduce. map_prompt_template and reduce_prompt_template are required keyword arguments.

Source code in executionkit/kit.py

async def map_reduce(
    self, inputs: Sequence[str], **kwargs: Any
) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.map_reduce.map_reduce` pattern.

    All keyword arguments are forwarded unchanged to :func:`map_reduce`.
    ``map_prompt_template`` and ``reduce_prompt_template`` are required
    keyword arguments.
    """
    return await self._run_tracked(map_reduce(self.provider, inputs, **kwargs))

pipe async

pipe(prompt: str, *steps: Callable[..., Any], **kwargs: Any) -> PatternResult[Any]

Run :func:~executionkit.compose.pipe with this Kit's provider.

All keyword arguments are forwarded unchanged to :func:pipe.

Source code in executionkit/kit.py

async def pipe(
    self, prompt: str, *steps: Callable[..., Any], **kwargs: Any
) -> PatternResult[Any]:
    """Run :func:`~executionkit.compose.pipe` with this Kit's provider.

    All keyword arguments are forwarded unchanged to :func:`pipe`.
    """
    return await self._run_tracked(pipe(self.provider, prompt, *steps, **kwargs))

react async

react(prompt: str, tools: Sequence[Tool], **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.react_loop.react_loop pattern.

All keyword arguments are forwarded unchanged to :func:react_loop. The provider must satisfy :class:~executionkit.provider.ToolCallingProvider; a :exc:TypeError is raised if it does not.

Source code in executionkit/kit.py

async def react(
    self, prompt: str, tools: Sequence[Tool], **kwargs: Any
) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.react_loop.react_loop` pattern.

    All keyword arguments are forwarded unchanged to :func:`react_loop`.
    The provider must satisfy :class:`~executionkit.provider.ToolCallingProvider`;
    a :exc:`TypeError` is raised if it does not.
    """
    provider = self.provider
    if not _provider_supports_tools(provider):
        msg = (
            f"react() requires a ToolCallingProvider; "
            f"{type(provider).__name__} does not support tool calling."
        )
        raise TypeError(msg)
    # _provider_supports_tools verified isinstance + supports_tools=True;
    # cast is safe here because mypy cannot narrow through the helper.
    tool_provider = cast("ToolCallingProvider", provider)
    return await self._run_tracked(
        react_loop(tool_provider, prompt, tools, **kwargs)
    )

refine async

refine(prompt: str, **kwargs: Any) -> PatternResult[str]

Run the :func:~executionkit.patterns.refine_loop.refine_loop pattern.

All keyword arguments are forwarded unchanged to :func:refine_loop.

Source code in executionkit/kit.py

async def refine(self, prompt: str, **kwargs: Any) -> PatternResult[str]:
    """Run the :func:`~executionkit.patterns.refine_loop.refine_loop` pattern.

    All keyword arguments are forwarded unchanged to :func:`refine_loop`.
    """
    return await self._run_tracked(refine_loop(self.provider, prompt, **kwargs))

stream_consensus async

stream_consensus(prompt: str, *, temperature: float = _STREAM_CONSENSUS_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, trace: TraceCallback | None = None) -> StreamingPatternResult

Stream a single live completion of prompt.

Consensus voting needs complete responses to compare, so the full pattern has no coherent token stream; this convenience method streams one generation (no voting). Token deltas arrive live and result.cost becomes accurate once the stream is drained, at which point the spend is folded into this Kit's cumulative :attr:usage.

Source code in executionkit/kit.py

async def stream_consensus(
    self,
    prompt: str,
    *,
    temperature: float = _STREAM_CONSENSUS_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    trace: TraceCallback | None = None,
) -> StreamingPatternResult:
    """Stream a single live completion of *prompt*.

    Consensus voting needs complete responses to compare, so the full
    pattern has no coherent token stream; this convenience method streams
    **one** generation (no voting).  Token deltas arrive live and
    ``result.cost`` becomes accurate once the stream is drained, at which
    point the spend is folded into this Kit's cumulative :attr:`usage`.
    """
    return await self._stream_single(
        prompt,
        temperature=temperature,
        max_tokens=max_tokens,
        max_cost=max_cost,
        trace=trace,
    )

stream_react_loop async

stream_react_loop(prompt: str, tools: Sequence[Tool] = (), *, temperature: float = _STREAM_REACT_TEMPERATURE, max_tokens: int = DEFAULT_MAX_TOKENS, max_cost: TokenUsage | None = None, trace: TraceCallback | None = None) -> StreamingPatternResult

Stream a single live model turn for prompt.

The full ReAct loop runs tools across multiple rounds and cannot be expressed as one token stream, so this convenience method streams a single model generation. tools is accepted for parity with :meth:react but is not executed (tool-call deltas carry no message content). result.cost is accurate after the stream drains and folds into this Kit's :attr:usage.

Source code in executionkit/kit.py

async def stream_react_loop(
    self,
    prompt: str,
    tools: Sequence[Tool] = (),
    *,
    temperature: float = _STREAM_REACT_TEMPERATURE,
    max_tokens: int = DEFAULT_MAX_TOKENS,
    max_cost: TokenUsage | None = None,
    trace: TraceCallback | None = None,
) -> StreamingPatternResult:
    """Stream a single live model turn for *prompt*.

    The full ReAct loop runs tools across multiple rounds and cannot be
    expressed as one token stream, so this convenience method streams a
    single model generation.  *tools* is accepted for parity with
    :meth:`react` but is **not** executed (tool-call deltas carry no
    message content).  ``result.cost`` is accurate after the stream drains
    and folds into this Kit's :attr:`usage`.
    """
    del tools  # accepted for parity with react(); not executed when streaming
    return await self._stream_single(
        prompt,
        temperature=temperature,
        max_tokens=max_tokens,
        max_cost=max_cost,
        trace=trace,
    )

turn async

turn(user_text: str, tools: Sequence[Tool] = (), **kwargs: Any) -> PatternResult[str]

Run one conversational turn, carrying history across calls.

Appends user_text to :attr:messages, runs :func:~executionkit.patterns.react_loop.react_loop over the full history, then replaces :attr:messages with the returned transcript so the next turn continues the conversation. Cost folds into :attr:usage, exactly like :meth:react.

The provider must satisfy :class:~executionkit.provider.ToolCallingProvider; a :exc:TypeError is raised otherwise. If the turn fails (e.g. budget exhausted), :attr:messages is left unchanged so the turn can be retried.

All other keyword arguments are forwarded unchanged to react_loop.

Source code in executionkit/kit.py

async def turn(
    self,
    user_text: str,
    tools: Sequence[Tool] = (),
    **kwargs: Any,
) -> PatternResult[str]:
    """Run one conversational turn, carrying history across calls.

    Appends *user_text* to :attr:`messages`, runs
    :func:`~executionkit.patterns.react_loop.react_loop` over the full
    history, then replaces :attr:`messages` with the returned transcript so
    the next ``turn`` continues the conversation. Cost folds into
    :attr:`usage`, exactly like :meth:`react`.

    The provider must satisfy
    :class:`~executionkit.provider.ToolCallingProvider`; a :exc:`TypeError`
    is raised otherwise. If the turn fails (e.g. budget exhausted),
    :attr:`messages` is left unchanged so the turn can be retried.

    All other keyword arguments are forwarded unchanged to ``react_loop``.
    """
    provider = self.provider
    if not _provider_supports_tools(provider):
        msg = (
            f"turn() requires a ToolCallingProvider; "
            f"{type(provider).__name__} does not support tool calling."
        )
        raise TypeError(msg)
    tool_provider = cast("ToolCallingProvider", provider)
    # Build the next history without mutating self.messages until the call
    # succeeds, so a failed turn does not leave a dangling user message.
    history = [*self.messages, user_message(user_text)]
    result = await self._run_tracked(
        react_loop(tool_provider, tools=tools, messages=history, **kwargs)
    )
    transcript = result.metadata.get("messages")
    self.messages = list(transcript) if transcript is not None else history
    return result

Kit tracks cumulative usage for calls made through its methods. Direct calls through kit.provider are not included.

Retry, pacing, and cost

executionkit.engine.retry.RetryConfig dataclass

RetryConfig(max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, exponential_base: float = 2.0, retryable: tuple[type[Exception], ...] = (RateLimitError, ProviderError), rate_limit_strategy: TokenBucket | None = None)

Immutable retry configuration with exponential backoff.

Attributes:

Name	Type	Description
max_retries	int	Maximum number of retry attempts after the initial call, so a call makes at most 1 + max_retries attempts in total. 0 means no retries — exactly one attempt.
base_delay	float	Base delay in seconds before first retry.
max_delay	float	Maximum delay cap in seconds.
exponential_base	float	Multiplier for exponential backoff.
retryable	tuple[type[Exception], ...]	Tuple of exception types that trigger retries.
rate_limit_strategy	TokenBucket | None	Optional TokenBucket strategy used to pace requests. When set, :func:with_retry acquires a token before each attempt and drains the bucket on RateLimitError so the provider's retry_after cooldown is honoured. None (default) preserves the original behaviour exactly.

__post_init__

__post_init__() -> None

Reject a negative retry budget at construction.

A negative value yields an empty attempt range, so with_retry never calls fn at all and falls through to its unreachable guard — a RuntimeError about the library's internals in answer to a bad argument. structured() already rejects the same input with a ValueError; this makes the two agree.

Source code in executionkit/engine/retry.py

def __post_init__(self) -> None:
    """Reject a negative retry budget at construction.

    A negative value yields an empty attempt range, so ``with_retry`` never
    calls ``fn`` at all and falls through to its ``unreachable`` guard —
    a RuntimeError about the library's internals in answer to a bad
    argument. ``structured()`` already rejects the same input with a
    ValueError; this makes the two agree.
    """
    if self.max_retries < 0:
        raise ValueError(f"max_retries must be >= 0, got {self.max_retries}")

get_delay

get_delay(attempt: int) -> float

Calculate jittered backoff delay for the given attempt (1-indexed).

Uses full jitter (random value in [0, capped_exponential]) to prevent thundering-herd effects when multiple coroutines retry simultaneously.

Source code in executionkit/engine/retry.py

def get_delay(self, attempt: int) -> float:
    """Calculate jittered backoff delay for the given attempt (1-indexed).

    Uses full jitter (random value in [0, capped_exponential]) to prevent
    thundering-herd effects when multiple coroutines retry simultaneously.
    """
    cap = min(
        self.base_delay * (self.exponential_base ** (attempt - 1)),
        self.max_delay,
    )
    return random.uniform(0.0, cap)  # noqa: S311

should_retry

should_retry(exc: Exception) -> bool

Check whether the given exception is retryable.

Source code in executionkit/engine/retry.py

def should_retry(self, exc: Exception) -> bool:
    """Check whether the given exception is retryable."""
    return isinstance(exc, self.retryable)

DEFAULT_RETRY is:

RetryConfig(
    max_retries=3,
    base_delay=1.0,
    max_delay=60.0,
    exponential_base=2.0,
)

max_retries counts calls after the initial attempt. The default therefore allows at most four dispatched attempts for a retryable failure.

executionkit.engine.rate_bucket.TokenBucket dataclass

TokenBucket(rate: float, capacity: float)

Coroutine-paced token bucket with Retry-After penalty support.

Attributes:

Name	Type	Description
rate	float	Tokens refilled per second (sustained throughput). Must be > 0.
capacity	float	Maximum tokens held — the burst size. Must be >= 1.0 so a single-token :meth:acquire can always eventually succeed.

acquire async

acquire() -> None

Block until one token is available, then consume it.

Honours any active penalty window (see :meth:drain) before refilling, so a prior Retry-After cooldown is always respected first.

Source code in executionkit/engine/rate_bucket.py

async def acquire(self) -> None:
    """Block until one token is available, then consume it.

    Honours any active penalty window (see :meth:`drain`) before refilling,
    so a prior ``Retry-After`` cooldown is always respected first.
    """
    while True:
        now = time.monotonic()
        if now < self._penalty_until:
            await asyncio.sleep(self._penalty_until - now)
            continue
        self._refill()
        if self._tokens >= 1.0:
            self._tokens -= 1.0
            return
        wait = (1.0 - self._tokens) / self.rate
        await asyncio.sleep(wait)

drain

drain(seconds: float) -> None

Force-drain the bucket as if the provider rate-limited us for seconds.

Called immediately after receiving a :class:~executionkit.errors.RateLimitError with retry_after. Removes seconds * rate tokens (floored at zero) and arms a penalty deadline seconds into the future so the next :meth:acquire waits out the full cooldown. last_refill is advanced so the refill calculation on the next :meth:acquire does not immediately undo the drain.

Source code in executionkit/engine/rate_bucket.py

def drain(self, seconds: float) -> None:
    """Force-drain the bucket as if the provider rate-limited us for ``seconds``.

    Called immediately after receiving a
    :class:`~executionkit.errors.RateLimitError` with ``retry_after``.
    Removes ``seconds * rate`` tokens (floored at zero) *and* arms a penalty
    deadline ``seconds`` into the future so the next :meth:`acquire` waits out
    the full cooldown.  ``last_refill`` is advanced so the refill calculation
    on the next :meth:`acquire` does not immediately undo the drain.
    """
    now = time.monotonic()
    self._tokens = max(0.0, self._tokens - seconds * self.rate)
    self._last_refill = now
    # Extend (never shorten) the penalty so repeated 429s accumulate cooldown.
    self._penalty_until = max(self._penalty_until, now + seconds)

executionkit.cost.CostTracker

CostTracker()

Mutable accumulator for token and call counts.

Source code in executionkit/cost.py

def __init__(self) -> None:
    self._input: int = 0
    self._output: int = 0
    self._calls: int = 0

call_count property

call_count: int

Number of LLM calls recorded so far.

total_tokens property

total_tokens: int

Total input + output tokens recorded so far.

add_usage

add_usage(usage: TokenUsage) -> None

Add pre-computed usage to the tracker (e.g. from a pattern result).

Use this instead of accessing private fields directly.

Source code in executionkit/cost.py

def add_usage(self, usage: TokenUsage) -> None:
    """Add pre-computed usage to the tracker (e.g. from a pattern result).

    Use this instead of accessing private fields directly.
    """
    self._input += usage.input_tokens
    self._output += usage.output_tokens
    self._calls += usage.llm_calls

record

record(response: LLMResponse) -> None

Record usage from a single LLM response.

Source code in executionkit/cost.py

def record(self, response: LLMResponse) -> None:
    """Record usage from a single LLM response."""
    self._input += response.input_tokens
    self._output += response.output_tokens
    self._calls += 1

record_without_call

record_without_call(response: LLMResponse) -> None

Record token usage from a response without incrementing the call counter.

Used by :func:~executionkit.patterns.base.checked_complete because :meth:reserve_call (called from _before_attempt) pre-increments _calls before the await to uphold the asyncio concurrency contract described in the module-level docstring. Token totals are therefore recorded here, after the provider returns, without a second increment to the call counter.

Source code in executionkit/cost.py

def record_without_call(self, response: LLMResponse) -> None:
    """Record token usage from a response without incrementing the call counter.

    Used by :func:`~executionkit.patterns.base.checked_complete` because
    :meth:`reserve_call` (called from ``_before_attempt``) pre-increments
    ``_calls`` before the ``await`` to uphold the asyncio concurrency
    contract described in the module-level docstring.  Token totals are
    therefore recorded here, after the provider returns, without a second
    increment to the call counter.
    """
    self._input += response.input_tokens
    self._output += response.output_tokens

reserve_call

reserve_call() -> None

Reserve a call slot before dispatching (for budget-safe accounting).

Called by :func:~executionkit.patterns.base.checked_complete before awaiting the provider call. Reservations intentionally count every wire attempt, including failed attempts, so call budgets cap retries as well as successes.

Concurrency contract: this method must be called in the same synchronous run-segment as the preceding budget check — with no await between them — to prevent concurrent asyncio coroutines from racing past the check. See module-level docstring for details. This guarantee does NOT hold under threading.

Source code in executionkit/cost.py

def reserve_call(self) -> None:
    """Reserve a call slot before dispatching (for budget-safe accounting).

    Called by :func:`~executionkit.patterns.base.checked_complete` before
    awaiting the provider call.  Reservations intentionally count every
    wire attempt, including failed attempts, so call budgets cap retries
    as well as successes.

    **Concurrency contract**: this method must be called in the same
    synchronous run-segment as the preceding budget check — with no
    ``await`` between them — to prevent concurrent asyncio coroutines
    from racing past the check.  See module-level docstring for details.
    This guarantee does NOT hold under threading.
    """
    self._calls += 1

snapshot

snapshot() -> TokenUsage

Return current totals as an immutable snapshot without mutating state.

Equivalent to :meth:to_usage; provided as the public, intention-revealing name for callers that capture point-in-time deltas (e.g. per-step cost attribution or loop checkpoint state). Calling it never advances any counter, so two consecutive snapshots with no intervening record are equal.

Source code in executionkit/cost.py

def snapshot(self) -> TokenUsage:
    """Return current totals as an immutable snapshot without mutating state.

    Equivalent to :meth:`to_usage`; provided as the public, intention-revealing
    name for callers that capture point-in-time deltas (e.g. per-step cost
    attribution or loop checkpoint state).  Calling it never advances any
    counter, so two consecutive snapshots with no intervening ``record`` are
    equal.
    """
    return self.to_usage()

to_usage

to_usage() -> TokenUsage

Return an immutable snapshot of accumulated usage.

Source code in executionkit/cost.py

def to_usage(self) -> TokenUsage:
    """Return an immutable snapshot of accumulated usage."""
    return TokenUsage(
        input_tokens=self._input,
        output_tokens=self._output,
        llm_calls=self._calls,
    )

executionkit.cost.estimate_cost

estimate_cost(usage: TokenUsage, *, input_rate: float, output_rate: float) -> float

Estimate the dollar cost of a :class:~executionkit.types.TokenUsage snapshot.

ExecutionKit ships no price table — rates change and differ by account tier. Pass per-token rates from your provider's pricing page.

Parameters:

Name	Type	Description	Default
usage	TokenUsage	Token-usage snapshot (e.g. from :meth:CostTracker.to_usage).	required
input_rate	float	Cost per input token (e.g. 3e-6 for $3 per million). Assumed non-negative and finite; no validation is performed.	required
output_rate	float	Cost per output token (e.g. 15e-6 for $15 per million). Assumed non-negative and finite; no validation is performed.	required

Returns:

Type	Description
float	Estimated cost in the same currency unit as the supplied rates.

Example::

tracker = CostTracker()
# … run some patterns …
cost = estimate_cost(tracker.to_usage(), input_rate=3e-6, output_rate=15e-6)
print(f"Estimated cost: ${cost:.4f}")

Source code in executionkit/cost.py

def estimate_cost(
    usage: TokenUsage,
    *,
    input_rate: float,
    output_rate: float,
) -> float:
    """Estimate the dollar cost of a :class:`~executionkit.types.TokenUsage` snapshot.

    ExecutionKit ships **no price table** — rates change and differ by account
    tier.  Pass per-token rates from your provider's pricing page.

    Args:
        usage: Token-usage snapshot (e.g. from :meth:`CostTracker.to_usage`).
        input_rate: Cost per *input* token (e.g. ``3e-6`` for $3 per million).
            Assumed non-negative and finite; no validation is performed.
        output_rate: Cost per *output* token (e.g. ``15e-6`` for $15 per million).
            Assumed non-negative and finite; no validation is performed.

    Returns:
        Estimated cost in the same currency unit as the supplied rates.

    Example::

        tracker = CostTracker()
        # … run some patterns …
        cost = estimate_cost(tracker.to_usage(), input_rate=3e-6, output_rate=15e-6)
        print(f"Estimated cost: ${cost:.4f}")
    """
    return usage.input_tokens * input_rate + usage.output_tokens * output_rate

executionkit.engine.convergence.ConvergenceDetector dataclass

ConvergenceDetector(delta_threshold: float = 0.01, patience: int = 3, score_threshold: float | None = None)

Tracks score history and detects convergence via delta + patience.

Convergence is declared when either: - score_threshold is set and the current score meets or exceeds it, or - The score delta has been below delta_threshold for patience consecutive iterations.

Attributes:

Name	Type	Description
delta_threshold	float	Minimum meaningful score improvement.
patience	int	How many consecutive stale iterations before stopping.
score_threshold	float | None	Optional absolute score target for early exit.

reset

reset() -> None

Clear all tracked state.

Source code in executionkit/engine/convergence.py

def reset(self) -> None:
    """Clear all tracked state."""
    self._scores.clear()
    self._stale_count = 0

should_stop

should_stop(score: float) -> bool

Record a score and return whether convergence is reached.

Parameters:

Name	Type	Description	Default
score	float	Evaluator score, must be in [0.0, 1.0] and not NaN.	required

Returns:

Type	Description
bool	True if the loop should stop (converged or threshold met).

Raises:

Type	Description
ValueError	If score is NaN or outside [0.0, 1.0].

Source code in executionkit/engine/convergence.py

def should_stop(self, score: float) -> bool:
    """Record a score and return whether convergence is reached.

    Args:
        score: Evaluator score, must be in [0.0, 1.0] and not NaN.

    Returns:
        True if the loop should stop (converged or threshold met).

    Raises:
        ValueError: If score is NaN or outside [0.0, 1.0].
    """
    if math.isnan(score) or not (0.0 <= score <= 1.0):
        raise ValueError(f"Invalid score: {score}")

    self._scores.append(score)

    # Absolute threshold check
    if self.score_threshold is not None and score >= self.score_threshold:
        return True

    # Delta-based convergence check (need at least 2 scores)
    if len(self._scores) >= 2:
        delta = abs(self._scores[-1] - self._scores[-2])
        if delta <= self.delta_threshold:
            self._stale_count += 1
        else:
            self._stale_count = 0

        if self._stale_count >= self.patience:
            return True

    return False

CostTracker is mutable and is safe for the package's asyncio accounting sequence. It is not thread-safe.

Routing

executionkit.routing.RouteRule dataclass

RouteRule(name: str, provider: LLMProvider, predicate: RoutePredicate, metadata: Mapping[str, Any] = dict())

A named provider selection rule.

executionkit.routing.Router

Router(*, rules: Sequence[RouteRule], fallback: LLMProvider)

Select a provider by evaluating rules before a pattern call.

Source code in executionkit/routing.py

def __init__(self, *, rules: Sequence[RouteRule], fallback: LLMProvider) -> None:
    self.rules = tuple(rules)
    self.fallback = fallback

run async

run(pattern: RoutedPattern, prompt: str, *, context: Mapping[str, Any] | None = None, **kwargs: Any) -> PatternResult[Any]

Select a provider from context, then call pattern with it.

Routing inputs are passed explicitly via context and forwarded only to the route predicates; **kwargs are forwarded only to pattern. Keeping the two disjoint stops routing keys (e.g. tier) from leaking into the pattern call — which would raise TypeError for any pattern that does not declare them.

Source code in executionkit/routing.py

async def run(
    self,
    pattern: RoutedPattern,
    prompt: str,
    *,
    context: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> PatternResult[Any]:
    """Select a provider from *context*, then call *pattern* with it.

    Routing inputs are passed explicitly via *context* and forwarded only to
    the route predicates; ``**kwargs`` are forwarded only to *pattern*.
    Keeping the two disjoint stops routing keys (e.g. ``tier``) from leaking
    into the pattern call — which would raise ``TypeError`` for any pattern
    that does not declare them.
    """
    # Drop a stray "prompt" key so it cannot collide with the positional
    # ``prompt`` argument to ``select`` (``select(prompt, prompt=...)`` would
    # raise TypeError). The predicate still receives the real prompt
    # positionally, so nothing is lost.
    route_context = {
        key: value for key, value in (context or {}).items() if key != "prompt"
    }
    provider = self.select(prompt, **route_context)
    return await pattern(provider, prompt, **kwargs)

Workflow

executionkit.workflow.Step dataclass

Step(name: str, run: WorkflowRun, depends_on: tuple[str, ...] = (), metadata: Mapping[str, Any] = dict())

A named workflow step with optional dependencies.

executionkit.workflow.Workflow

Workflow(steps: Sequence[Step])

Run named steps once their dependencies are available.

Source code in executionkit/workflow.py

def __init__(self, steps: Sequence[Step]) -> None:
    self.steps = tuple(steps)
    self._validate()

run async

run(initial_context: Mapping[str, Any] | None = None, *, trace: TraceCallback | None = None, approval_gate: ApprovalGate | None = None, checkpoint_fn: CheckpointFn | None = None, resume_from: WorkflowCheckpoint | None = None) -> WorkflowResult

Execute the workflow, optionally checkpointing after each batch.

Parameters

initial_context: Key/value pairs added to the step context before execution. Keys must not match workflow step names. trace: Optional async or sync callback receiving :class:~executionkit.observability.TraceEvent objects. approval_gate: Optional gate consulted before each step runs. checkpoint_fn: Called with a :class:WorkflowCheckpoint after each batch of completed steps. The caller is responsible for persisting the checkpoint; this library imposes no storage requirement. resume_from: A previously saved :class:WorkflowCheckpoint. Steps whose names appear in resume_from.outputs are skipped. Accumulated outputs and token usage are restored. When None, execution starts from the beginning.

Raises

ValueError If a key in initial_context collides with the name of one of this workflow's steps.

Source code in executionkit/workflow.py

async def run(
    self,
    initial_context: Mapping[str, Any] | None = None,
    *,
    trace: TraceCallback | None = None,
    approval_gate: ApprovalGate | None = None,
    checkpoint_fn: CheckpointFn | None = None,
    resume_from: WorkflowCheckpoint | None = None,
) -> WorkflowResult:
    """Execute the workflow, optionally checkpointing after each batch.

    Parameters
    ----------
    initial_context:
        Key/value pairs added to the step context before execution. Keys
        must not match workflow step names.
    trace:
        Optional async or sync callback receiving
        :class:`~executionkit.observability.TraceEvent` objects.
    approval_gate:
        Optional gate consulted before each step runs.
    checkpoint_fn:
        Called with a :class:`WorkflowCheckpoint` after each batch of
        completed steps.  The caller is responsible for persisting the
        checkpoint; this library imposes no storage requirement.
    resume_from:
        A previously saved :class:`WorkflowCheckpoint`. Steps whose names
        appear in ``resume_from.outputs`` are skipped. Accumulated outputs
        and token usage are restored. When ``None``, execution starts from
        the beginning.

    Raises
    ------
    ValueError
        If a key in ``initial_context`` collides with the name of one of
        this workflow's steps.
    """
    if initial_context is not None:
        step_names = {step.name for step in self.steps}
        colliding = sorted(set(initial_context) & step_names)
        if colliding:
            raise ValueError(
                "initial_context key(s) collide with workflow step name(s): "
                f"{colliding}. A colliding key would be silently returned as "
                "that step's output without the step ever running — rename "
                "the initial_context key(s) or the step(s)."
            )

    # ------------------------------------------------------------------
    # Restore state from checkpoint (if any)
    # ------------------------------------------------------------------
    if resume_from is not None:
        outputs: dict[str, Any] = dict(resume_from.outputs)
        # A checkpoint's outputs genuinely mean "already executed" — this
        # is the one case where dict membership is a valid completion
        # signal, since resume_from only ever holds prior step results.
        completed: set[str] = set(resume_from.outputs)
        total_cost = resume_from.cost
    else:
        outputs = dict(initial_context or {})
        # Fresh run: nothing has executed yet, regardless of what
        # initial_context seeded into `outputs`. Completion is tracked
        # independently of `outputs` membership so a seed can never be
        # mistaken for a completed step (the bug this fix closes).
        completed = set()
        total_cost = TokenUsage()

    pending = {step.name: step for step in self.steps if step.name not in completed}
    completed_count = len(self.steps) - len(pending)

    while pending:
        ready = [
            step
            for step in pending.values()
            if all(dep in outputs for dep in step.depends_on)
        ]
        if not ready:
            raise ExecutionKitError("Workflow dependencies could not be resolved")

        results = await gather_strict(
            [
                self._run_step(
                    step,
                    outputs,
                    trace=trace,
                    approval_gate=approval_gate,
                )
                for step in ready
            ]
        )
        for step, output in zip(ready, results, strict=True):
            if isinstance(output, PatternResult):
                total_cost += output.cost
                outputs[step.name] = output.value
            else:
                outputs[step.name] = output
            pending.pop(step.name)

        completed_count += len(ready)

        if checkpoint_fn is not None:
            maybe_checkpoint = checkpoint_fn(
                WorkflowCheckpoint(
                    step_index=completed_count,
                    outputs=MappingProxyType(dict(outputs)),
                    cost=total_cost,
                )
            )
            if inspect.isawaitable(maybe_checkpoint):
                await maybe_checkpoint

    return WorkflowResult(outputs=MappingProxyType(outputs), cost=total_cost)

executionkit.workflow.WorkflowCheckpoint dataclass

WorkflowCheckpoint(step_index: int, outputs: MappingProxyType[str, Any], cost: TokenUsage)

Snapshot of workflow state after a completed step batch.

Captures the step index (number of completed steps), accumulated outputs, and token budget consumed so far. Both to_dict and from_dict use only plain Python types so the caller can serialise with any backend (JSON, pickle, database, etc.) without a dependency on this module.

cost instance-attribute

cost: TokenUsage

Aggregate token usage consumed up to this checkpoint.

outputs instance-attribute

outputs: MappingProxyType[str, Any]

Accumulated step outputs, keyed by step name.

step_index instance-attribute

step_index: int

Number of steps that have been completed.

from_dict classmethod

from_dict(data: dict[str, Any]) -> WorkflowCheckpoint

Restore a :class:WorkflowCheckpoint from a plain-Python dict.

Raises

ValueError If the dict is missing a required key or contains a value of the wrong type. The message names the exact missing or malformed field so that callers can produce actionable diagnostics.

Source code in executionkit/workflow.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> WorkflowCheckpoint:
    """Restore a :class:`WorkflowCheckpoint` from a plain-Python dict.

    Raises
    ------
    ValueError
        If the dict is missing a required key or contains a value of the
        wrong type.  The message names the exact missing or malformed field
        so that callers can produce actionable diagnostics.
    """
    required_top: tuple[str, ...] = ("step_index", "outputs", "cost")
    required_cost: tuple[str, ...] = ("input_tokens", "output_tokens", "llm_calls")

    for key in required_top:
        if key not in data:
            raise ValueError(
                f"WorkflowCheckpoint.from_dict: missing required key {key!r}"
            )

    if not isinstance(data["cost"], dict):
        raise ValueError(
            "WorkflowCheckpoint.from_dict: 'cost' must be a dict, "
            f"got {type(data['cost']).__name__!r}"
        )
    cost_data: dict[str, Any] = data["cost"]

    for key in required_cost:
        if key not in cost_data:
            raise ValueError(
                f"WorkflowCheckpoint.from_dict: missing required key 'cost.{key}'"
            )

    if not isinstance(data["outputs"], dict):
        raise ValueError(
            "WorkflowCheckpoint.from_dict: 'outputs' must be a dict, "
            f"got {type(data['outputs']).__name__!r}"
        )

    return cls(
        step_index=int(data["step_index"]),
        outputs=MappingProxyType(dict(data["outputs"])),
        cost=TokenUsage(
            input_tokens=int(cost_data["input_tokens"]),
            output_tokens=int(cost_data["output_tokens"]),
            llm_calls=int(cost_data["llm_calls"]),
        ),
    )

to_dict

to_dict() -> dict[str, Any]

Serialise to a plain-Python dict.

JSON-serializability depends on the caller ensuring that all step outputs stored in this checkpoint are themselves JSON-safe primitives. Non-primitive output values (e.g. custom objects, bytes) will cause json.dumps to raise when applied to the returned dict.

Source code in executionkit/workflow.py

def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain-Python dict.

    JSON-serializability depends on the caller ensuring that all step
    outputs stored in this checkpoint are themselves JSON-safe primitives.
    Non-primitive output values (e.g. custom objects, bytes) will cause
    ``json.dumps`` to raise when applied to the returned dict.
    """
    return {
        "step_index": self.step_index,
        "outputs": dict(self.outputs),
        "cost": {
            "input_tokens": self.cost.input_tokens,
            "output_tokens": self.cost.output_tokens,
            "llm_calls": self.cost.llm_calls,
        },
    }

executionkit.workflow.WorkflowResult dataclass

WorkflowResult(outputs: MappingProxyType[str, Any], cost: TokenUsage = TokenUsage())

Outputs and aggregate cost from a workflow run.

Planning

executionkit.planning.PlanStep dataclass

PlanStep(name: str, instruction: str, run: PlanRun, metadata: Mapping[str, Any] = dict())

A human-readable executable plan step.

executionkit.planning.Plan

Plan(steps: Sequence[PlanStep])

Execute named plan steps in order.

Source code in executionkit/planning.py

def __init__(self, steps: Sequence[PlanStep]) -> None:
    names = [step.name for step in steps]
    if len(names) != len(set(names)):
        raise ValueError("Plan step names must be unique")
    self.steps = tuple(steps)

executionkit.planning.PlanResult dataclass

PlanResult(outputs: MappingProxyType[str, Any], cost: TokenUsage = TokenUsage())

Outputs and aggregate cost from plan execution.

Approval

executionkit.approval.ApprovalRequest dataclass

ApprovalRequest(action: str, subject: str, metadata: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))())

A request to approve an operation before it runs.

executionkit.approval.ApprovalDecision dataclass

ApprovalDecision(approved: bool, reason: str = '')

Approval result.

executionkit.approval.ApprovalGate

ApprovalGate(callback: ApprovalCallback, *, timeout_seconds: float | None = None, on_timeout: Literal['approve', 'deny', 'raise'] = 'raise')

Async-compatible approval callback wrapper.

Parameters

callback: Synchronous or async callable that receives an :class:ApprovalRequest and returns an :class:ApprovalDecision (or a truthy/falsy value). timeout_seconds: If given, the callback must resolve within this many seconds. When it does not, on_timeout controls the fallback behaviour. None (default) preserves the original blocking behaviour. on_timeout: What to do when timeout_seconds elapses before the callback resolves.

* ``"raise"`` (default) — raise :class:`ApprovalTimeoutError`.
* ``"approve"`` — treat the timed-out request as approved.
* ``"deny"``   — treat the timed-out request as denied.

.. warning:: Setting on_timeout="approve" creates a fail-open gate: if the approval callback does not respond in time, the operation is automatically permitted. This is a potential privilege-escalation vector. Only use this option when you have deliberately decided that availability is more important than security for the guarded operation.

Source code in executionkit/approval.py

def __init__(
    self,
    callback: ApprovalCallback,
    *,
    timeout_seconds: float | None = None,
    on_timeout: Literal["approve", "deny", "raise"] = "raise",
) -> None:
    if on_timeout == "approve":
        warnings.warn(
            self._APPROVE_ON_TIMEOUT_WARNING,
            UserWarning,
            stacklevel=2,
        )
    self._callback = callback
    self._timeout_seconds = timeout_seconds
    self._on_timeout = on_timeout

executionkit.approval.ApprovalDeniedError

ApprovalDeniedError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: ExecutionKitError

Raised when an approval-gated operation is denied.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.approval.ApprovalTimeoutError

ApprovalTimeoutError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: ExecutionKitError

Raised when an approval gate times out and on_timeout is "raise".

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

Tracing

executionkit.observability.TraceEvent dataclass

TraceEvent(kind: str, payload: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))())

A structured event emitted during pattern and coordination calls.

TraceCallback accepts a TraceEvent and may return None or an awaitable.

executionkit.observability.emit_trace async

emit_trace(trace: TraceCallback | None, event: TraceEvent) -> None

Emit event to an optional callback.

Source code in executionkit/observability.py

async def emit_trace(trace: TraceCallback | None, event: TraceEvent) -> None:
    """Emit *event* to an optional callback."""

    if trace is None:
        return
    maybe_awaitable = trace(event)
    if inspect.isawaitable(maybe_awaitable):
        await maybe_awaitable

The lower-level OpenTelemetry helpers live in executionkit.observability. They are not package-root exports:

executionkit.observability.llm_span

llm_span(model: str) -> Generator[Any, None, None]

Context manager that wraps a single LLM call in an OTel span.

When opentelemetry-api is not installed the context manager is a no-op — the yielded value is None and no tracing infrastructure is touched. The call path is identical in both cases so code that checks for OTel availability lives only here.

Usage::

with llm_span(model) as span:
    response = await provider.complete(messages)
    record_llm_span_attributes(span, model, response)

Parameters:

Name	Type	Description	Default
model	str	The model identifier string (e.g. "gpt-4o").	required

Yields:

Type	Description
Any	An OTel Span when the SDK is available, None otherwise.

Source code in executionkit/observability.py

@contextmanager
def llm_span(model: str) -> Generator[Any, None, None]:
    """Context manager that wraps a single LLM call in an OTel span.

    When ``opentelemetry-api`` is **not** installed the context manager is a
    no-op — the yielded value is ``None`` and no tracing infrastructure is
    touched.  The call path is identical in both cases so code that checks
    for OTel availability lives only here.

    Usage::

        with llm_span(model) as span:
            response = await provider.complete(messages)
            record_llm_span_attributes(span, model, response)

    Args:
        model: The model identifier string (e.g. ``"gpt-4o"``).

    Yields:
        An OTel ``Span`` when the SDK is available, ``None`` otherwise.
    """
    if not _OTEL_AVAILABLE:
        yield None
        return

    # Local import: only reached when opentelemetry-api is installed.
    from opentelemetry import trace as otel_trace

    tracer = otel_trace.get_tracer("executionkit")
    with tracer.start_as_current_span(LLM_CALL_SPAN_NAME) as span:
        span.set_attribute("llm.model", model)
        yield span

executionkit.observability.record_llm_span_attributes

record_llm_span_attributes(span: Any, model: str, input_tokens: int, output_tokens: int, cost_usd: float | None = None) -> None

Set standard LLM attributes on an active OTel span.

This is a no-op when span is None (OTel absent) or when the span is non-recording (e.g. a sampled-out or no-op span from the SDK).

Attributes set:

• llm.model — already set by :func:llm_span; repeated here for completeness if callers construct spans themselves.
• llm.input_tokens — prompt / input token count.
• llm.output_tokens — completion / output token count.
• cost_usd — estimated dollar cost; omitted when None so we never emit a fabricated value when no rate table is available.

Parameters:

Name	Type	Description	Default
span	Any	The span returned by the :func:llm_span context manager, or None when OTel is absent.	required
model	str	Model identifier string.	required
input_tokens	int	Number of input (prompt) tokens consumed.	required
output_tokens	int	Number of output (completion) tokens produced.	required
cost_usd	float | None	Optional pre-computed cost in USD. Pass None to omit.	None

Source code in executionkit/observability.py

def record_llm_span_attributes(
    span: Any,
    model: str,
    input_tokens: int,
    output_tokens: int,
    cost_usd: float | None = None,
) -> None:
    """Set standard LLM attributes on an active OTel span.

    This is a no-op when *span* is ``None`` (OTel absent) or when the span
    is non-recording (e.g. a sampled-out or no-op span from the SDK).

    Attributes set:

    * ``llm.model`` — already set by :func:`llm_span`; repeated here for
      completeness if callers construct spans themselves.
    * ``llm.input_tokens`` — prompt / input token count.
    * ``llm.output_tokens`` — completion / output token count.
    * ``cost_usd`` — estimated dollar cost; **omitted** when ``None`` so we
      never emit a fabricated value when no rate table is available.

    Args:
        span: The span returned by the :func:`llm_span` context manager, or
            ``None`` when OTel is absent.
        model: Model identifier string.
        input_tokens: Number of input (prompt) tokens consumed.
        output_tokens: Number of output (completion) tokens produced.
        cost_usd: Optional pre-computed cost in USD.  Pass ``None`` to omit.
    """
    if span is None:
        return

    # Skip non-recording spans (sampled-out / no-op). ``is_recording()`` is the
    # public OTel API and avoids importing SDK-internal span classes — which
    # also keeps this module import-clean when opentelemetry is not installed.
    if hasattr(span, "is_recording") and not span.is_recording():
        return

    span.set_attribute("llm.model", model)
    span.set_attribute("llm.input_tokens", input_tokens)
    span.set_attribute("llm.output_tokens", output_tokens)
    if cost_usd is not None:
        span.set_attribute("cost_usd", cost_usd)

Exceptions

All package exceptions derive from ExecutionKitError. Each instance carries cost and metadata.

ExecutionKitError
├── LLMError
│   ├── RateLimitError
│   ├── PermanentError
│   └── ProviderError
└── PatternError
    ├── BudgetExhaustedError
    ├── ConsensusFailedError
    └── MaxIterationsError

ApprovalDeniedError and ApprovalTimeoutError also derive from ExecutionKitError.

executionkit.errors.ExecutionKitError

ExecutionKitError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: Exception

Base exception for all ExecutionKit errors.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.LLMError

LLMError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: ExecutionKitError

Errors originating from LLM provider communication.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.RateLimitError

RateLimitError(message: str, *, retry_after: float = 1.0, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: LLMError

Provider returned HTTP 429 — retryable after retry_after seconds.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    retry_after: float = 1.0,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message, cost=cost, metadata=metadata)
    self.retry_after: float = retry_after

executionkit.errors.PermanentError

PermanentError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: LLMError

Non-retryable provider error (e.g. 401 authentication failure).

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.ProviderError

ProviderError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: LLMError

Catch-all retryable provider error for unexpected HTTP failures.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.PatternError

PatternError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: ExecutionKitError

Errors raised by reasoning pattern logic.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.BudgetExhaustedError

BudgetExhaustedError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: PatternError

Token or call budget exceeded.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.ConsensusFailedError

ConsensusFailedError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: PatternError

Consensus pattern could not reach agreement.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}

executionkit.errors.MaxIterationsError

MaxIterationsError(message: str, *, cost: TokenUsage | None = None, metadata: dict[str, Any] | None = None)

Bases: PatternError

Loop pattern exceeded its iteration limit.

Source code in executionkit/errors.py

def __init__(
    self,
    message: str,
    *,
    cost: TokenUsage | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    super().__init__(message)
    self.cost: TokenUsage = cost if cost is not None else TokenUsage()
    self.metadata: dict[str, Any] = metadata if metadata is not None else {}