stitch-eng-sdk 0.2.1

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()

Remote Servers & Async Support

If you have a remote production server, you can pass the URL and an API key:

client = stitch.Client("https://stitch.mycompany.com", api_key="sk_secret")

The SDK also provides an AsyncClient for modern asyncio applications like FastAPI:

from stitch import AsyncClient

async with AsyncClient("http://localhost:8080") as client:
    run = await client.run(process_order, input={"amount": 100})
    result = await run.wait()

Requirements

• Python ≥ 3.9

License

MIT