Python 版 Twitter API：2026 年教程，附可运行的代码

#Tweepy 在 2026 年还能用吗？

能用。Tweepy 4.17.0 在 2026 年 7 月 2 日发布，X 现在也发布了自己的 Python SDK xdk，版本为 0.10.6。两者都是官方 X API 的客户端，所以它们发出的每个调用都需要开发者账号、应用的 bearer 令牌和预付额度。

额度是旧教程最容易漏掉的部分。新开发者没有免费额度，X 对响应里返回的每个对象计费：每条帖子 $0.005，每个用户 $0.010，每月最多读取 300 万条帖子（见 X 的价格表（在新标签页打开））。从 2026 年 11 月 22 日起，通过扩展字段带出的作者和其他对象按相同费率计费（见 X 在 10 月 23 日发布的通知（在新标签页打开）），所以常见的 Tweepy 搜索会为附带的每个作者付费：

import tweepy

client = tweepy.Client(bearer_token="YOUR_BEARER_TOKEN")
resp = client.search_recent_tweets(
    "python lang:en",
    max_results=100,
    expansions=["author_id"],
)
# 100 posts: $0.50. From Nov 22, 2026, with authors: up to $1.50

这个库本身免费且开源。它的主页至今还写着背后的 API 有免费和付费两档，对新账号来说已经不是这样了。所有费率、上限和日期都整理在 X API 价格一文里。

官方路线在第一次调用前需要什么

• 在 console.x.com 开通的开发者账号，用 X 账号开通，并同意 X 的开发者协议。
• 一个应用，它的 bearer 令牌 X 只显示一次（见 X 的接入指南（在新标签页打开））。
• 提前购买的额度，按返回的每条帖子和每个用户扣费。
• 发帖、点赞或关注等操作需要的 OAuth 用户令牌。

读取公开帖子和用户资料用不到这些。自带密钥的 Twitter/X 数据 API 通过普通 HTTPS 就能回答，脚本只需要 requests 这一个客户端。

#用一个 API 密钥在 Python 中获取推文

要从 Python 读取 X 数据，可以用 Sorsa API，一款 Twitter（X）API 与爬虫：用一个 ApiKey 请求头代替 OAuth，所有套餐（包括免费账号）都是每秒 20 次请求，搜索结果每 1,000 条推文 $0.07 起。它以 JSON 返回公开的帖子、用户资料和粉丝，每条推文里都带着作者的完整资料。它从不要求登录 X，所以你的 X 账号不会有任何风险。

准备密钥并发送第一个请求

1. 获取密钥。 注册后，在控制台复制你的 API 密钥。前 100 次请求免费，无需绑卡，永不过期。

2. 安装 requests 并保存密钥。 把密钥放在环境变量里，而不是写进代码，并且从脚本或服务器调用 API，永远不要从浏览器 JavaScript 调用。

   pip install requests
   export SORSA_API_KEY="your_key"  # PowerShell: $env:SORSA_API_KEY="your_key"
   

3. 搜索推文。 用 X 的搜索语法向 POST /search-tweets（在新标签页打开） 发送一条查询，密钥放在 ApiKey 请求头里。

   import os
   import requests
   
   API = "https://api.sorsa.io/v3"
   HEADERS = {"ApiKey": os.environ.get("SORSA_API_KEY", "YOUR_API_KEY")}
   
   r = requests.post(
       f"{API}/search-tweets",
       headers=HEADERS,
       json={"query": "python lang:en", "order": "latest"},
       timeout=30,
   )
   r.raise_for_status()
   for tweet in r.json()["tweets"]:
       print(tweet["user"]["username"], tweet["full_text"][:60])
   

   curl -X POST "https://api.sorsa.io/v3/search-tweets" \
     -H "ApiKey: YOUR_API_KEY" \
     -H "Content-Type: application/json" \
     -d '{"query": "python lang:en", "order": "latest"}'
   

4. 读取这一页。 每个响应最多包含 20 条推文，以及用于获取下一页的 next_cursor。下面的值是按文档格式给出的演示数据。

   {
     "tweets": [
       {
         "id": "1985432176540981234",
         "full_text": "Loaded a week of #python posts into pandas in four lines.",
         "created_at": "2026-11-01T18:42:07Z",
         "lang": "en",
         "likes_count": 42,
         "retweet_count": 7,
         "reply_count": 3,
         "quote_count": 1,
         "view_count": 3810,
         "bookmark_count": 5,
         "is_reply": false,
         "user": {
           "id": "1592041833471827968",
           "username": "py_data_notebook",
           "display_name": "Py Data Notebook",
           "followers_count": 1840,
           "verified": false
         }
       }
     ],
     "next_cursor": "DAABCgABF7Y8kQ3wAAA"
   }
   

   显示完整响应

每条推文都带着各项计数和作者，所以脚本不需要设置扩展字段或字段清单。下面是多数脚本会读取的字段：

• id

  含义

  推文的 19 位 ID，以字符串形式发送：请保持为字符串

  示例

  1985432176540981234

• full_text

  含义

  帖子的完整文本

  示例

  Loaded a week of #python posts

• created_at

  含义

  发布时间，UTC 的 ISO 8601 格式

  示例

  2026-11-01T18:42:07Z

• likes_count, retweet_count, reply_count, quote_count

  含义

  互动数

  示例

  42

• view_count, bookmark_count

  含义

  浏览数和收藏数

  示例

  3810

• user

  含义

  作者的资料：username、followers_count 等 19 个字段

  示例

  py_data_notebook, 1840

• quoted_status, retweeted_status

  含义

  被引用或被转发的推文，结构相同

  示例

  一个推文对象

#用一个辅助函数重试 429 和服务器错误

让每个请求都经过同一个函数，只重试值得重试的情况。429 表示你的密钥在一秒内发起了超过 20 次请求，5xx 则是服务器端的临时故障。本指南后面的内容都建立在这个 call 辅助函数之上。

import os
import time

import requests

API = "https://api.sorsa.io/v3"
RETRY = {429, 500, 502, 503, 504}

KEY = os.environ.get("SORSA_API_KEY", "YOUR_API_KEY")

session = requests.Session()
session.headers["ApiKey"] = KEY


def call(method, path, **kwargs):
    """Send one request; retry 429 and 5xx up to 4 times."""
    for attempt in range(5):
        r = session.request(method, API + path, timeout=30, **kwargs)
        if r.status_code not in RETRY or attempt == 4:
            break
        time.sleep(2 ** attempt)  # 1, 2, 4, 8 seconds
    r.raise_for_status()
    return r.json()

401 或 403 永远不值得重试：说明密钥有误或缺失，或者额度已用完，JSON 里的 message 字段会写明是哪种。失败的请求不会计费，所以重试不花钱。Sorsa API 的所有接口都只读，因此重试 /search-tweets 这样的 POST 请求不会重复创建任何内容；而且限制每秒重置一次，所以等一秒就能消除 429。

#用 next_cursor 翻页读取结果

游标是通往下一页结果的不透明键：把 next_cursor 原样传回去，响应里没有它时就停止。搜索和时间线每页最多返回 20 条推文，而且一页可能少于 20 条但后面仍有数据，所以不能靠条数判断是否已到末尾。

def pages(path, body, max_pages=100):
    """Yield the pages of a POST endpoint until the cursor runs out."""
    for _ in range(max_pages):
        page = call("POST", path, json=body)
        yield page
        cursor = page.get("next_cursor")
        if not cursor:
            break
        body = {**body, "next_cursor": cursor}


for page in pages("/search-tweets", {"query": "python lang:en"}):
    for tweet in page["tweets"]:
        print(tweet["id"], tweet["created_at"])

沿着同一个游标读取时，要保持查询和 order 不变，并把 max_pages 当作预算，这样宽泛的查询花的钱永远不会超出你的计划。GET 接口则把游标作为查询参数：GET /followers（在新标签页打开） 每页最多返回 200 份用户资料，从 params 里读取 next_cursor。有两个任务各有完整的脚本：下载某个用户的全部推文可以从保存的游标续传整条时间线，导出 Twitter 粉丝的指南则把粉丝列表写入 CSV。

#每次请求获取 100 条推文或用户资料

如果你已经有一批 ID 或用户名，就分批发送。POST /tweet-info-bulk（在新标签页打开） 最多接受 100 个推文 ID 或链接，GET /info-batch（在新标签页打开） 最多接受 100 个用户名，各算一次请求。

def chunks(items, size=100):
    for i in range(0, len(items), size):
        yield items[i:i + size]


tweets = []
for batch in chunks(tweet_ids):
    tweets += call("POST", "/tweet-info-bulk",
                   json={"tweet_links": batch})["tweets"]

profiles = []
for batch in chunks(usernames):
    profiles += call("GET", "/info-batch",
                     params={"usernames": batch})["users"]

每次请求的价格都与返回内容的多少无关，所以一批 100 条推文的价格和一条推文一样：每 1,000 条推文 $0.014 起，比官方 X API 的 $5.00 便宜 250 多倍。用 requests 时，params 里的列表会以重复参数的形式发出（usernames=a&usernames=b），这正是 GET /info-batch 读取的格式。批量响应不带游标：一个响应就包含整批数据，放在 tweets 或 users 里。

#把推文保存为 CSV 或 pandas DataFrame

pandas.json_normalize 一次调用就能把推文对象列表变成 DataFrame，嵌套的作者会变成 user.username 这样用点连接的列。

import pandas as pd

rows = []
for page in pages("/search-tweets", {"query": "python lang:en"},
                  max_pages=50):
    rows += page["tweets"]

df = pd.json_normalize(rows)
df = df[["id", "created_at", "user.username", "full_text",
         "likes_count", "retweet_count", "view_count"]]
df["created_at"] = pd.to_datetime(df["created_at"])
df.to_csv("tweets.csv", index=False)

import csv

with open("tweets.csv", "w", newline="", encoding="utf-8") as f:
    out = csv.writer(f)
    out.writerow(["id", "created_at", "username", "full_text"])
    for page in pages("/search-tweets", {"query": "python lang:en"}):
        for t in page["tweets"]:
            out.writerow([t["id"], t["created_at"],
                          t["user"]["username"], t["full_text"]])

50 页可得 1,000 条推文，适合在 notebook 里用。pandas 版本会把所有行都放在内存里直到结束，所以任务达到几十万条推文及以上时，请改用 csv 版本：它每页一到就写入，不管运行多久，内存占用都保持平稳。

须知

推文 ID 有 19 位，而 Excel 只保留 15 位，所以双击打开的 CSV 会把每个 ID 的最后几位变成零。读回文件时用 pd.read_csv("tweets.csv", dtype={"id": str})；在 Excel 里则通过“数据”“从文本/CSV”导入，并把 ID 列设为文本。

#用 asyncio 跑到每秒 20 次请求

Sorsa API 在所有套餐上对每个密钥允许每秒 20 次请求，只要工作能拆成相互独立的链，一个 Python 进程用 asyncio 就能达到这个速率。一条游标链无法拆分，因为每一页都需要上一页的游标。所以要让多条链并行运行，每个账号一条，或者搜索时每天一条，并共用一个调度器，让密钥不超过上限。这个版本需要先 pip install httpx。

import asyncio
import os
import time
from datetime import date, timedelta

import httpx

API = "https://api.sorsa.io/v3"
HEADERS = {"ApiKey": os.environ.get("SORSA_API_KEY", "YOUR_API_KEY")}
RETRY = {429, 500, 502, 503, 504}


class Pace:
    """Start at most `rate` requests a second, evenly spaced."""

    def __init__(self, rate=20):
        self.gap, self.next = 1 / rate, 0.0
        self.lock = asyncio.Lock()

    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            start = max(self.next, now)
            self.next = start + self.gap
        await asyncio.sleep(start - now)


async def fetch(client, pace, body):
    for attempt in range(5):
        await pace.wait()
        r = await client.post("/search-tweets", json=body)
        if r.status_code not in RETRY or attempt == 4:
            break
        await asyncio.sleep(2 ** attempt)
    r.raise_for_status()
    return r.json()


async def lane(client, pace, query, max_pages=100):
    """One cursor chain: its pages run one after another."""
    body, tweets = {"query": query}, []
    for _ in range(max_pages):
        page = await fetch(client, pace, body)
        tweets += page["tweets"]
        if not page.get("next_cursor"):
            break
        body = {**body, "next_cursor": page["next_cursor"]}
    return tweets


async def main(queries):
    pace = Pace(19)  # one pacer per key, one start under the limit
    async with httpx.AsyncClient(base_url=API, headers=HEADERS,
                                 timeout=30) as client:
        return await asyncio.gather(
            *(lane(client, pace, q) for q in queries))


day = date(2026, 10, 1)
days = [day + timedelta(n) for n in range(32)]  # Oct 1 to Nov 1
queries = [f"python lang:en since:{a} until:{b}"
           for a, b in zip(days, days[1:])]
results = asyncio.run(main(queries))  # Jupyter: await main(queries)

不管有多少条链在等，Pace 都会均匀地分配起始时间。每秒 19 次给上限留出了一次的余量，所以网络抖动不会让某一秒超过 Sorsa API 的 20 次，漏过去的 429 也会由重试兜住。每个密钥只保留一个调度器：两个进程各带一个调度器，合起来的发起速率会是上限的两倍。按每秒 20 次请求算，每批 100 条的批量请求每秒最多拉取 2,000 条推文，搜索页每秒最多 400 条。

须知

Jupyter notebook 本身已经在运行事件循环，所以 asyncio.run(main(queries)) 在那里会因 RuntimeError 而停止。在 notebook 单元格里请改写成 results = await main(queries)。

下面的规划器会把一个任务换算成请求数、单个密钥所需的时间，以及按 Sorsa API 最低费率计算的成本。

#一个 Python 任务要花多少钱

Sorsa API 按请求收费，与接口返回多少内容无关：搜索结果每 1,000 条推文 $0.07 起，批量请求每 1,000 条 $0.014 起，失败的请求不计费。官方 X API 则按对象收费：每 1,000 条帖子 $5.00，从 2026 年 11 月 22 日起，如果代码扩展出每条帖子的作者（大多数 Tweepy 脚本都是这样），每 1,000 条最高 $15.00。

每 1,000 次请求的费率从 Starter 的 $4.90 到 Hyperscale 的 $1.40 不等，每档套餐对所有接口一视同仁。查看剩余额度不花钱：GET /key-usage-info（在新标签页打开） 永远不会计入你的请求次数。

usage = call("GET", "/key-usage-info")
print(usage["remaining_requests"], "left until", usage["valid_until"])

#把 Tweepy 代码迁移到 Sorsa API

Tweepy 的大多数调用都对应 Sorsa API 的一个接口，而且代码会变短：不需要扩展字段，也不需要字段清单，因为每条推文已经带有作者和各项计数。下表把 Tweepy Client 的每个方法和对应接口配成了对；如果要迁移整个代码库，从 Twitter/X API 迁移的指南列出了每个 v2 接口和响应字段的对应关系。

• search_recent_tweets

  接口

  POST /search-tweets

  有什么变化

  可查 2006 年以来的任意日期，而不只是最近 7 天

• search_all_tweets

  接口

  POST /search-tweets

  有什么变化

  同一个接口，在查询里加上 since: 和 until:

• get_users_tweets

  接口

  POST /user-tweets

  有什么变化

  只需用户名，也没有 3,200 条帖子的上限

• get_tweets

  接口

  POST /tweet-info-bulk

  有什么变化

  每次请求 100 个 ID 或链接

• get_users

  接口

  GET /info-batch

  有什么变化

  每次请求 100 个用户名

• get_users_followers

  接口

  GET /followers

  有什么变化

  每页 200 个粉丝

• get_retweeters

  接口

  POST /retweeters（在新标签页打开）

  有什么变化

  同样的游标循环

• create_tweet

  接口

  无

  有什么变化

  只读：发帖请走官方 X API

字段名也会变：text 变成 full_text，public_metrics["like_count"] 变成 likes_count，author_id 加 includes["users"] 变成 user 对象，next_token 变成 next_cursor。Tweepy 的 wait_on_rate_limit=True 没有对应的设置，因为 X 的 15 分钟窗口已经不在代码里了：call 的重试和调度器会处理这个每秒重置的限制。POST /user-tweets（在新标签页打开） 还能越过最新 3,200 条帖子的界限，而官方时间线到这里就停了（见 X 的时间线文档（在新标签页打开））。

#今天就跑出你的第一次搜索

把 X 数据读进 Python 只需要 requests、一个密钥和四个小辅助函数：负责重试的 call、负责游标的 pages、负责每批 100 条的 chunks，以及限制在每秒 20 次请求的 Pace。建议从 Sorsa API 开始，它是专为大规模读取 X 公开数据而打造的 Twitter API：每次搜索请求 20 条推文，每 1,000 条 $0.07 起，作者信息就在每条推文里，另有 100 次无需绑卡、永不过期的免费请求。同样的调用用 JavaScript 怎么写，见 Node.js 版 Twitter API指南；无代码工作流则可以通过 n8n 版 Twitter API 来使用。

#常见问题