Python SDK

import httpx
from staqly import Staqly

staqly = Staqly(
    base_url="https://app.staqly.ai",  # default: $STAQLY_API_URL, then http://localhost:3333
    api_key="stk_...",  # default: $STAQLY_API_KEY; "" = ignore the env var
    timeout=60.0,  # seconds per request (or an httpx.Timeout)
    max_retries=2,  # retries for 408/429/5xx/network errors
    default_headers={"X-Team": "growth"},
    user_agent="my-agent/1.0",  # appended to "staqly-sdk-python/<version>"
    http_client=httpx.Client(proxy="http://proxy:8080"),  # your own client/transport
)

staqly.posts.list(status="scheduled", request_options={"timeout": 5, "max_retries": 0})
staqly.posts.create("Posts", content="x", request_options={"idempotency_key": f"job-{job_id}"})

# a draft with a text of its own on one channel (x and publisher are channels from channels.list())
post = staqly.posts.create(
    "Product updates",  # ID or exact name
    content="Three days, every avatar half price.",
    media=[media.id],  # from staqly.media.upload
    targets=[
        {"channel_id": x.id, "content": "Three days. Half price."},
        {"channel_id": publisher.id},  # "settings": {...} sets the provider's per-post fields
    ],
    tags=["sale"],
    metadata={"campaign": "spring"},
)

staqly.posts.preview(post.id)  # the text per channel and what each platform objects to
staqly.posts.schedule(post.id, "2026-11-03T09:00:00Z")  # onto the calendar
staqly.posts.cancel(post.id)  # back to draft
staqly.posts.publish(post.id)  # now (202: a job does the work)
staqly.posts.retry_target(post.id, x.id)  # one failed channel again
staqly.posts.update(post.id, content="Fixed", expected_version=post.version)  # ConflictError when someone edited it
copy = staqly.posts.duplicate(post.id)  # a new draft; published posts are read-only

# list, search and page through everything
staqly.posts.list(status="scheduled", publish_after="2026-11-02", publish_before="2026-11-09")  # the calendar
staqly.posts.search("half price", collection_id=post.collection_id)

for p in staqly.posts.iterate(channel_id=x.id, status="published"):
    print(p.content)

staqly.posts.history(post.id)  # its events, newest first
staqly.posts.tags()  # every tag with its count

providers = staqly.providers.list()  # the catalog: auth, status, limits, fields
# your own automation; token platforms take their credentials in `fields` once they are available
publisher = staqly.channels.connect(provider="custom", name="My publisher")
staqly.channels.test(publisher.id)  # against the platform; marks needs_reconnect on failure
staqly.channels.disable(publisher.id)  # or enable(), update(id, name=..., settings=...)

schedule = staqly.schedules.create(
    "Product updates",
    name="Weekday mornings",
    kind="recurring",
    rule={"times": ["09:00"], "weekdays": [1, 2, 3, 4, 5], "tz": "Europe/Warsaw"},
    channels=[publisher.id],
    pick="next",
    source="agent",
)
staqly.schedules.run(schedule.id)  # now; the regular next run is unchanged

from staqly import WaitTimeoutError

try:
    post = staqly.posts.wait_for_published(post_id, timeout=30 * 60)  # seconds, default 10 min

    if post.status == "failed":
        for t in post.targets:
            if t.status == "failed":
                print(t.channel_name, t.error)
except WaitTimeoutError as e:
    print("still publishing", e.post)

for w in staqly.work.list():
    if w.kind == "target":
        try:
            url = publish_somewhere(w.post, w.channel)
            staqly.posts.report_result(w.post.id, w.channel.id, status="published", url=url)
        except Exception as e:
            staqly.posts.report_result(w.post.id, w.channel.id, status="failed", error=str(e))
    else:
        # a schedule found nothing to publish: write drafts into w.collection
        staqly.posts.create(w.collection.id, content=draft_one())

import os

from fastapi import FastAPI, Request, Response
from staqly.webhooks import WebhookVerificationError, verify_webhook

app = FastAPI()


@app.post("/hooks/staqly")
async def staqly_hook(request: Request) -> Response:
    body = await request.body()  # the raw body, exactly as received

    try:
        event = verify_webhook(os.environ["STAQLY_WEBHOOK_SECRET"], body, request.headers)
    except WebhookVerificationError:
        return Response("invalid signature", status_code=400)

    if event.type == "post.publish_requested":
        await publish_myself(event.data["post"], event.data["channel"])

    return Response("ok")

from staqly import ConflictError

try:
    staqly.posts.update(post_id, content="x", expected_version=3)
except ConflictError:
    pass  # someone edited the post meanwhile; reload it

import asyncio

from staqly import AsyncStaqly


async def main() -> None:
    async with AsyncStaqly() as staqly:
        post = await staqly.posts.create("Drafts", content="Hello", targets=[{"channel_id": channel_id}])
        await staqly.posts.publish(post.id)
        done = await staqly.posts.wait_for_published(post.id)
        print(done.status)

        async for p in staqly.posts.iterate(status="published"):
            print(p.content)


asyncio.run(main())