zhivex_ai.workflows

Public import paths and stability come from the installed artifact.

Use the root for Stable APIs and focused namespaces for extensions. Legacy root extension imports remain classified below.

CallbackWorkflowAdapter

Stable · workflow

from zhivex_ai.workflows import CallbackWorkflowAdapter

CallbackWorkflowAdapter(backend: 'str', callback: 'WorkflowStepExecutor', capabilities: 'WorkflowAdapterCapabilities' = <factory>) -> None

CallbackWorkflowAdapter(backend: 'str', callback: 'WorkflowStepExecutor', capabilities: 'WorkflowAdapterCapabilities' = <factory>)

GraphWorkflow

Stable · workflow

from zhivex_ai.workflows import GraphWorkflow

GraphWorkflow(*, name: 'str', steps: 'Sequence[WorkflowStep]', edges: 'Sequence[WorkflowEdge]', definition_version: 'str' = '1', entrypoints: 'Sequence[str] | None' = None, checkpoint_store: 'WorkflowCheckpointStore | None' = None, run_store: 'AgentRunStore | None' = None, adapter: 'WorkflowAdapter | None' = None, interrupt_before: 'Mapping[str, str | None] | Sequence[str]' = (), interrupt_after: 'Mapping[str, str | None] | Sequence[str]' = (), max_concurrency: 'int | None' = None, lease_manager: 'WorkflowLeaseManager | None' = None, lease_ttl_ms: 'int' = 30000, lease_heartbeat_ms: 'int | None' = None, observer: 'AgentObserver | None' = None) -> 'None'

Durable DAG workflow with append-only checkpoints and explicit resume.

InMemoryWorkflowCheckpointStore

Stable · workflow

from zhivex_ai.workflows import InMemoryWorkflowCheckpointStore

InMemoryWorkflowCheckpointStore() -> 'None'

InMemoryWorkflowLeaseManager

Stable · workflow

from zhivex_ai.workflows import InMemoryWorkflowLeaseManager

InMemoryWorkflowLeaseManager() -> 'None'

LoopAgent

Stable · workflow

from zhivex_ai.workflows import LoopAgent

LoopAgent(*, name: 'str', steps: 'Sequence[WorkflowStep]', max_iterations: 'int', stop_condition: 'WorkflowStopCondition | None' = None, run_store: 'AgentRunStore | None' = None) -> 'None'

ParallelAgent

Stable · workflow

from zhivex_ai.workflows import ParallelAgent

ParallelAgent(*, name: 'str', steps: 'Sequence[WorkflowStep]', run_store: 'AgentRunStore | None' = None) -> 'None'

PostgresWorkflowCheckpointStore

Stable · workflow

from zhivex_ai.workflows import PostgresWorkflowCheckpointStore

PostgresWorkflowCheckpointStore(dsn: 'str | None' = None, *, table_prefix: 'str' = 'zhivex_ai', namespace: 'str' = 'default', pool: 'Any | None' = None, pool_min_size: 'int' = 1, pool_max_size: 'int' = 5) -> 'None'

PostgresWorkflowLeaseManager

Stable · workflow

from zhivex_ai.workflows import PostgresWorkflowLeaseManager

PostgresWorkflowLeaseManager(dsn: 'str | None' = None, *, table_prefix: 'str' = 'zhivex_ai', namespace: 'str' = 'default', pool: 'Any | None' = None, pool_min_size: 'int' = 1, pool_max_size: 'int' = 5) -> 'None'

SQLiteWorkflowCheckpointStore

Stable · workflow

from zhivex_ai.workflows import SQLiteWorkflowCheckpointStore

SQLiteWorkflowCheckpointStore(path: 'str', *, namespace: 'str' = 'default') -> 'None'

SQLiteWorkflowLeaseManager

Stable · workflow

from zhivex_ai.workflows import SQLiteWorkflowLeaseManager

SQLiteWorkflowLeaseManager(path: 'str', *, namespace: 'str' = 'default') -> 'None'

SequentialAgent

Stable · workflow

from zhivex_ai.workflows import SequentialAgent

SequentialAgent(*, name: 'str', steps: 'Sequence[WorkflowStep]', run_store: 'AgentRunStore | None' = None) -> 'None'

WORKFLOW_ADAPTER_SCHEMA_VERSION

Stable · workflow

from zhivex_ai.workflows import WORKFLOW_ADAPTER_SCHEMA_VERSION

WORKFLOW_ADAPTER_SCHEMA_VERSION (public type alias or constant; no callable signature)

WORKFLOW_CHECKPOINT_SCHEMA_VERSION

Stable · workflow

from zhivex_ai.workflows import WORKFLOW_CHECKPOINT_SCHEMA_VERSION

WORKFLOW_CHECKPOINT_SCHEMA_VERSION (public type alias or constant; no callable signature)

WorkflowAdapter

Stable · workflow

from zhivex_ai.workflows import WorkflowAdapter

WorkflowAdapter(*args, **kwargs)

Base class for protocol classes.

Protocol classes are defined as::

    class Proto(Protocol):
        def meth(self) -> int:
            ...

Such classes are primarily used with static type checkers that recognize
structural subtyping (static duck-typing).

For example::

    class C:
        def meth(self) -> int:
            return 0

    def func(x: Proto) -> int:
        return x.meth()

    func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with
@typing.runtime_checkable act as simple-minded runtime protocols that check
only the presence of given attributes, ignoring their type signatures.
Protocol classes can be generic, they are defined as::

    class GenProto[T](Protocol):
        def meth(self) -> T:
            ...

WorkflowAdapterCapabilities

Stable · workflow

from zhivex_ai.workflows import WorkflowAdapterCapabilities

WorkflowAdapterCapabilities(durable_steps: 'bool' = False, native_step_retries: 'bool' = False, signals: 'bool' = False, explicit_resume: 'bool' = False, fork: 'bool' = False, cancellation: 'bool' = False, durable_timers: 'bool' = False) -> None

WorkflowAdapterCapabilities(durable_steps: 'bool' = False, native_step_retries: 'bool' = False, signals: 'bool' = False, explicit_resume: 'bool' = False, fork: 'bool' = False, cancellation: 'bool' = False, durable_timers: 'bool' = False)

WorkflowAgent

Stable · workflow

from zhivex_ai.workflows import WorkflowAgent

WorkflowAgent(*args, **kwargs)

Base class for protocol classes.

Protocol classes are defined as::

    class Proto(Protocol):
        def meth(self) -> int:
            ...

Such classes are primarily used with static type checkers that recognize
structural subtyping (static duck-typing).

For example::

    class C:
        def meth(self) -> int:
            return 0

    def func(x: Proto) -> int:
        return x.meth()

    func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with
@typing.runtime_checkable act as simple-minded runtime protocols that check
only the presence of given attributes, ignoring their type signatures.
Protocol classes can be generic, they are defined as::

    class GenProto[T](Protocol):
        def meth(self) -> T:
            ...

WorkflowBuilder

Stable · workflow

from zhivex_ai.workflows import WorkflowBuilder

WorkflowBuilder(name: 'str', *, definition_version: 'str' = '1') -> 'None'

Build a validated acyclic workflow graph without mutating prior builders.

WorkflowCheckpoint

Stable · workflow

from zhivex_ai.workflows import WorkflowCheckpoint

WorkflowCheckpoint(checkpoint_id: 'str', run_id: 'str', workflow_name: 'str', definition_version: 'str', definition_digest: 'str', sequence: 'int' = 0, schema_version: 'int' = 2, status: 'WorkflowCheckpointStatus' = 'running', session_id: 'str | None' = None, parent_run_id: 'str | None' = None, idempotency_key: 'str | None' = None, state: 'dict[str, JsonValue]' = <factory>, nodes: 'dict[str, WorkflowNodeCheckpoint]' = <factory>, edge_decisions: 'dict[str, bool]' = <factory>, ready_nodes: 'list[str]' = <factory>, pending_interrupt: 'WorkflowInterrupt | None' = None, transition: 'WorkflowTransition | None' = None, forked_from_run_id: 'str | None' = None, forked_from_checkpoint_id: 'str | None' = None, resume_values: 'dict[str, JsonValue]' = <factory>, created_at_ms: 'int | None' = None, updated_at_ms: 'int | None' = None, metadata: 'dict[str, JsonValue]' = <factory>, migration_history: 'list[WorkflowCheckpointMigration]' = <factory>) -> None

WorkflowCheckpoint(checkpoint_id: 'str', run_id: 'str', workflow_name: 'str', definition_version: 'str', definition_digest: 'str', sequence: 'int' = 0, schema_version: 'int' = 2, status: 'WorkflowCheckpointStatus' = 'running', session_id: 'str | None' = None, parent_run_id: 'str | None' = None, idempotency_key: 'str | None' = None, state: 'dict[str, JsonValue]' = <factory>, nodes: 'dict[str, WorkflowNodeCheckpoint]' = <factory>, edge_decisions: 'dict[str, bool]' = <factory>, ready_nodes: 'list[str]' = <factory>, pending_interrupt: 'WorkflowInterrupt | None' = None, transition: 'WorkflowTransition | None' = None, forked_from_run_id: 'str | None' = None, forked_from_checkpoint_id: 'str | None' = None, resume_values: 'dict[str, JsonValue]' = <factory>, created_at_ms: 'int | None' = None, updated_at_ms: 'int | None' = None, metadata: 'dict[str, JsonValue]' = <factory>, migration_history: 'list[WorkflowCheckpointMigration]' = <factory>)

WorkflowCheckpointMigration

Stable · workflow

from zhivex_ai.workflows import WorkflowCheckpointMigration

WorkflowCheckpointMigration(migration_id: 'str', from_version: 'int', to_version: 'int', applied_at_ms: 'int', metadata: 'dict[str, JsonValue]' = <factory>) -> None

WorkflowCheckpointMigration(migration_id: 'str', from_version: 'int', to_version: 'int', applied_at_ms: 'int', metadata: 'dict[str, JsonValue]' = <factory>)

WorkflowCheckpointStatus

Stable · workflow

from zhivex_ai.workflows import WorkflowCheckpointStatus

WorkflowCheckpointStatus(*args, **kwargs)

WorkflowCheckpointStore

Stable · workflow

from zhivex_ai.workflows import WorkflowCheckpointStore

WorkflowCheckpointStore(*args, **kwargs)

Base class for protocol classes.

Protocol classes are defined as::

    class Proto(Protocol):
        def meth(self) -> int:
            ...

Such classes are primarily used with static type checkers that recognize
structural subtyping (static duck-typing).

For example::

    class C:
        def meth(self) -> int:
            return 0

    def func(x: Proto) -> int:
        return x.meth()

    func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with
@typing.runtime_checkable act as simple-minded runtime protocols that check
only the presence of given attributes, ignoring their type signatures.
Protocol classes can be generic, they are defined as::

    class GenProto[T](Protocol):
        def meth(self) -> T:
            ...

WorkflowConflictError

Stable · errors

from zhivex_ai.workflows import WorkflowConflictError

WorkflowConflictError (public type alias or constant; no callable signature)

Raised when concurrent workflow state cannot be committed safely.

WorkflowContext

Stable · workflow

from zhivex_ai.workflows import WorkflowContext

WorkflowContext(run_id: 'str', workflow_name: 'str', source: 'str', target: 'str', state: 'Mapping[str, JsonValue]', source_status: 'str', source_output: 'JsonValue | None' = None, resume_values: 'Mapping[str, JsonValue]' = <factory>) -> None

WorkflowContext(run_id: 'str', workflow_name: 'str', source: 'str', target: 'str', state: 'Mapping[str, JsonValue]', source_status: 'str', source_output: 'JsonValue | None' = None, resume_values: 'Mapping[str, JsonValue]' = <factory>)

WorkflowDefinitionMismatchError

Stable · errors

from zhivex_ai.workflows import WorkflowDefinitionMismatchError

WorkflowDefinitionMismatchError (public type alias or constant; no callable signature)

Raised when persisted state belongs to a different workflow definition.

WorkflowEdge

Stable · workflow

from zhivex_ai.workflows import WorkflowEdge

WorkflowEdge(source: 'str', target: 'str', condition: 'WorkflowEdgeCondition | None' = None, name: 'str | None' = None, definition_revision: 'str | None' = None) -> None

A directed workflow edge with optional application-owned identity.

Set ``definition_revision`` when the condition depends on configuration or
closure values that cannot be represented reliably by its Python source.

WorkflowEdgeCondition

Stable · workflow

from zhivex_ai.workflows import WorkflowEdgeCondition

WorkflowEdgeCondition(*args, **kwargs)

WorkflowErrorPolicy

Stable · workflow

from zhivex_ai.workflows import WorkflowErrorPolicy

WorkflowErrorPolicy(*args, **kwargs)

WorkflowExecutionLease

Stable · workflow

from zhivex_ai.workflows import WorkflowExecutionLease

WorkflowExecutionLease(run_id: 'str', owner_id: 'str', token: 'str', fencing_token: 'int', acquired_at_ms: 'int', renewed_at_ms: 'int', expires_at_ms: 'int') -> None

WorkflowExecutionLease(run_id: 'str', owner_id: 'str', token: 'str', fencing_token: 'int', acquired_at_ms: 'int', renewed_at_ms: 'int', expires_at_ms: 'int')

WorkflowFunctionContext

Stable · workflow

from zhivex_ai.workflows import WorkflowFunctionContext

WorkflowFunctionContext(run_id: 'str', workflow_name: 'str', step_name: 'str', attempt: 'int', idempotency_key: 'str', input: 'JsonValue', state: 'Mapping[str, JsonValue]', resume_values: 'Mapping[str, JsonValue]' = <factory>, deps: 'Any' = None) -> None

WorkflowFunctionContext(run_id: 'str', workflow_name: 'str', step_name: 'str', attempt: 'int', idempotency_key: 'str', input: 'JsonValue', state: 'Mapping[str, JsonValue]', resume_values: 'Mapping[str, JsonValue]' = <factory>, deps: 'Any' = None)

WorkflowFunctionExecutor

Stable · workflow

from zhivex_ai.workflows import WorkflowFunctionExecutor

WorkflowFunctionExecutor(*args, **kwargs)

WorkflowFunctionResult

Stable · workflow

from zhivex_ai.workflows import WorkflowFunctionResult

WorkflowFunctionResult(output: 'JsonValue' = None, state_patch: 'Mapping[str, JsonValue]' = <factory>, metadata: 'Mapping[str, JsonValue]' = <factory>) -> None

WorkflowFunctionResult(output: 'JsonValue' = None, state_patch: 'Mapping[str, JsonValue]' = <factory>, metadata: 'Mapping[str, JsonValue]' = <factory>)

WorkflowGraph

Stable · workflow

from zhivex_ai.workflows import WorkflowGraph

WorkflowGraph(*, name: 'str', steps: 'Sequence[WorkflowStep]', edges: 'Sequence[WorkflowEdge]', definition_version: 'str' = '1', entrypoints: 'Sequence[str] | None' = None, checkpoint_store: 'WorkflowCheckpointStore | None' = None, run_store: 'AgentRunStore | None' = None, adapter: 'WorkflowAdapter | None' = None, interrupt_before: 'Mapping[str, str | None] | Sequence[str]' = (), interrupt_after: 'Mapping[str, str | None] | Sequence[str]' = (), max_concurrency: 'int | None' = None, lease_manager: 'WorkflowLeaseManager | None' = None, lease_ttl_ms: 'int' = 30000, lease_heartbeat_ms: 'int | None' = None, observer: 'AgentObserver | None' = None) -> 'None'

Durable DAG workflow with append-only checkpoints and explicit resume.

WorkflowInterrupt

Stable · workflow

from zhivex_ai.workflows import WorkflowInterrupt

WorkflowInterrupt(interrupt_id: 'str', node_name: 'str', reason: 'str | None' = None, payload: 'JsonValue | None' = None, created_at_ms: 'int | None' = None, phase: 'WorkflowInterruptPhase' = 'before', metadata: 'dict[str, JsonValue]' = <factory>) -> None

WorkflowInterrupt(interrupt_id: 'str', node_name: 'str', reason: 'str | None' = None, payload: 'JsonValue | None' = None, created_at_ms: 'int | None' = None, phase: 'WorkflowInterruptPhase' = 'before', metadata: 'dict[str, JsonValue]' = <factory>)

WorkflowInterruptError

Stable · errors

from zhivex_ai.workflows import WorkflowInterruptError

WorkflowInterruptError (public type alias or constant; no callable signature)

Raised when a workflow interrupt cannot be resumed safely.

WorkflowInterruptPhase

Stable · workflow

from zhivex_ai.workflows import WorkflowInterruptPhase

WorkflowInterruptPhase(*args, **kwargs)

WorkflowLeaseLostError

Stable · errors

from zhivex_ai.workflows import WorkflowLeaseLostError

WorkflowLeaseLostError (public type alias or constant; no callable signature)

Raised when a workflow worker no longer owns its execution lease.

WorkflowLeaseManager

Stable · workflow

from zhivex_ai.workflows import WorkflowLeaseManager

WorkflowLeaseManager(*args, **kwargs)

Base class for protocol classes.

Protocol classes are defined as::

    class Proto(Protocol):
        def meth(self) -> int:
            ...

Such classes are primarily used with static type checkers that recognize
structural subtyping (static duck-typing).

For example::

    class C:
        def meth(self) -> int:
            return 0

    def func(x: Proto) -> int:
        return x.meth()

    func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with
@typing.runtime_checkable act as simple-minded runtime protocols that check
only the presence of given attributes, ignoring their type signatures.
Protocol classes can be generic, they are defined as::

    class GenProto[T](Protocol):
        def meth(self) -> T:
            ...

WorkflowNodeCheckpoint

Stable · workflow

from zhivex_ai.workflows import WorkflowNodeCheckpoint

WorkflowNodeCheckpoint(node_name: 'str', status: 'WorkflowNodeStatus' = 'pending', attempt: 'int' = 0, idempotency_key: 'str | None' = None, child_run_id: 'str | None' = None, output: 'JsonValue | None' = None, error: 'str | None' = None, started_at_ms: 'int | None' = None, finished_at_ms: 'int | None' = None, metadata: 'dict[str, JsonValue]' = <factory>, suspension: 'dict[str, JsonValue] | None' = None) -> None

WorkflowNodeCheckpoint(node_name: 'str', status: 'WorkflowNodeStatus' = 'pending', attempt: 'int' = 0, idempotency_key: 'str | None' = None, child_run_id: 'str | None' = None, output: 'JsonValue | None' = None, error: 'str | None' = None, started_at_ms: 'int | None' = None, finished_at_ms: 'int | None' = None, metadata: 'dict[str, JsonValue]' = <factory>, suspension: 'dict[str, JsonValue] | None' = None)

WorkflowNodeStatus

Stable · workflow

from zhivex_ai.workflows import WorkflowNodeStatus

WorkflowNodeStatus(*args, **kwargs)

WorkflowRetryPolicy

Stable · workflow

from zhivex_ai.workflows import WorkflowRetryPolicy

WorkflowRetryPolicy(max_attempts: 'int' = 1, backoff_ms: 'int' = 250, max_backoff_ms: 'int' = 5000, retry_if: 'WorkflowRetryPredicate | None' = None) -> None

Retry policy for a complete logical workflow step.

``WorkflowStep.max_retries`` remains the provider/model retry setting passed
to ``run_agent``. This policy is deliberately separate because retrying a
complete step can repeat tools or external side effects.

WorkflowRetryPredicate

Stable · workflow

from zhivex_ai.workflows import WorkflowRetryPredicate

WorkflowRetryPredicate(*args, **kwargs)

WorkflowRunNotFoundError

Stable · errors

from zhivex_ai.workflows import WorkflowRunNotFoundError

WorkflowRunNotFoundError (public type alias or constant; no callable signature)

Raised when a requested workflow run or checkpoint cannot be found.

WorkflowRunResult

Stable · workflow

from zhivex_ai.workflows import WorkflowRunResult

WorkflowRunResult(run_id: 'str', name: 'str', session: 'AgentSession', state: 'WorkflowState', step_results: 'list[WorkflowStepResult]', text: 'str' = '', status: 'WorkflowRunStatus' = 'completed', trace: 'list[WorkflowTraceEvent]' = <factory>, state_snapshot: 'AgentRunState | None' = None, checkpoint: 'WorkflowCheckpoint | None' = None, forked_from_run_id: 'str | None' = None) -> None

WorkflowRunResult(run_id: 'str', name: 'str', session: 'AgentSession', state: 'WorkflowState', step_results: 'list[WorkflowStepResult]', text: 'str' = '', status: 'WorkflowRunStatus' = 'completed', trace: 'list[WorkflowTraceEvent]' = <factory>, state_snapshot: 'AgentRunState | None' = None, checkpoint: 'WorkflowCheckpoint | None' = None, forked_from_run_id: 'str | None' = None)

WorkflowRunStatus

Stable · workflow

from zhivex_ai.workflows import WorkflowRunStatus

WorkflowRunStatus(*args, **kwargs)

WorkflowState

Stable · workflow

from zhivex_ai.workflows import WorkflowState

WorkflowState(*args, **kwargs)

WorkflowStep

Stable · workflow

from zhivex_ai.workflows import WorkflowStep

WorkflowStep(name: 'str', agent: 'Agent | None' = None, prompt: 'str | None' = None, input_template: 'str | None' = None, output_key: 'str | None' = None, metadata_key: 'str | None' = None, max_retries: 'int | None' = None, timeout_ms: 'int | None' = None, error_policy: 'WorkflowErrorPolicy' = 'fail_fast', retry_policy: 'WorkflowRetryPolicy | None' = None, idempotency_key: 'str | None' = None, executor_ref: 'str | None' = None, metadata: 'dict[str, JsonValue]' = <factory>, executor: 'WorkflowFunctionExecutor | None' = None, definition_revision: 'str | None' = None) -> None

A workflow node and its runtime configuration.

``definition_revision`` is an application-owned stable token for semantic
configuration that the SDK cannot inspect reliably, such as agent model or
instruction changes, tool configuration, and values captured by an
executor closure. Change it whenever that external configuration changes
so durable resume and fork operations fail closed on the new definition.

WorkflowStepExecutor

Stable · workflow

from zhivex_ai.workflows import WorkflowStepExecutor

WorkflowStepExecutor(*args, **kwargs)

WorkflowStepExecutorRegistry

Stable · workflow

from zhivex_ai.workflows import WorkflowStepExecutorRegistry

WorkflowStepExecutorRegistry() -> 'None'

WorkflowStepOutcome

Stable · workflow

from zhivex_ai.workflows import WorkflowStepOutcome

WorkflowStepOutcome(workflow_run_id: 'str', node_id: 'str', activation_index: 'int', step_idempotency_key: 'str', status: 'WorkflowStepStatus', output: 'JsonValue' = None, state_patch: 'dict[str, JsonValue]' = <factory>, metadata: 'dict[str, JsonValue]' = <factory>, error: 'dict[str, JsonValue] | None' = None, suspension: 'dict[str, JsonValue] | None' = None, child_run_id: 'str | None' = None, schema_version: 'int' = 1) -> None

WorkflowStepOutcome(workflow_run_id: 'str', node_id: 'str', activation_index: 'int', step_idempotency_key: 'str', status: 'WorkflowStepStatus', output: 'JsonValue' = None, state_patch: 'dict[str, JsonValue]' = <factory>, metadata: 'dict[str, JsonValue]' = <factory>, error: 'dict[str, JsonValue] | None' = None, suspension: 'dict[str, JsonValue] | None' = None, child_run_id: 'str | None' = None, schema_version: 'int' = 1)

WorkflowStepRequest

Stable · workflow

from zhivex_ai.workflows import WorkflowStepRequest

WorkflowStepRequest(workflow_name: 'str', definition_version: 'str', definition_digest: 'str', workflow_run_id: 'str', node_id: 'str', executor_ref: 'str', activation_index: 'int' = 0, attempt: 'int' = 1, state_revision: 'int' = 0, input: 'JsonValue' = None, state: 'dict[str, JsonValue]' = <factory>, metadata: 'dict[str, JsonValue]' = <factory>, checkpoint_id: 'str | None' = None, correlation_ids: 'dict[str, str]' = <factory>, schema_version: 'int' = 1) -> None

WorkflowStepRequest(workflow_name: 'str', definition_version: 'str', definition_digest: 'str', workflow_run_id: 'str', node_id: 'str', executor_ref: 'str', activation_index: 'int' = 0, attempt: 'int' = 1, state_revision: 'int' = 0, input: 'JsonValue' = None, state: 'dict[str, JsonValue]' = <factory>, metadata: 'dict[str, JsonValue]' = <factory>, checkpoint_id: 'str | None' = None, correlation_ids: 'dict[str, str]' = <factory>, schema_version: 'int' = 1)

WorkflowStepResult

Stable · workflow

from zhivex_ai.workflows import WorkflowStepResult

WorkflowStepResult(name: 'str', status: 'WorkflowStepStatus', output: 'AgentRunResult | None' = None, error: 'Exception | None' = None, iteration: 'int | None' = None, output_text: 'str' = '', agent_run_id: 'str | None' = None, attempts: 'int' = 1) -> None

WorkflowStepResult(name: 'str', status: 'WorkflowStepStatus', output: 'AgentRunResult | None' = None, error: 'Exception | None' = None, iteration: 'int | None' = None, output_text: 'str' = '', agent_run_id: 'str | None' = None, attempts: 'int' = 1)

WorkflowStepStatus

Stable · workflow

from zhivex_ai.workflows import WorkflowStepStatus

WorkflowStepStatus(*args, **kwargs)

WorkflowStopCondition

Stable · workflow

from zhivex_ai.workflows import WorkflowStopCondition

WorkflowStopCondition(*args, **kwargs)

WorkflowTraceEvent

Stable · workflow

from zhivex_ai.workflows import WorkflowTraceEvent

WorkflowTraceEvent(type: 'str', workflow_name: 'str', step_name: 'str | None' = None, status: 'str | None' = None, iteration: 'int | None' = None, run_id: 'str | None' = None, error: 'str | None' = None) -> None

WorkflowTraceEvent(type: 'str', workflow_name: 'str', step_name: 'str | None' = None, status: 'str | None' = None, iteration: 'int | None' = None, run_id: 'str | None' = None, error: 'str | None' = None)

WorkflowTransition

Stable · workflow

from zhivex_ai.workflows import WorkflowTransition

WorkflowTransition(type: 'str', at_ms: 'int', node_name: 'str | None' = None, from_status: 'str | None' = None, to_status: 'str | None' = None, detail: 'dict[str, JsonValue]' = <factory>) -> None

WorkflowTransition(type: 'str', at_ms: 'int', node_name: 'str | None' = None, from_status: 'str | None' = None, to_status: 'str | None' = None, detail: 'dict[str, JsonValue]' = <factory>)

cancel_workflow

Stable · workflow

from zhivex_ai.workflows import cancel_workflow

cancel_workflow(workflow: 'WorkflowGraph', run_id: 'str', *, reason: 'str | None' = None, session: 'AgentSession | None' = None) -> 'WorkflowRunResult'

create_dbos_workflow_adapter

Beta · workflow

from zhivex_ai.workflows import create_dbos_workflow_adapter

create_dbos_workflow_adapter(callback: 'WorkflowStepExecutor') -> 'CallbackWorkflowAdapter'

beta named-engine factory; not a certified DBOS integration

create_in_memory_workflow_checkpoint_store

Stable · workflow

from zhivex_ai.workflows import create_in_memory_workflow_checkpoint_store

create_in_memory_workflow_checkpoint_store() -> 'InMemoryWorkflowCheckpointStore'

create_in_memory_workflow_lease_manager

Stable · workflow

from zhivex_ai.workflows import create_in_memory_workflow_lease_manager

create_in_memory_workflow_lease_manager() -> 'InMemoryWorkflowLeaseManager'

create_postgres_workflow_checkpoint_store

Stable · workflow

from zhivex_ai.workflows import create_postgres_workflow_checkpoint_store

create_postgres_workflow_checkpoint_store(dsn: 'str | None' = None, *, table_prefix: 'str' = 'zhivex_ai', namespace: 'str' = 'default', pool: 'Any | None' = None, pool_min_size: 'int' = 1, pool_max_size: 'int' = 5) -> 'PostgresWorkflowCheckpointStore'

create_postgres_workflow_lease_manager

Stable · workflow

from zhivex_ai.workflows import create_postgres_workflow_lease_manager

create_postgres_workflow_lease_manager(dsn: 'str | None' = None, *, table_prefix: 'str' = 'zhivex_ai', namespace: 'str' = 'default', pool: 'Any | None' = None, pool_min_size: 'int' = 1, pool_max_size: 'int' = 5) -> 'PostgresWorkflowLeaseManager'

create_prefect_workflow_adapter

Beta · workflow

from zhivex_ai.workflows import create_prefect_workflow_adapter

create_prefect_workflow_adapter(callback: 'WorkflowStepExecutor') -> 'CallbackWorkflowAdapter'

beta named-engine factory; not a certified Prefect integration

create_restate_workflow_adapter

Beta · workflow

from zhivex_ai.workflows import create_restate_workflow_adapter

create_restate_workflow_adapter(callback: 'WorkflowStepExecutor') -> 'CallbackWorkflowAdapter'

beta named-engine factory; not a certified Restate integration

create_sqlite_workflow_checkpoint_store

Stable · workflow

from zhivex_ai.workflows import create_sqlite_workflow_checkpoint_store

create_sqlite_workflow_checkpoint_store(path: 'str', *, namespace: 'str' = 'default') -> 'SQLiteWorkflowCheckpointStore'

create_sqlite_workflow_lease_manager

Stable · workflow

from zhivex_ai.workflows import create_sqlite_workflow_lease_manager

create_sqlite_workflow_lease_manager(path: 'str', *, namespace: 'str' = 'default') -> 'SQLiteWorkflowLeaseManager'

create_temporal_workflow_adapter

Beta · workflow

from zhivex_ai.workflows import create_temporal_workflow_adapter

create_temporal_workflow_adapter(callback: 'WorkflowStepExecutor') -> 'CallbackWorkflowAdapter'

beta named-engine factory; not a certified Temporal integration

deserialize_workflow_checkpoint

Stable · workflow

from zhivex_ai.workflows import deserialize_workflow_checkpoint

deserialize_workflow_checkpoint(payload: 'dict[str, Any]') -> 'WorkflowCheckpoint'

fork_workflow

Stable · workflow

from zhivex_ai.workflows import fork_workflow

fork_workflow(workflow: 'WorkflowGraph', run_id: 'str', *, checkpoint_id: 'str | None' = None, state_updates: 'Mapping[str, JsonValue] | None' = None, idempotency_key: 'str | None' = None, deps: 'Any' = None, session: 'AgentSession | None' = None) -> 'WorkflowRunResult'

migrate_workflow_checkpoint

Stable · workflow

from zhivex_ai.workflows import migrate_workflow_checkpoint

migrate_workflow_checkpoint(checkpoint: 'WorkflowCheckpoint', *, target_version: 'int' = 2, applied_at_ms: 'int | None' = None) -> 'WorkflowCheckpoint'

Return a migrated checkpoint without mutating or persisting the source value.

Migrations are explicit, sequential, and auditable. Callers control the timestamp
when deterministic release or fixture evidence is required.

migrate_workflow_checkpoint_payload

Stable · workflow

from zhivex_ai.workflows import migrate_workflow_checkpoint_payload

migrate_workflow_checkpoint_payload(payload: 'dict[str, Any]', *, target_version: 'int' = 2, applied_at_ms: 'int | None' = None) -> 'dict[str, Any]'

Deserialize, migrate, and serialize a persisted checkpoint payload.

migrate_workflow_run_checkpoint

Stable · workflow

from zhivex_ai.workflows import migrate_workflow_run_checkpoint

migrate_workflow_run_checkpoint(store: 'WorkflowCheckpointStore', run_id: 'str', *, target_version: 'int' = 2, applied_at_ms: 'int | None' = None) -> 'WorkflowCheckpoint'

Append the migrated latest checkpoint using the store's compare-and-swap contract.

resume_workflow

Stable · workflow

from zhivex_ai.workflows import resume_workflow

resume_workflow(workflow: 'WorkflowGraph', run_id: 'str', *, interrupt_id: 'str | None' = None, resume_value: 'JsonValue | None' = None, state_updates: 'Mapping[str, JsonValue] | None' = None, approval_id: 'str | None' = None, approved: 'bool' = True, reason: 'str | None' = None, node_name: 'str | None' = None, deps: 'Any' = None, session: 'AgentSession | None' = None) -> 'WorkflowRunResult'

run_workflow

Stable · workflow

from zhivex_ai.workflows import run_workflow

run_workflow(workflow: 'WorkflowAgent', *, session: 'AgentSession | None' = None, prompt: 'str | None' = None, parent_run_id: 'str | None' = None) -> 'WorkflowRunResult'

serialize_workflow_checkpoint

Stable · workflow

from zhivex_ai.workflows import serialize_workflow_checkpoint

serialize_workflow_checkpoint(checkpoint: 'WorkflowCheckpoint') -> 'dict[str, Any]'

validate_workflow_expectations

Stable · workflow

from zhivex_ai.workflows import validate_workflow_expectations

validate_workflow_expectations(result: 'WorkflowRunResult', expectations: 'object') -> 'list[str]'

workflow_checkpoint_from_json

Stable · workflow

from zhivex_ai.workflows import workflow_checkpoint_from_json

workflow_checkpoint_from_json(value: 'str') -> 'WorkflowCheckpoint'

workflow_checkpoint_to_json

Stable · workflow

from zhivex_ai.workflows import workflow_checkpoint_to_json

workflow_checkpoint_to_json(checkpoint: 'WorkflowCheckpoint') -> 'str'

workflow_step

Stable · workflow

from zhivex_ai.workflows import workflow_step

workflow_step(name: 'str', agent: 'Agent | None' = None, *, prompt: 'str | None' = None, input_template: 'str | None' = None, output_key: 'str | None' = None, metadata_key: 'str | None' = None, max_retries: 'int | None' = None, timeout_ms: 'int | None' = None, error_policy: 'WorkflowErrorPolicy' = 'fail_fast', retry_policy: 'WorkflowRetryPolicy | None' = None, idempotency_key: 'str | None' = None, executor_ref: 'str | None' = None, metadata: 'dict[str, JsonValue] | None' = None, executor: 'WorkflowFunctionExecutor | None' = None) -> 'WorkflowStep'