Schedule a month of posts from a spreadsheet

Export a CSV, save each draft, and queue a month of X posts with a repeatable Python script.

• ✓ An active ThreadMaster subscription and a key from Settings → Developer API; set THREADMASTER_API_KEY in your environment.
• ✓ Python 3 with requests installed (python -m pip install requests). Run each recipe's Python blocks in order in the same script.
• ✓ A connected X account and threads:write, publish:read and publish:write scopes.
• ✓ A calendar.csv file with date,topic,content columns; a writable directory for calendar-state.json.

1. Export dates, topics and content

   Export calendar.csv with a topic, content and an ISO-8601 date including a timezone offset in every row. Replace the example dates with future dates so the script can schedule your posts.

   date,topic,content
   2027-01-05T09:00:00-08:00,API lessons,"Start with clear errors. Make retries safe."
   2027-01-07T09:00:00-08:00,Preview first,"Review the split before scheduling your thread."

2. Save drafts and schedule each row

   The script checks monthly quota, saves each thread id and uses a stable publishing UUID per row so retries skip saved jobs. Run one copy at a time, keeping the CSV filename, row order and calendar-state.json unchanged.

   import os
   import requests
   
   BASE = "https://threadmaster.ai/api/public/v1"
   session = requests.Session()
   session.headers.update({"Authorization": "Bearer " + os.environ["THREADMASTER_API_KEY"]})
   
   def api(method, path, **kwargs):
       response = session.request(method, BASE + path, timeout=90, **kwargs)
       response.raise_for_status()
       return response.json()
   
   import csv, json, time
   from pathlib import Path
   from datetime import datetime, timedelta, timezone
   from uuid import uuid5, NAMESPACE_URL
   
   source = Path("calendar.csv")
   journal = Path("calendar-state.json")
   state = json.loads(journal.read_text()) if journal.exists() else {}
   def save():
       temporary = journal.with_suffix(".tmp")
       temporary.write_text(json.dumps(state, indent=2))
       temporary.replace(journal)
   
   quota = api("GET", "/quota/")
   with source.open(newline="") as handle:
       for number, row in enumerate(csv.DictReader(handle), start=1):
           key = str(uuid5(NAMESPACE_URL, f"{source.resolve().as_uri()}#row-{number}"))
           saved = state.setdefault(key, {"row": row})
           if saved["row"] != row:
               raise ValueError("Row changed: manage the saved job before reimporting")
           if "job_id" in saved:
               continue
           when = datetime.fromisoformat(row["date"])
           if when.utcoffset() is None or when < datetime.now(timezone.utc) + timedelta(minutes=2):
               raise ValueError("Use an offset and a date at least two minutes ahead")
           if not quota["unlimited"] and quota["remaining_posts"] <= 0:
               raise RuntimeError("Review monthly quota before continuing")
           if "thread_id" not in saved:
               draft = api("POST", "/threads/", json={
                   "topic": row["topic"], "content": row["content"]
               })
               saved["thread_id"] = draft["id"]
               save()
           while True:
               try:
                   job = api("POST", "/x/posts/", json={
                       "thread_id": saved["thread_id"], "scheduled_for": row["date"]
                   }, headers={"Idempotency-Key": key})
                   break
               except requests.HTTPError as exc:
                   response = exc.response
                   if response is None or response.status_code != 429:
                       raise
                   error = response.json()["error"]
                   if error["code"] == "quota_exhausted":
                       raise RuntimeError("Monthly publishing quota exhausted; review quota before continuing") from exc
                   if error["code"] != "rate_limited":
                       raise
                   details = error.get("details", {})
                   time.sleep(details.get("retry_after", 60))
           saved["job_id"] = job["id"]
           save()
           print(number, job["id"], job["status"])
           quota = api("GET", "/quota/")

   Keep calendar-state.json for safe retries; a crash before saving a draft id can leave an unused draft. The script waits for retry_after when the hourly publishing limit is hit.

3. List the scheduled jobs

   Follow the cursor to inspect every scheduled X job, including its timestamp and thread id.

   params = {"status": "scheduled", "limit": 200}
   while True:
       page = api("GET", "/x/posts/", params=params)
       for job in page["data"]:
           print(job["id"], job["thread_id"], job["scheduled_for"])
       if not page["has_more"]:
           break
       params["cursor"] = page["next_cursor"]

4. Move or cancel a scheduled job

   Use a job id from the listing. Choose either reschedule or cancel; edit the publishing job rather than reimporting a changed CSV row.

   job_id = int(input("Scheduled job id: "))
   action = input("reschedule or cancel: ")
   if action == "reschedule":
       when = input("New ISO-8601 timestamp with offset (at least 2 minutes ahead): ")
       print(api("POST", f"/x/posts/{job_id}/reschedule/", json={"scheduled_for": when}))
   elif action == "cancel":
       print(api("POST", f"/x/posts/{job_id}/cancel/"))

• Always send ISO-8601 timestamps with an offset and allow at least two minutes before publishing.
• Check GET /quota/ for your monthly allowance; the script waits for retry_after when the hourly publishing limit is hit. A month-long calendar does not bypass quotas.
• Reuse the Idempotency-Key per row and retain calendar-state.json. Do not reorder rows or move the CSV when retrying this example.

Check the API rate limits and quotas before running your workflow at scale.