Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.xquik.com/llms.txt

Use this file to discover all available pages before exploring further.

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
from pathlib import Path

from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper()
csv_output = Path("xquik-tweet-search.csv")
jsonl_output = Path("xquik-tweet-search.jsonl")
cursor = None

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=[
            "id",
            "text",
            "author_username",
            "created_at",
            "like_count",
            "reply_count",
            "retweet_count",
            "quote_count",
            "view_count",
        ],
    )
    writer.writeheader()

    while True:
        page = client.x.tweets.search(
            q="from:xquikcom webhook OR SDK",
            query_type="Latest",
            cursor=cursor,
        )

        for tweet in page.tweets:
            row = {
                "id": tweet.id,
                "text": tweet.text,
                "author_username": tweet.author.username if tweet.author else "",
                "created_at": tweet.created_at or "",
                "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,
            }
            writer.writerow(row)
            jsonl_file.write(tweet.to_json(indent=None) + "\n")

        if not page.has_next_page or not page.next_cursor:
            break
        cursor = page.next_cursor
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 for a bounded request from 1 to 200. Omit it for cursor loops.

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 with the job checkpoint and pass it back as cursor.
Write page.tweets to CSV for analysts. Write JSON Lines for queues and data lakes to xquik-tweet-search.jsonl with tweet.to_json(indent=None), or use .to_dict() records before loading a warehouse. Load the projected rows into pandas or openpyxl when account teams need an XLSX workbook. Pass id, text, author.username, created_at, and engagement counts to a CRM or enrichment job.

​
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 calls POST /extractions/estimate through client.extractions.estimate_cost, creates the job with client.extractions.run, reads rows with client.extractions.retrieve, and downloads files with client.extractions.export_results. 
import json
import time
from pathlib import Path

from x_twitter_scraper import XTwitterScraper

client = XTwitterScraper()
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,
)

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("Reply extraction failed.")

    time.sleep(3)

after = None

with Path("xquik-tweet-replies.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-tweet-replies.csv"),
)
reply_extractor requires target_tweet_id. client.extractions.retrieve returns results, has_more, and next_cursor; store those values when you need resumable JSON pagination. 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"],
            "retryable": payload["retryable"],
            "poll": "GET /x/write-actions/{id}",
            **base,
        }

    return {
        "status": "posted",
        "tweet_id": payload["tweetId"],
        **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 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. Each tweet or reply write costs 10 credits. 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")

​
Webhooks & References

Last modified on May 16, 2026