threadify-sdk 0.2.5

Python SDK for Threadify — service-delivery intelligence. Track every customer request from start to finish across every system, team, and partner. Visit: https://threadify.dev

Threadify Python SDK

Python SDK for Threadify — service-delivery intelligence that tracks every customer request from start to finish across every system, team, and partner.

• Documentation: https://docs.threadify.dev
• Create an account: https://threadify.dev

Installation

pip install threadify-sdk

For local development:

pip install -e ".[dev]"

Quick Start

import asyncio
import logging
from threadify import Threadify


async def main():
    try:
        conn = await Threadify.connect(
            "your-api-key",
            service_name="orders-service",
        )
    except Exception as e:
        logging.error(f"Failed to connect: {e}")
        return

    try:
        thread = await conn.start("Order-123")
        
        # Easy chaining!
        await (
            thread.step("order_received")
            .add_context({"orderId": "ORD-123"})
            .success("Order accepted")
        )
    except Exception as e:
        logging.error(f"Error in thread: {e}")
    finally:
        await conn.close()


if __name__ == "__main__":
    asyncio.run(main())

ws_url defaults to wss://eng.threadify.dev/threads. Override it if you are using a self-hosted or regional endpoint.

Configuration

Connect options

Use keyword arguments with Threadify.connect(...):

• service_name
• ws_url (optional, defaults to production)
• graphql_url
• debug
• max_in_flight
• connect_timeout

Example:

from threadify import Threadify

conn = await Threadify.connect(
    "your-api-key",
    service_name="inventory-service",
    debug=True,
)

Join options

Use Connection.join(...) with keyword arguments:

• token=...
• thread_id=... and role=...

Example:

thread = await conn.join(token="jwt-token")

Or:

thread = await conn.join(
    thread_id="thread-123",
    role="supplier",
)

Subscriptions

Preferred API:

• subscribe(event, step_name, handler)
• unsubscribe(event, step_name)

Supported event patterns:

• step.success
• step.failed
• step.*
• rule.passed
• rule.violated
• rule.*
• *

Example:

def handle_notification(notification):
    if notification.is_violated:
        print(notification.message)



conn.subscribe("rule.violated", "payment_step", handle_notification)

Versioning & Releases

This SDK follows Semantic Versioning and Conventional Commits. Releases are automated via GitHub Actions.

• fix: ... -> Patch bump
• feat: ... -> Minor bump
• feat!: ... or BREAKING CHANGE: ... -> Major bump

Testing

To run the SDK tests, execute:

make test

Alternatively, use pytest:

python3 -m pytest