How to Upload Media
with the X API v2

import os
import time
import requests

TOKEN = os.environ["X_USER_ACCESS_TOKEN"]  # OAuth 2.0 user token with media.write
BASE = "https://api.x.com/2/media/upload"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
CHUNK = 4 * 1024 * 1024  # stay under the 5MB per-request cap

def upload_media(path, media_type, media_category):
    total = os.path.getsize(path)

    init = requests.post(BASE, headers=HEADERS, data={
        "command": "INIT",
        "media_type": media_type,
        "total_bytes": total,
        "media_category": media_category,
    })
    init.raise_for_status()
    media_id = init.json()["data"]["id"]

    with open(path, "rb") as f:
        index = 0
        while chunk := f.read(CHUNK):
            requests.post(BASE, headers=HEADERS, data={
                "command": "APPEND",
                "media_id": media_id,
                "segment_index": index,
            }, files={"media": chunk}).raise_for_status()
            index += 1

    fin = requests.post(BASE, headers=HEADERS, data={
        "command": "FINALIZE",
        "media_id": media_id,
    })
    fin.raise_for_status()

    info = fin.json()["data"].get("processing_info")
    while info and info["state"] in ("pending", "in_progress"):
        time.sleep(info.get("check_after_secs", 5))
        status = requests.get(BASE, headers=HEADERS, params={
            "command": "STATUS",
            "media_id": media_id,
        })
        status.raise_for_status()
        info = status.json()["data"].get("processing_info")

    if info and info["state"] == "failed":
        raise RuntimeError(f"Media processing failed: {info}")
    return media_id

image_id = upload_media("chart.png", "image/png", "tweet_image")
video_id = upload_media("demo.mp4", "video/mp4", "tweet_video")

tweet = requests.post(
    "https://api.x.com/2/tweets",
    headers={**HEADERS, "Content-Type": "application/json"},
    json={"text": "New demo video is up.", "media": {"media_ids": [video_id]}},
)
print(tweet.json())

import { readFile } from "node:fs/promises";

const TOKEN = process.env.X_USER_ACCESS_TOKEN; // OAuth 2.0 user token with media.write
const BASE = "https://api.x.com/2/media/upload";
const CHUNK = 4 * 1024 * 1024;

async function xUpload(form) {
  const res = await fetch(BASE, {
    method: "POST",
    headers: { Authorization: `Bearer ${TOKEN}` },
    body: form,
  });
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  return res.json();
}

export async function uploadMedia(path, mediaType, mediaCategory) {
  const buffer = await readFile(path);

  const init = new FormData();
  init.append("command", "INIT");
  init.append("media_type", mediaType);
  init.append("total_bytes", String(buffer.length));
  init.append("media_category", mediaCategory);
  const { data } = await xUpload(init);
  const mediaId = data.id;

  for (let i = 0; i * CHUNK < buffer.length; i++) {
    const form = new FormData();
    form.append("command", "APPEND");
    form.append("media_id", mediaId);
    form.append("segment_index", String(i));
    form.append("media", new Blob([buffer.subarray(i * CHUNK, (i + 1) * CHUNK)]));
    await xUpload(form);
  }

  const fin = new FormData();
  fin.append("command", "FINALIZE");
  fin.append("media_id", mediaId);
  const finalized = await xUpload(fin);

  let info = finalized.data.processing_info;
  while (info && (info.state === "pending" || info.state === "in_progress")) {
    await new Promise((r) => setTimeout(r, (info.check_after_secs ?? 5) * 1000));
    const res = await fetch(`${BASE}?command=STATUS&media_id=${mediaId}`, {
      headers: { Authorization: `Bearer ${TOKEN}` },
    });
    info = (await res.json()).data.processing_info;
  }
  if (info && info.state === "failed") throw new Error("Media processing failed");
  return mediaId;
}

const videoId = await uploadMedia("demo.mp4", "video/mp4", "tweet_video");

await fetch("https://api.x.com/2/tweets", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${TOKEN}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    text: "New demo video is up.",
    media: { media_ids: [videoId] },
  }),
});

curl -X POST https://opentweet.io/api/v1/upload \
  -H "Authorization: Bearer ot_your_key" \
  -F "file=@demo.mp4"

{
  "url": "https://cdn.opentweet.io/media/demo.mp4",
  "type": "video",
  "content_type": "video/mp4",
  "size_bytes": 10485760,
  "filename": "demo.mp4"
}

// 1. Upload once, get back a URL
const form = new FormData();
form.append("file", new Blob([await readFile("demo.mp4")], { type: "video/mp4" }), "demo.mp4");

const up = await fetch("https://opentweet.io/api/v1/upload", {
  method: "POST",
  headers: { "Authorization": `Bearer ${process.env.OPENTWEET_API_KEY}` },
  body: form,
});
const { url } = await up.json();

// 2. Attach it to a post via media_urls
await fetch("https://opentweet.io/api/v1/posts", {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${process.env.OPENTWEET_API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    text: "New demo video is up.",
    media_urls: [url],
    publish_now: true,
  }),
});