arazzo-exec 0.1.0

Standalone Arazzo workflow execution SDK and CLI

arazzo-exec

arazzo-exec is an extensible Python SDK and CLI for parsing, validating, planning, and executing Arazzo workflows.

It is built for workflows that combine standard Arazzo/OpenAPI/AsyncAPI steps with custom execution extensions such as DB assertions, jobs, Temporal workflows, Kafka messages, auth providers, and local Testcontainers resources.

Design goals

• One package, optional extras.
• Core without globals or service-locator patterns.
• DI-compatible, but not coupled to a DI framework.
• Typed dataclass AST that preserves x-* extensions and raw source data.
• Explicit operation resolution and executor selection through registries.
• Async-first runner with a sync wrapper.
• Core does not know about DB, Temporal, Kafka, Testcontainers, or company-specific integrations.
• Extensions are public ABC-based contracts, not hidden callbacks.
• Typed execution events, event readers, and JSONL logs.
• Per-run security policy and secret redaction.
• Run-scoped resource lifecycle for clients, pools, containers, and future integrations.

Installation

Run the CLI without adding it to a project:

uvx --from "arazzo-exec[all]" arazzo-exec doctor
uvx --from "arazzo-exec[all]" arazzo-exec extensions

Add the SDK to a Python project:

uv add arazzo-exec

Traditional PyPI install:

python -m pip install arazzo-exec

Optional extras:

uv add "arazzo-exec[mapping]"      # adaptix Retort support
uv add "arazzo-exec[di]"           # dishka for CLI/integration DI helpers
uv add "arazzo-exec[kafka]"        # confluent-kafka
uv add "arazzo-exec[db]"           # psycopg + pymysql
uv add "arazzo-exec[temporal]"     # temporalio
uv add "arazzo-exec[containers]"   # testcontainers
uv add "arazzo-exec[schemathesis]" # schemathesis bridge
uv add "arazzo-exec[extensions]"   # common heavy extension deps
uv add "arazzo-exec[all]"          # all extras

arazzo-exec[core] is intentionally a no-op extra for UX compatibility. With one Python distribution, extras can only add dependencies; they cannot make the base install smaller.

Package layout

arazzo_exec/
  core/
    ast/          # ArazzoDocument, Workflow, Step, Action, Criterion, ExtensionBag
    io/           # YAML/JSON parsing, source loading, source index/cache
    operations/   # operationId/operationPath/channelPath resolver registry
    execution/    # runner, planner, DAG scheduler, state, resources, results
    executors/    # explicit step executor registry and selectors
    extensions/   # public ABCs, registry, manifest, discovery, extension policy
    events/       # typed events, sinks, readers, JSONL, redaction
    auth/         # auth provider ABCs and default providers
    security/     # per-run policy, network checks, secrets, redaction
    validation/   # diagnostics and executable-profile validation
    expressions/  # runtime expression engine and namespaces
    criteria/     # standard criterion evaluation

  extensions/
    runtime/       # x-ae-timeout, x-ae-retry, x-ae-assert, x-ae-outputs
    auth/          # runtime auth provider
    jobs/          # x-ae-job
    asyncapi/      # in-memory AsyncAPI executor
    kafka/         # Kafka AsyncAPI executor
    db/            # x-ae-db, x-ae-db-assert
    temporal/      # x-ae-temporal
    testcontainers/# runtime resources for Postgres, MySQL, Redpanda
    schemathesis/  # optional future bridge

  cli/             # arazzo-exec command line interface

Quick SDK usage

from arazzo_exec import ArazzoRunner
from arazzo_exec.extensions import build_default_registry

runner = ArazzoRunner.from_path(
    "workflow.arazzo.yaml",
    registry=build_default_registry(),
)

result = runner.run(
    "happyPath",
    inputs={"invoiceId": "inv-1001"},
    runtime={"baseUrl": "http://127.0.0.1:8000"},
)

if not result.success:
    print(result.error)
    raise SystemExit(1)

print(result.outputs)

Async runner:

from arazzo_exec import AsyncArazzoRunner
from arazzo_exec.extensions import build_default_registry

runner = AsyncArazzoRunner.from_path(
    "workflow.arazzo.yaml",
    registry=build_default_registry(),
)

result = await runner.run(
    "happyPath",
    inputs={"invoiceId": "inv-1001"},
    runtime={"parallel": True},
)

Parse only:

from arazzo_exec import parse_arazzo_path

document = parse_arazzo_path("workflow.arazzo.yaml")
workflow = document.workflow_by_id("happyPath")
step = workflow.step_by_id("createInvoice")

print(document.arazzo)
print(step.extensions.values)

Custom source types can provide an OperationResolver and a StepExecutor. The resolver answers "what operation does this Arazzo reference mean?" and the executor answers "how do I run it?"

from arazzo_exec.core import StepExecutorKind, StepExecutorRegistration
from arazzo_exec.core.executors import ResolvedOperationKindSelector
from arazzo_exec.core.operations import (
    OperationResolverRegistration,
    SourceTypeOperationResolverSelector,
)

registry.register_operation_resolver(
    OperationResolverRegistration(
        name="company.graphql.schema",
        selector=SourceTypeOperationResolverSelector("graphql"),
        resolver=CompanyGraphQLOperationResolver(),
        produced_kinds=("graphql.operation",),
    )
)

registry.register_step_executor(
    StepExecutorRegistration(
        name="company.graphql.http",
        kind=StepExecutorKind.STANDARD_OPERATION,
        selector=ResolvedOperationKindSelector("graphql.operation"),
        executor=CompanyGraphQLExecutor(),
    )
)

sourceDescriptions:
  - name: users
    type: graphql
    url: ./schema.graphql

workflows:
  - workflowId: userFlow
    steps:
      - stepId: getUser
        operationId: $sourceDescriptions.users.GetUser
        x-ae-executor:
          name: company.graphql.http

Validate:

from arazzo_exec import build_validation_pipeline
from arazzo_exec.extensions import build_default_registry

pipeline = build_validation_pipeline(build_default_registry())
result = pipeline.validate(document)

for diagnostic in result.diagnostics:
    print(diagnostic.severity, diagnostic.code, diagnostic.pointer, diagnostic.message)

CLI quick start

arazzo-exec doctor
arazzo-exec extensions
arazzo-exec validate workflow.arazzo.yaml
arazzo-exec list-workflows workflow.arazzo.yaml
arazzo-exec inspect workflow.arazzo.yaml --workflow happyPath
arazzo-exec env-check workflow.arazzo.yaml

Run a workflow:

arazzo-exec run workflow.arazzo.yaml \
  --workflow happyPath \
  --input inputs/happy-path.yaml \
  --base-url http://127.0.0.1:8000

Run with JSONL events:

arazzo-exec run workflow.arazzo.yaml \
  --workflow happyPath \
  --input inputs/happy-path.yaml \
  --events-jsonl reports/events.jsonl

arazzo-exec events reports/events.jsonl --count-by-type
arazzo-exec events reports/events.jsonl --errors

Run with security controls:

arazzo-exec run workflow.arazzo.yaml \
  --workflow happyPath \
  --dry-run \
  --allow-host api.company.local \
  --allow-method GET \
  --allow-method POST \
  --deny-private-network \
  --no-remote-sources \
  --no-job-execution

Load custom extensions:

arazzo-exec run workflow.arazzo.yaml \
  --workflow happyPath \
  --extension my_package.arazzo:MyExtension

arazzo-exec run workflow.arazzo.yaml \
  --workflow happyPath \
  --discover-extensions

Use Kafka executor explicitly:

arazzo-exec run workflow.arazzo.yaml \
  --workflow kafkaFlow \
  --kafka

Arazzo example

arazzo: 1.1.0
info:
  title: Invoice workflow
  version: 1.0.0

sourceDescriptions:
  - name: billing
    type: openapi
    url: ./openapi.yaml

workflows:
  - workflowId: happyPath
    inputs:
      type: object
      properties:
        invoiceId:
          type: string
    steps:
      - stepId: createInvoice
        operationId: $sourceDescriptions.billing.createInvoice
        requestBody:
          payload:
            invoiceId: $inputs.invoiceId
        successCriteria:
          - condition: $statusCode == 201
        x-ae-db-assert:
          dsnEnv: AE_DATABASE_URL
          queryOne: "select status from invoices where id = :invoice_id"
          args:
            invoice_id: $response.body.id
          expect:
            status: CREATED
        x-ae-outputs:
          invoiceId: $response.body.id
          status: $response.body.status

    outputs:
      invoiceId: $steps.createInvoice.outputs.invoiceId
      status: $steps.createInvoice.outputs.status

Built-in extension keys

Core/registry:

x-ae-runtime
x-ae-executor
x-ae-inputs
x-ae-parameters

Runtime controls:

x-ae-timeout
x-ae-retry
x-ae-successCriteria
x-ae-assert
x-ae-outputs

Execution extensions:

x-ae-job
x-ae-db
x-ae-db-assert
x-ae-temporal
x-ae-kafka

Runtime resources:

x-ae-runtime.testcontainers
x-ae-runtime.containers

Use your own namespace for project-specific extensions:

x-company-*
x-team-*
x-product-*

Executor selection

Step execution is registry-driven. There is no hidden priority or fallback logic.

A step gets exactly one primary executor from registered selectors:

• OpenAPI operation executor for OpenAPI operationId / operationPath.
• In-memory AsyncAPI executor for AsyncAPI operationId / operationPath / channelPath in the default registry.
• Kafka AsyncAPI executor when registered through build_kafka_registry() or CLI --kafka.
• Nested workflow executor for workflowId steps.
• Extension executors such as x-ae-job, x-ae-db, or x-ae-temporal.

If more than one executor matches, validation and execution fail with an ambiguity error. Use x-ae-executor.name to select explicitly:

- stepId: publishOrder
  operationId: $sourceDescriptions.events.publishOrder
  action: send
  x-ae-executor:
    name: kafka.asyncapi
  x-ae-kafka:
    connection: default

Runtime config

Runtime can be passed from SDK as a dict or declared at the root via x-ae-runtime.

x-ae-runtime:
  parallel: true
  baseUrl: http://127.0.0.1:8000
  auth:
    billing:
      bearerEnv: BILLING_TOKEN
      headers:
        X-Tenant: tenant-a
  security:
    dryRun: false
    allowedHosts:
      - 127.0.0.1
      - api.company.local
    allowedMethods:
      - GET
      - POST
    denyPrivateNetwork: false
    allowJobExecution: true
    maxResponseBytes: 1048576
  testcontainers:
    postgres:
      billing:
        image: postgres:16
    redpanda:
      default: {}

SDK runtime overrides root runtime:

result = runner.run(
    "happyPath",
    runtime={"parallel": False, "security": {"dryRun": True}},
)

Events

Every run emits typed events:

result = runner.run("happyPath")
reader = result.event_reader()

print(reader.count_by_type())
print(reader.failed_steps())

Use JSONL for persistent logs:

from arazzo_exec import ArazzoRunner, JsonlEventSink, EventBus, RuntimeServices
from arazzo_exec.extensions import build_default_registry

sink = JsonlEventSink("events.jsonl")
services = RuntimeServices.default_local(event_bus=EventBus(sinks=(sink,)))
runner = ArazzoRunner.from_path("workflow.arazzo.yaml", registry=build_default_registry(), services=services)
result = runner.run("happyPath")

Read logs back:

from arazzo_exec import JsonlEventLog

reader = JsonlEventLog.read("events.jsonl")
print(reader.by_type("step.failed"))

Custom extension skeleton

from dataclasses import dataclass
from typing import Any

from arazzo_exec.core import (
    ArazzoExtension,
    ExtensionManifest,
    ExtensionRegistry,
    StepExecutor,
    StepExecutorKind,
    StepExecutorRegistration,
    StepResult,
)
from arazzo_exec.core.executors import ExtensionFieldSelector


@dataclass(frozen=True, slots=True)
class CompanyCheckExecutor(StepExecutor):
    async def execute(self, request, spec: Any) -> StepResult:
        actual = request.evaluator().evaluate(spec["actual"])
        expected = request.evaluator().evaluate(spec["expected"])
        success = actual == expected
        return StepResult.succeeded(
            request.step.step_id,
            outputs={"actual": actual, "expected": expected, "success": success},
        ) if success else StepResult.failed(
            request.step.step_id,
            f"Expected {expected!r}, got {actual!r}",
        )


@dataclass(frozen=True, slots=True)
class CompanyExtension(ArazzoExtension):
    @property
    def manifest(self) -> ExtensionManifest:
        return ExtensionManifest(
            name="company.check",
            version="1.0.0",
            keys=("x-company-check",),
            package_extra=None,
            description="Company assertion executor.",
        )

    def register(self, registry: ExtensionRegistry) -> None:
        registry.register_step_executor(
            StepExecutorRegistration(
                name="company.check",
                kind=StepExecutorKind.EXTENSION_OPERATION,
                selector=ExtensionFieldSelector("x-company-check"),
                executor=CompanyCheckExecutor(),
            )
        )

Register manually:

from arazzo_exec.extensions import build_default_registry

registry = build_default_registry()
registry.register_extension(CompanyExtension())

Or expose an entry point:

[project.entry-points."arazzo_exec.extensions"]
company_check = "my_package.arazzo:CompanyExtension"

More documentation

• ARCHITECTURE.md
• docs/CLI.md
• docs/EXTENSIONS.md
• docs/EVENTS.md
• docs/OPERATIONS.md
• docs/SECURITY.md