Publish Your First Threads Post with Python

"""threads_client.py - a tiny client for the Threads API."""
import os
import time
from functools import lru_cache

import requests
from dotenv import load_dotenv

load_dotenv()

BASE_URL = "https://graph.threads.net/v1.0"
TOKEN = os.environ["THREADS_ACCESS_TOKEN"]


class ThreadsError(Exception):
    """Raised when the Threads API returns an error."""


def api(method, path, **params):
    """Call the Threads API and return the JSON response."""
    params["access_token"] = TOKEN
    if method == "GET":
        resp = requests.get(f"{BASE_URL}/{path}", params=params, timeout=30)
    else:
        resp = requests.post(f"{BASE_URL}/{path}", data=params, timeout=30)
    data = resp.json()
    if "error" in data:
        err = data["error"]
        raise ThreadsError(f"[{err.get('code')}] {err.get('message')}")
    return data


@lru_cache(maxsize=1)
def get_me():
    """Return the connected account's id and username (cached)."""
    return api("GET", "me", fields="id,username")


def wait_until_ready(container_id, timeout=300):
    """Poll a media container until Threads has finished processing it."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        info = api("GET", container_id, fields="status,error_message")
        status = info.get("status")
        if status == "FINISHED":
            return
        if status in ("ERROR", "EXPIRED"):
            raise ThreadsError(f"Container {status}: {info.get('error_message')}")
        time.sleep(5)
    raise ThreadsError("Timed out waiting for media to process")


def create_container(user_id, **fields):
    """Create a media container and return its id."""
    return api("POST", f"{user_id}/threads", **fields)["id"]


def publish(text="", image_urls=None, topic_tag=None):
    """Publish a text, image or carousel post. Returns (post_id, permalink)."""
    image_urls = image_urls or []
    if len(text) > 500:
        raise ValueError("Text is over the 500-character limit")
    if len(image_urls) > 20:
        raise ValueError("A carousel can have at most 20 items")

    user_id = get_me()["id"]
    extra = {"topic_tag": topic_tag} if topic_tag else {}

    if not image_urls:
        container = create_container(user_id, media_type="TEXT", text=text, **extra)
    elif len(image_urls) == 1:
        container = create_container(user_id, media_type="IMAGE",
                                     image_url=image_urls[0], text=text, **extra)
        wait_until_ready(container)
    else:
        children = []
        for url in image_urls:
            child = create_container(user_id, media_type="IMAGE",
                                     image_url=url, is_carousel_item="true")
            wait_until_ready(child)
            children.append(child)
        container = create_container(user_id, media_type="CAROUSEL",
                                     children=",".join(children), text=text, **extra)
        wait_until_ready(container)

    post_id = api("POST", f"{user_id}/threads_publish", creation_id=container)["id"]
    permalink = api("GET", post_id, fields="permalink").get("permalink")
    return post_id, permalink


def remaining_posts():
    """How many more posts this account can publish in the current 24 hours."""
    user_id = get_me()["id"]
    data = api("GET", f"{user_id}/threads_publishing_limit",
               fields="quota_usage,config")["data"][0]
    return data["config"]["quota_total"] - data["quota_usage"]

"""post.py - publish to Threads from the command line.

Usage:
  python post.py "Hello, Threads!"
  python post.py "New drop" --image https://example.com/a.jpg
  python post.py "Swipe ->" --image URL1 --image URL2 --topic Python
"""
import argparse
import sys

from threads_client import ThreadsError, get_me, publish, remaining_posts


def main():
    parser = argparse.ArgumentParser(description="Publish a post to Threads.")
    parser.add_argument("text", help="Post text (up to 500 characters)")
    parser.add_argument("--image", action="append", default=[],
                        help="Public image URL (repeat for a carousel)")
    parser.add_argument("--topic", help="Optional topic tag")
    args = parser.parse_args()

    try:
        me = get_me()
        print(f"→ Connected as @{me['username']}")
        if remaining_posts() < 1:
            sys.exit("✕ Daily publishing limit reached - try again later.")
        post_id, link = publish(args.text, args.image, args.topic)
        print(f"✓ Published {post_id}")
        print(f"  {link}")
    except (ThreadsError, ValueError) as err:
        sys.exit(f"✕ {err}")


if __name__ == "__main__":
    main()