stitch-eng-sdk 0.2.3

Python SDK for the Stitch workflow execution engine

stitch-eng-sdk

Python SDK for the Stitch workflow execution engine.

Install

pip install stitch-eng-sdk

Quickstart (Embedded Zero-Config Mode)

Stitch comes with an embedded Go binary so you don't even need to start a server locally!

import threading
import stitch

@stitch.task(max_attempts=5)
def charge_payment(order_id, amount):
    return {"txn_id": "txn_123"}

@stitch.task
def send_email(email, txn_id):
    return {"sent": True}

@stitch.workflow
def process_order(inp):
    # Pass the output of one task directly to the next to create a dependency!
    payment = charge_payment(order_id=inp["order_id"], amount=inp["amount"])
    send_email(email=inp["email"], txn_id=payment["txn_id"])

# 1. Start the embedded server and client
client = stitch.Client()

# 2. Start a background worker to poll for tasks
worker = stitch.Worker(client, tasks=[charge_payment, send_email])
threading.Thread(target=worker.start, daemon=True).start()

# 3. Submit the workflow and wait for it to finish!
result = client.run(process_order, input={
    "order_id": "ORD-001",
    "amount": 9897,
    "email": "[e-mail]",
}).wait()

print(result.status)  # → "completed"

worker.stop()
client.close()

Cloud Mode

To connect to the hosted Stitch backend, just pass your API key — the server URL is handled automatically:

client = stitch.Client(api_key="sk_your_api_key")

Full Cloud Example

import threading
import stitch

@stitch.task(max_attempts=5)
def charge_payment(order_id, amount):
    return {"txn_id": "txn_123"}

@stitch.task
def send_email(email, txn_id):
    return {"sent": True}

@stitch.workflow
def process_order(inp):
    payment = charge_payment(order_id=inp["order_id"], amount=inp["amount"])
    send_email(email=inp["email"], txn_id=payment["txn_id"])

# Connect to Stitch Cloud
client = stitch.Client(api_key="sk_your_api_key")

# Start a background worker
worker = stitch.Worker(client, tasks=[charge_payment, send_email])
threading.Thread(target=worker.start, daemon=True).start()

# Submit and wait
result = client.run(process_order, input={
    "order_id": "ORD-001",
    "amount": 9897,
    "email": "[e-mail]",
}).wait()

print(result.status)  # → "completed"

worker.stop()
client.close()