Post one thread to X, Threads and a Facebook Page

Schedule on X, then use its publication webhook to trigger immediate Threads and Facebook posts.

• ✓ 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.
• ✓ Connected X and Threads accounts and a Facebook Page; threads:write, publish:read, publish:write and webhooks:manage scopes.
• ✓ An HTTPS receiver using the documented signature verifier, a durable work queue and a campaign record in your storage.

1. Check accounts and select a Page

   Check connected and needs_reauth flags before posting. Facebook pages are returned as a bare array; choose the intended page_id explicitly.

   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()
   
   accounts = api("GET", "/accounts/")
   pages = api("GET", "/facebook/pages/")
   print(accounts)
   for page in pages:
       print(page["page_id"], page["name"], page["status"])
   page_id = input("Facebook page_id to publish to: ")
   if page_id not in {page["page_id"] for page in pages}:
       raise ValueError("Choose a Page returned by the API")

2. Create the approved thread

   Save the reviewed content as a manual thread, or use POST /threads/generate/ with a topic and review the result. Store its id and chosen Page in a campaign record before scheduling.

   thread = api("POST", "/threads/", json={
       "topic": "What we learned shipping our API",
       "content": "Start with clear errors.\n\nPreview before publishing.\n\nKeep retries safe."
   })
   print(thread["id"])
   # Persist your campaign's thread id and page_id in your own storage.

3. Subscribe, then schedule the X job

   Register the receiver once, then schedule X so its publication webhook can trigger the other platforms. Save the scheduled time and UUID for retries; omitting scheduled_for would publish to X immediately.

   from datetime import datetime, timedelta, timezone
   from uuid import uuid5, NAMESPACE_URL
   
   webhook = api("POST", "/webhooks/", json={
       "url": "https://your-site.example/threadmaster-events",
       "events": ["post.published", "post.failed"]
   })
   # Store webhook["secret"] securely; register only once per integration.
   scheduled_for = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()
   key = str(uuid5(NAMESPACE_URL, f"https://your-site.example/campaign/{thread['id']}/x"))
   job = api("POST", "/x/posts/", json={
       "thread_id": thread["id"], "scheduled_for": scheduled_for
   }, headers={"Idempotency-Key": key})
   print(job["id"], job["status"])
   # Save the X job id, time and key with the campaign.

4. Publish to Threads and Facebook from a worker

   Verify the webhook, save and queue it once per event id, then return 2xx promptly. For post.published matching the campaign's X job id, run the worker with the saved Page id to publish immediately to Threads and Facebook, using separate UUIDs.

   def cross_post(event, x_job_id, page_id):
       job = event["data"]
       if (event["type"] != "post.published" or job["platform"] != "x"
               or job["id"] != x_job_id):
           return
       thread_id = job["thread_id"]
       def key(destination):
           return str(uuid5(NAMESPACE_URL, f"https://your-site.example/x-job/{job['id']}/{destination}"))
       threads_job = api("POST", "/threads-app/posts/", json={
           "thread_id": thread_id
       }, headers={"Idempotency-Key": key("threads")})
       facebook_job = api("POST", "/facebook/posts/", json={
           "thread_id": thread_id, "page_id": page_id, "mode": "comments"
       }, headers={"Idempotency-Key": key("facebook-" + page_id)})
       return threads_job, facebook_job

   Connect your queue and storage, then save and check each job's status, error_code and permalink. An accepted request can still return a failed or partial job.

5. Handle connection and quota errors

   Catch HTTP errors in your worker: pause for reconnection on 409 account_not_connected and check quota and its reset time on 429 quota_exhausted. Reuse the destination keys when resuming so retries stay safe, and pass other errors to your retry or alerting policy.

   def run_campaign_event(event, x_job_id, page_id):
       try:
           return cross_post(event, x_job_id, page_id)
       except requests.HTTPError as exc:
           response = exc.response
           error = response.json()["error"]
           if response.status_code == 409 and error["code"] == "account_not_connected":
               print("Reconnect the account in Settings before resuming")
           elif response.status_code == 429 and error["code"] == "quota_exhausted":
               print("Pause and review quota:", api("GET", "/quota/"))
           raise  # Let your queue retain the work for an explicit retry policy

Use the documented Python or Node signature verification snippet in your receiver.

• Threads and Facebook are immediate-only. Schedule X, then trigger the other platforms from its post.published event.
• Handle 409 account_not_connected and 429 quota_exhausted; also inspect successful responses for failed or partial job statuses.
• Facebook mode single publishes the whole thread as one Page post; comments publishes a head post followed by a comment chain.
• Filter by platform and the saved X job id to avoid reacting to your own Threads/Facebook events. Keep a separate Idempotency-Key for each destination.

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