> ## 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. # Python SDK > Use the Xquik Python SDK to search tweets, export CSV, JSON Lines, or XLSX, post tweets, upload media, send DMs, and build X API workers.
For the complete documentation index, see llms.txt.
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 ```bash theme={null} pip install x_twitter_scraper ``` ## Authenticate ```bash theme={null} export X_TWITTER_SCRAPER_API_KEY="xq_YOUR_KEY_HERE" ``` ```python theme={null} 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: ```python theme={null} import json import sys from x_twitter_scraper import XTwitterScraper client = XTwitterScraper() page = client.x.tweets.search( q="from:username 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: ```python theme={null} 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. ```python theme={null} import csv import json from pathlib import Path from x_twitter_scraper import XTwitterScraper client = XTwitterScraper() query = "from:username 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: Python argument `q` maps to REST `q`. Use it for the required X search query with keywords, handles, hashtags, or operators. 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`. Python argument `cursor` maps to REST `cursor`. Pass the opaque cursor from `page.next_cursor` to request the next page. Python argument `since_time` maps to REST `sinceTime`. Use it as the ISO 8601 lower time bound. Python argument `until_time` maps to REST `untilTime`. Use it as the ISO 8601 upper time bound. 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: 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. Python field `page.has_next_page`. JSON field `has_next_page`. Tells your script whether another page exists. 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`. ```python theme={null} import json import time from pathlib import Path from x_twitter_scraper import XTwitterScraper client = XTwitterScraper() target_username = "username" 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: ```python theme={null} 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. ```python theme={null} import json import sys from typing import Any from x_twitter_scraper import XTwitterScraper client = XTwitterScraper() def create_tweet_handoff( action: dict[str, Any], base: dict[str, Any], ) -> dict[str, Any]: result = action.get("result") or {} return { "status": action["status"], "terminal": action["terminal"], "safe_to_retry": action["safeToRetry"], "write_action_id": action["id"], "request_hash": action["request"]["hash"], "tweet_id": result.get("id") or action.get("tweetId"), "charged": action["billing"]["charged"], "charged_credits": action["billing"]["chargedCredits"], "poll": None if action["terminal"] else action["statusUrl"], **base, } ``` Use `with_raw_response.create` to capture the durable write action. Send a unique `Idempotency-Key`, store `id`, `request.hash`, `billing`, `result`, and `statusUrl`, then poll while `terminal` is false. Retry only when `safeToRetry` is true, using a new key. ```python theme={null} response = client.x.tweets.with_raw_response.create( account="@username", text="New demo video is live.", media=["https://example.com/product-demo.mp4"], ) payload = response.json() tweet_handoff = create_tweet_handoff( payload, { "account": "@username", "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`: ```python theme={null} response = client.x.tweets.with_raw_response.create( account="@username", 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": "@username", "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: ```python theme={null} import json import sys from pathlib import Path media = client.x.media.upload( account="@username", file=Path("./handoff.png"), ) dm = client.x.dm.send( user_id="44196397", account="@username", 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": "@username", "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 30 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`. Throws `BadRequestError`. Throws `AuthenticationError`. Throws `PermissionDeniedError`. Throws `NotFoundError`. Throws `UnprocessableEntityError`. Throws `RateLimitError`. Throws `InternalServerError`. ```python theme={null} 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. ```python theme={null} page = client.x.tweets.search(q="xquik", limit=20) if page.has_next_page: print("More results are available") ``` ## Webhooks & References * [Search Tweets](/api-reference/x/search-tweets) * [Create Extraction](/api-reference/extractions/create) * [Get Extraction](/api-reference/extractions/get) * [Export Extraction](/api-reference/extractions/export) * [Create Tweet](/api-reference/x-write/create-tweet) * [Get Write Action Status](/api-reference/x-write/get-write-action-status) * [Upload Media](/api-reference/x-write/upload-media) * [Send Direct Message](/api-reference/x-write/send-dm) * [Webhook Overview](/webhooks/overview) * [Verify HMAC Signatures](/webhooks/verification) * [REST API Overview](/api-reference/overview) * [OpenAPI JSON](https://xquik.com/openapi.json) * [Source Repository](https://github.com/Xquik-dev/x-twitter-scraper-python)