Python SDK - Xquik

Use the Python SDK for typed sync or async access to Xquik from scripts, notebooks, workers, agent tools, and backend services. Use this page when you need Python to search tweets, scrape tweets to CSV, JSON Lines, or XLSX, export followers, post tweets, upload media, send direct messages, monitor tweets, or hand X data to pandas, a warehouse, a queue, CRM, or an agent workflow.

Install

pip install x_twitter_scraper

Authenticate

export X_TWITTER_SCRAPER_API_KEY="xq_YOUR_KEY_HERE"

import os
from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper(
    api_key=os.environ.get("X_TWITTER_SCRAPER_API_KEY"),
)

Basic Example

Search tweets and write durable JSON Lines handoff rows:

import json
import sys

from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper()

page = client.x.tweets.search(
    q="from:xquikcom webhook OR SDK",
    limit=10,
)

for tweet in page.tweets:
    row = {
        "tweet_id": tweet.id,
        "text": tweet.text,
        "author_username": tweet.author.username if tweet.author else None,
        "created_at": tweet.created_at,
    }
    sys.stdout.write(json.dumps(row, ensure_ascii=False) + "\n")

Async clients use the same generated method names:

import asyncio
import json
import sys

from x_twitter_scraper import AsyncXTwitterScraper

client = AsyncXTwitterScraper()

async def main() -> None:
    page = await client.x.tweets.search(q="xquik", limit=10)
    for tweet in page.tweets:
        row = {
            "tweet_id": tweet.id,
            "text": tweet.text,
            "author_username": tweet.author.username if tweet.author else None,
            "created_at": tweet.created_at,
        }
        sys.stdout.write(json.dumps(row, ensure_ascii=False) + "\n")

asyncio.run(main())

Workflow: Search Tweets to CSV, JSON Lines, or XLSX

This job is for data scripts, notebooks, scheduled workers, and agent tools that need tweet search results in a durable handoff file. It calls GET /x/tweets/search through client.x.tweets.search, uses the generated TweetSearchParams shape, and writes analyst-friendly CSV plus JSON Lines for queues, warehouses, and replayable processing.

import csv
import json
from pathlib import Path

from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper()
query = "from:xquikcom webhook OR SDK"
csv_output = Path("xquik-tweet-search.csv")
jsonl_output = Path("xquik-tweet-search.jsonl")
cursor = None
page_index = 0

with csv_output.open("w", newline="", encoding="utf-8") as csv_file, jsonl_output.open(
    "w",
    encoding="utf-8",
) as jsonl_file:
    writer = csv.DictWriter(
        csv_file,
        fieldnames=[
            "source",
            "query",
            "tweet_id",
            "text",
            "author_id",
            "author_username",
            "author_name",
            "created_at",
            "like_count",
            "reply_count",
            "retweet_count",
            "quote_count",
            "view_count",
            "bookmark_count",
            "is_note_tweet",
            "page_index",
            "page_cursor",
            "next_cursor",
            "has_next_page",
        ],
    )
    writer.writeheader()

    while True:
        page_cursor = cursor
        page = client.x.tweets.search(
            q=query,
            query_type="Latest",
            cursor=cursor,
        )

        for tweet in page.tweets:
            row = {
                "source": "xquik.python.search",
                "query": query,
                "tweet_id": tweet.id,
                "text": tweet.text,
                "author_id": tweet.author.id if tweet.author else None,
                "author_username": tweet.author.username if tweet.author else None,
                "author_name": tweet.author.name if tweet.author else None,
                "created_at": tweet.created_at,
                "like_count": tweet.like_count or 0,
                "reply_count": tweet.reply_count or 0,
                "retweet_count": tweet.retweet_count or 0,
                "quote_count": tweet.quote_count or 0,
                "view_count": tweet.view_count or 0,
                "bookmark_count": tweet.bookmark_count or 0,
                "is_note_tweet": tweet.is_note_tweet or False,
                "page_index": page_index,
                "page_cursor": page_cursor,
                "next_cursor": page.next_cursor or None,
                "has_next_page": page.has_next_page,
            }
            writer.writerow(row)
            jsonl_file.write(json.dumps(row, ensure_ascii=False) + "\n")

        if not page.has_next_page or not page.next_cursor:
            break
        cursor = page.next_cursor
        page_index += 1

The method accepts the same query inputs as the REST endpoint:

q

Python argument q maps to REST q. Use it for the required X search query with keywords, handles, hashtags, or operators.

limit

Python argument limit maps to REST limit. Use it as a 1 to 200 upper bound for a bounded pull. If page.has_next_page is true, keep the same q, filters, query_type, and limit when you continue with page.next_cursor.

cursor

Python argument cursor maps to REST cursor. Pass the opaque cursor from page.next_cursor to request the next page.

since_time

Python argument since_time maps to REST sinceTime. Use it as the ISO 8601 lower time bound.

until_time

Python argument until_time maps to REST untilTime. Use it as the ISO 8601 upper time bound.

query_type

Python argument query_type maps to REST queryType. Use Latest for chronological results or Top for engagement-ranked results.

Returned Data & Handoff

client.x.tweets.search returns a PaginatedTweets Pydantic model:

page.tweets

JSON field tweets. Contains SearchTweet records with id, text, optional author, created_at, like_count, reply_count, retweet_count, quote_count, bookmark_count, view_count, and is_note_tweet when available.

page.has_next_page

Python field page.has_next_page. JSON field has_next_page. Tells your script whether another page exists.

page.next_cursor

JSON field next_cursor. Store it only when page.has_next_page is true. For bounded pulls that return fewer tweets than limit, pass it back as cursor with the same query, filters, query_type, and limit.

Project page.tweets into CSV rows for analysts and JSON Lines rows for queues and data lakes in xquik-tweet-search.jsonl. Load the same projected rows into pandas or openpyxl when account teams need an XLSX workbook. Store tweet_id, author_username, engagement counts, page_index, page_cursor, next_cursor, and has_next_page so a script can resume from the last saved cursor without replaying raw SDK models. For explicit limit pulls, resume with the same query, filters, query_type, and limit; only cursor changes.

Workflow: Follower Export to CSV, JSON, or XLSX

Use this workflow when a Python script, notebook, worker, or agent tool needs an owned follower list for a CRM import, warehouse load, analyst CSV file, XLSX workbook, or resumable JSON handoff. It calls POST /extractions/estimate through client.extractions.estimate_cost, creates the job with client.extractions.run, reads saved rows with client.extractions.retrieve, and downloads files with client.extractions.export_results.

client.extractions.run returns the queued 202 Accepted receipt from POST /extractions: REST id, toolType, and status: "running" as Python job.id, job.tool_type, and job.status. Store job.id immediately, then poll client.extractions.retrieve before reading pages or calling client.extractions.export_results. Credit reservation happens after the job starts. If available credits changed since estimate_cost, the run can fetch only the affordable count before export or mark the job failed with insufficient_credits.

import json
import time
from pathlib import Path

from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper()
target_username = "xquikcom"

estimate = client.extractions.estimate_cost(
    tool_type="follower_explorer",
    target_username=target_username,
)

if not estimate.allowed:
    raise RuntimeError("Insufficient credits for follower export.")

job = client.extractions.run(
    tool_type="follower_explorer",
    target_username=target_username,
)

while True:
    status_page = client.extractions.retrieve(job.id, limit=1)
    status = status_page.job.get("status")

    if status == "completed":
        break

    if status == "failed":
        raise RuntimeError("Follower export failed.")

    time.sleep(10)

after = None

with Path("xquik-followers.jsonl").open("w", encoding="utf-8") as jsonl_file:
    while True:
        page = client.extractions.retrieve(job.id, limit=1000, after=after)

        for row in page.results:
            jsonl_file.write(json.dumps(row, ensure_ascii=False) + "\n")

        if not page.has_more or not page.next_cursor:
            break

        after = page.next_cursor

client.extractions.export_results(job.id, format="csv").write_to_file(
    Path("xquik-followers.csv"),
)
client.extractions.export_results(job.id, format="json").write_to_file(
    Path("xquik-followers.json"),
)
client.extractions.export_results(job.id, format="xlsx").write_to_file(
    Path("xquik-followers.xlsx"),
)

follower_explorer requires target_username. Persist job.id, target_username, estimate.estimated_results, and estimate.source before polling so a notebook restart, queue retry, or worker restart can resume the same follower export. client.extractions.retrieve returns results, has_more, and next_cursor; pass next_cursor back as after when you need stored JSON pages before exporting files. Use xquik-followers.jsonl for queue replay or warehouse loads, xquik-followers.json for app ingestion, xquik-followers.csv for CRM import, and xquik-followers.xlsx for analyst handoff. Map exported User ID or row xUserId as the CRM unique key. Cost: 1 credit per follower extracted or returned. Exports are free after the extraction job exists.

Workflow: Tweet Replies to CSV, JSON, or XLSX

Use this workflow when a Python script, notebook, worker, or agent tool needs every reply under one tweet as a saved extraction, JSON Lines handoff, or CSV/JSON/XLSX file export. It uses client.extractions.estimate_cost, run, retrieve, and export_results. Reuse the polling, row pagination, and export structure from the follower workflow. Only the tool type, target field, and output filenames change:

target_tweet_id = "1893704267862470862"
estimate = client.extractions.estimate_cost(
    tool_type="reply_extractor",
    target_tweet_id=target_tweet_id,
)

if not estimate.allowed:
    raise RuntimeError("Insufficient credits for reply extraction.")

job = client.extractions.run(
    tool_type="reply_extractor",
    target_tweet_id=target_tweet_id,
)

# Run the same polling and JSONL pagination loops from the follower workflow.
# Set the JSON Lines destination to Path("xquik-replies.jsonl").

client.extractions.export_results(job.id, format="csv").write_to_file(
    Path("xquik-replies.csv"),
)
client.extractions.export_results(job.id, format="json").write_to_file(
    Path("xquik-replies.json"),
)
client.extractions.export_results(job.id, format="xlsx").write_to_file(
    Path("xquik-replies.xlsx"),
)

reply_extractor requires target_tweet_id. client.extractions.retrieve returns results, has_more, and next_cursor; the shared pagination loop passes next_cursor back as after. Use xquik-replies.jsonl for queue replay or warehouse loads, xquik-replies.json for app ingestion, xquik-replies.csv for CRM import, and xquik-replies.xlsx for analyst handoff. client.extractions.export_results supports csv, json, and xlsx for file handoff. Cost: 1 credit per reply extracted or returned.

Workflow: Post Media Tweets and DM Attachments

Use this workflow when a Python worker, notebook, support queue, or agent needs to post a media-backed tweet, reply with media, or send one uploaded media item in a DM. Tweet and reply media posts use public media URLs directly on client.x.tweets.create. Send up to 4 image URLs or exactly 1 MP4 video URL up to 100 MB. Do not mix video with other media. Do not upload first when the media URL is already public.

import json
import sys
from typing import Any

from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper()

def create_tweet_handoff(
    payload: dict[str, Any],
    base: dict[str, Any],
) -> dict[str, Any]:
    if (
        payload.get("error") == "x_write_unconfirmed"
        and payload.get("status") == "pending_confirmation"
    ):
        return {
            "status": "pending_confirmation",
            "write_action_id": payload["writeActionId"],
            "charged": payload["charged"],
            "charged_credits": payload["chargedCredits"],
            "retryable": payload["retryable"],
            "poll": "GET /x/write-actions/{id}",
            **base,
        }

    return {
        "status": "posted",
        "tweet_id": payload["tweetId"],
        "charged": payload["charged"],
        "charged_credits": payload["chargedCredits"],
        **(
            {"write_action_id": payload["writeActionId"]}
            if "writeActionId" in payload
            else {}
        ),
        **base,
    }

The generated Python model covers confirmed tweet_id responses. Use with_raw_response.create when a write worker must branch on the REST 202 x_write_unconfirmed response, store writeActionId and chargedCredits, and poll Get Write Action Status before sending another write.

response = client.x.tweets.with_raw_response.create(
    account="@xquikcom",
    text="New demo video is live.",
    media=["https://example.com/product-demo.mp4"],
)
payload = response.json()

tweet_handoff = create_tweet_handoff(
    payload,
    {
        "account": "@xquikcom",
        "media": ["https://example.com/product-demo.mp4"],
    },
)

sys.stdout.write(json.dumps(tweet_handoff, ensure_ascii=False) + "\n")

To post an image reply, add reply_to_tweet_id:

response = client.x.tweets.with_raw_response.create(
    account="@xquikcom",
    text="Here is the requested screenshot.",
    reply_to_tweet_id="1893704267862470862",
    media=["https://example.com/export-preview.png"],
)
payload = response.json()

reply_handoff = create_tweet_handoff(
    payload,
    {
        "account": "@xquikcom",
        "reply_to_tweet_id": "1893704267862470862",
        "media": ["https://example.com/export-preview.png"],
    },
)

sys.stdout.write(json.dumps(reply_handoff, ensure_ascii=False) + "\n")

For DM attachments, upload the local file first and pass the returned media.media_id as the only media_ids item:

import json
import sys
from pathlib import Path

media = client.x.media.upload(
    account="@xquikcom",
    file=Path("./handoff.png"),
)

dm = client.x.dm.send(
    user_id="44196397",
    account="@xquikcom",
    text="Here is the asset.",
    media_ids=[media.media_id],
)

dm_handoff = {
    "status": "sent",
    "message_id": dm.message_id,
    "media_id": media.media_id,
    "account": "@xquikcom",
    "user_id": "44196397",
}

sys.stdout.write(json.dumps(dm_handoff, ensure_ascii=False) + "\n")

client.x.tweets.create returns tweet.tweet_id for confirmed posts. Raw create responses can also include the pending write fields above when confirmation is still running. client.x.media.upload returns media.media_id for DM attachments, and client.x.dm.send returns dm.message_id for support tickets, CRM records, queue jobs, or agent memory. Keep DM body text in private systems. Shared logs, public artifacts, queue status, and agent handoffs should store message_id, optional media_id, account, user_id, and send status instead of full DM bodies. Leave reply_to_message_id unset even if generated SDK types expose it; the REST endpoint rejects DM reply threading. Text-only tweet and reply writes cost 10 credits. Tweet media adds 2 credits per started MB across attached files. Uploading media costs 10 credits, and sending the DM costs 10 credits. Do not pass uploaded media.media_id values to client.x.tweets.create; that method uses media with public media URLs.

Cost, Limits & Retries

Tweet search costs 1 credit per tweet returned. If remaining credits cannot cover a bounded limit request, the API can return fewer tweets; if 0 paid results are affordable, it returns 402 insufficient_credits. Read calls are rate-limited, and 429 responses include Retry-After. The client retries connection errors, 408, 409, 429, and 5xx responses by default. Handle RateLimitError with backoff, and fix 400, 401, 403, 404, or 422 responses before retrying.

Error Handling

All SDK exceptions inherit from x_twitter_scraper.APIError.

400 Bad Request

Throws BadRequestError.

401 Unauthenticated

Throws AuthenticationError.

403 Permission Denied

Throws PermissionDeniedError.

404 Not Found

Throws NotFoundError.

422 Unprocessable Entity

Throws UnprocessableEntityError.

429 Rate Limited

Throws RateLimitError.

5xx Server Error

Throws InternalServerError.

import x_twitter_scraper

try:
    account = client.account.retrieve()
except x_twitter_scraper.APIConnectionError as exc:
    print(f"Connection failed: {exc.__cause__}")
except x_twitter_scraper.APIStatusError as exc:
    print(f"HTTP {exc.status_code}")

The client retries connection errors, 408, 409, 429, and 5xx responses by default. Pass max_retries when constructing the client to change retry behavior.

Pagination

List endpoints return page models with has_next_page and cursor fields.

page = client.x.tweets.search(q="xquik", limit=20)

if page.has_next_page:
    print("More results are available")

​Install

​Authenticate

​Basic Example

​Workflow: Search Tweets to CSV, JSON Lines, or XLSX

q