Argyll logo Argyll

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_invoke() \
    .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_description(text), and with_tags(*tags), which merges into the Step’s tag set, sorted and deduped

Attributes

  • required(name, type): required input
  • optional(name, type, default): optional input with default value
  • const(name, type, value): a fixed value baked into the Step definition
  • meta(name, meta_key): a named metadata field (e.g. flow_id, webhook_url) injected at execution time
  • output(name, type): output attribute
  • with_for_each(name): mark an input for parallel expansion over its collected values, requiring a collect policy of all or some
  • 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_invoke(), with_async_invoke(): 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 a chain of fallback Goal sets
  • 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(Goals(...)) to set the chain of fallback Goal sets at once, or with_goal(id) to add a Goal to the last set in the chain. else_ carries a trailing underscore because else is a Python keyword:

from argyll import Client, Goals

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

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

Initial values are arrays. Wrap each value in a list. Each key declares the resolved values for that Attribute, so an empty list declares zero values and an omitted key leaves the Attribute to providers. See Initial Values Are Authoritative.

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,
)
Previous Go Step Generator