Metadata-Version: 2.4
Name: stitch-eng-sdk
Version: 0.2.3
Summary: Python SDK for the Stitch workflow execution engine
Author: Stitch Engineering
License-Expression: MIT
Project-URL: Homepage, https://github.com/stitch-eng/stitch
Project-URL: Source, https://github.com/stitch-eng/stitch
Project-URL: Issues, https://github.com/stitch-eng/stitch/issues
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.23.0

# stitch-eng-sdk

Python SDK for the [Stitch](https://github.com/stitch-eng/stitch) workflow execution engine.

## Install

```bash
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!

```python
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": "user@example.com",
}).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:

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

### Full Cloud Example

```python
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": "user@example.com",
}).wait()

print(result.status)  # → "completed"

worker.stop()
client.close()
```
