Python SDK

Use the Python builder to register endpoints, serve Step handlers, and start or inspect Flows.

Installation

pip install argyll-sdk

Client

from argyll import Client

client = Client("http://localhost:8080")

Register a Script Step

from argyll import Client, AttributeType, ScriptConfig, ScriptLanguage

client = Client("http://localhost:8080")

client.new_step() \
    .with_id("price-calculator") \
    .with_name("Price Calculator") \
    .required("quantity", AttributeType.NUMBER) \
    .required("unit_price", AttributeType.NUMBER) \
    .output("total", AttributeType.NUMBER) \
    .with_script(ScriptConfig(
        language=ScriptLanguage.LUA,
        script="return {total = quantity * unit_price}",
    )) \
    .register()

Executable Script Steps use Lua. ScriptLanguage.JPATH identifies JSONPath for predicates and attribute mappings, not executable Script Steps.

Serve a Sync Step

start() registers the Step and runs an HTTP server that handles invocations:

from argyll import Client, StepContext, AttributeType

client = Client("http://localhost:8080")

def handle_greeting(ctx: StepContext, args: dict) -> dict:
    name = args.get("name", "World")
    return {"greeting": f"Hello, {name}!"}

client.new_step() \
    .with_id("greet") \
    .with_name("Greet") \
    .required("name", AttributeType.STRING) \
    .output("greeting", AttributeType.STRING) \
    .start(handle_greeting)

Serve an Async Step

import threading
from argyll import AsyncContext, AttributeType, Client, StepContext

client = Client("http://localhost:8080")

def handle_process(ctx: StepContext, args: dict) -> dict:
    async_ctx = AsyncContext(context=ctx, webhook_url=ctx.metadata["webhook_url"])

    def background():
        result = do_long_work(args)
        async_ctx.success({"result": result})

    threading.Thread(target=background).start()
    return {}  # return immediately

client.new_step() \
    .with_id("process") \
    .with_name("Process") \
    .with_async_execution() \
    .required("input", AttributeType.OBJECT) \
    .output("result", AttributeType.OBJECT) \
    .start(handle_process)

Serve a Step with Compensation

Use with_compensate_handler to register a handler that undoes a completed Work Item. start() generates the compensation URL:

def handle_charge(ctx: StepContext, args: dict) -> dict:
    charge_id = process_charge(args["amount"])
    return {"charge_id": charge_id}

def handle_compensate(ctx: StepContext, args: dict) -> None:
    refund_charge(args["charge_id"])

client.new_step() \
    .with_id("charge-card") \
    .with_name("Charge Card") \
    .required("amount", AttributeType.NUMBER) \
    .output("charge_id", AttributeType.STRING) \
    .compensated("charge_id") \
    .with_compensate_handler(handle_compensate) \
    .start(handle_charge)

The compensation handler receives a flat argument dictionary containing the selected Attributes under their invocation names. Raise WorkNotCompletedError to request another attempt, or another exception to report permanent failure.

Builder Methods

Identity

with_id(id), with_name(name), with_label(k, v), with_labels(labels)

Attributes

required(name, type): required input
optional(name, type, default): optional input with default value
const(name, type, value): constant — a fixed value baked into the Step definition
meta(name, meta_key): metadata — injects a named metadata field (e.g. flow_id, webhook_url) at execution time
output(name, type): output attribute
with_for_each(name): mark an array input for parallel expansion
compensated(*names): include Attributes in compensation requests

Execution Type

with_script(...) selects Script execution

Action Mode

with_invoke_mode(ActionMode), with_compensate_mode(ActionMode): set how an action reports its result, ActionMode.SYNC (default) or ActionMode.ASYNC
with_sync_execution(), with_async_execution(): shorthand for the invoke action’s mode

HTTP

with_endpoint(url), with_method(method), with_timeout(ms), with_health_check(url)
with_compensate(url): set the endpoint URL and compensated handling
with_compensate_method(method): compensate HTTP method, defaults to POST
with_compensate_timeout(ms): compensate timeout, defaults to the Step timeout
with_compensate_handler(handler): register a compensation handler (auto-generates the URL when used with start())

Script

with_script(ScriptConfig(...)) - executable Script Steps use Lua

Predicate

with_predicate(ScriptConfig(...)) - predicates may use Lua or JSONPath (ScriptLanguage.JPATH)

Behavior

with_handling(Handling): choose standard, memoized, or compensated handling
with_flow_goals(*goals): configure a Flow-type Step
with_flow_space(space_id): restrict its child flow to a Space

Lifecycle

register(): register or update the Step definition, retrying transient failures
start(handler): register or update with retries, then serve

List Steps

steps = client.list_steps()

Start a Flow

Use with_goals(*ids) to set all goals at once, or with_goal(id) to add one at a time:

from argyll import Client

client = Client("http://localhost:8080")

client.new_flow("order-123") \
    .with_goals("send-confirmation") \
    .with_initial_state({
        "customer_id": ["cust-456"],
        "order_amount": [99.99],
    }) \
    .start()

Initial values are arrays. Wrap each value in a list.

Query Flow State

flow = client.flow("order-123")
state = flow.get_state()

Imports

from argyll import (
    Client, StepBuilder, FlowBuilder, FlowClient,
    StepContext, AsyncContext,
    AttributeType, AttributeRole, InputCollect,
    ActionMode, BackoffType, Handling, ScriptLanguage,
)
from argyll.errors import (
    ArgyllError, ClientError, HTTPError, StepRegistrationError,
    StepValidationError, FlowError, WebhookError,
)