返回市场
派克斯麦普

派克斯麦普

作者:hellocybernetics2 星标更新:2025-10-22

项目介绍

技术文档摘要

X(Twitter)API 客户端

这是一个用于与X(Twitter)API集成的Python客户端。您可以通过MCP从AI助手(如Claude、Gemini等)操作X API。

该库的作用

此库是一个X API客户端。虽然它也充当MCP服务器,但名称x_client意味着它是X(Twitter)服务器的客户端。

graph TD
    A["AI Agent<br>Claude / Gemini / Codex 等"] -->|通过stdio的MCP协议| B["MCP服务器入口点<br>(x_client.integrations.mcp_server)"]

    %% 子图:通过分离ID和显示名来稳定解析
    subgraph x_client_library["x_client库——单个Python进程"]
        B -->|内部调用| C["XMCPAdapter"]
        C -->|内部调用| D["服务层<br>(PostService, MediaService)"]
        D -->|内部调用| E["客户端层<br>(TweepyClient)"]
        D -->|内部调用| F["客户端层<br>(OriginalClient ...未来)"]
    end

    E -->|X API HTTP / REST| G["X服务器<br>Twitter (X)"]
    F -->|X API HTTP / REST| G["X服务器<br>Twitter (X)"]

    style B fill:#e1f5ff
    style C fill:#e1f5ff
    style D fill:#e1f5ff
    style E fill:#e1f5ff
    style F fill:#e1f5ff
    style A fill:#fff4e6
    style G fill:#f3e5f5

角色澄清

  • AI代理(MCP客户端):AI助手如Claude Code、Claude Desktop、Gemini。
  • MCP服务器:由本库提供的兼容MCP协议的服务器。
  • X客户端:本库的核心功能。一个X API客户端。
  • X服务器:Twitter/X的主要服务器。

换句话说,这个库有两个方面:

  1. 从MCP的角度来看:它作为一个MCP服务器,为AI代理提供工具。
  2. 从X API的角度来看:它作为一个X API客户端,与X服务器通信。

也可以不使用MCP作为库来使用X API(参见README.md底部)。

需求

  • Python 3.11或更高版本
  • X(Twitter)开发者账户及一组API密钥
  • 包管理工具uv(推荐)

使用MCP(模型上下文协议)

您可以从AI助手(如Claude Code、Claude Desktop、codex-cli、Gemini等)操作X API。

🚀 推荐设置:统一执行使用uvx

通过在所有环境中使用uvx,可以实现自动依赖管理和始终保持最新。

配置

在每个AI工具的MCP配置文件中描述以下内容:

TOML格式(Codex-CLI等)

  • 发布到PyPI
[mcp.servers.x_client]
command = "uvx"
args = ["--from", "pyx-mcp", "x-mcp-server"]

[mcp.servers.x_client.env]
X_API_KEY = "your-api-key"
X_API_SECRET = "your-api-secret"
X_ACCESS_TOKEN = "your-access-token"
X_ACCESS_TOKEN_SECRET = "your-access-token-secret"
  • 最新来自GitHub
[mcp.servers.x_client]
command = "uvx"
args = ["--from", "git+https://github.com/hellocybernetics/pyX-MCP", "x-mcp-server"]

[mcp.servers.x_client.env]
X_API_KEY = "your-api-key"
X_API_SECRET = "your-api-secret"
X_ACCESS_TOKEN = "your-access-token"
X_ACCESS_TOKEN_SECRET = "your-access-token-secret"

JSON格式(Claude Code、Gemini CLI等)

  • 发布到PyPI
{
  "mcpServers": {
    "x_client": {
      "command": "uvx",
      "args": ["--from", "pyx-mcp", "x-mcp-server"],
      "env": {
        "X_API_KEY": "your-api-key",
        "X_API_SECRET": "your-api-secret",
        "X_ACCESS_TOKEN": "your-access-token",
        "X_ACCESS_TOKEN_SECRET": "your-access-token-secret"
      }
    }
  }
}
  • 最新来自GitHub
{
  "mcpServers": {
    "x_client": {
      "command": "uvx",
      "args": ["--from", "git+https://github.com/hellocybernetics/pyX-MCP", "x-mcp-server"],
      "env": {
        "X_API_KEY": "your-api-key",
        "X_API_SECRET": "your-api-secret",
        "X_ACCESS_TOKEN": "your-access-token",
        "X_ACCESS_TOKEN_SECRET": "your-access-token-secret"
      }
    }
  }
}

重要:设置完成后,请完全重启您的AI工具。

操作检查

向您的AI助手询问如下:

"列出可用的X API工具"

或者

"发布'来自MCP的问候!'"

uvx设置的好处

  • 环境独立:无需Node.js,仅需Python环境
  • 自动依赖管理:uv自动构建并缓存虚拟环境
  • 始终最新--from pyx-mcp自动从PyPI获取最新版本
  • 统一配置:所有AI助手的相同配置方法

提供的功能

以下工具可通过MCP使用:

发帖功能

  • create_post:文本帖子、带有图片/视频的帖子、回复、引用帖子
  • delete_post:删除帖子
  • get_post:通过ID获取帖子
  • create_thread:自动将长文本拆分为线程帖子

转发功能

  • repost_post:转发帖子
  • undo_repost:取消转发

搜索功能

  • search_recent_posts:搜索最近7天的帖子(含作者信息)

媒体上传

  • upload_image:上传图片(JPEG/PNG/WebP/GIF,最大5MB)
  • upload_video:上传视频(MP4,最大512MB,支持分块上传)

认证和状态检查

  • get_auth_status:获取认证状态和速率限制信息

使用示例

您:"发布'来自Claude的MCP问候!'"

Claude:使用create_post工具...
       发布完成!帖子ID:1234567890
您:"搜索关于'MCP协议'的近期帖子"

Claude:使用search_recent_posts工具...
       找到3篇帖子:
       1. @user1:我尝试使用MCP...
       2. @user2:模型上下文协议是...

架构

AI助手 ↔ MCP服务器(stdio) ↔ XMCPAdapter ↔ 服务层 ↔ X API

错误处理

  • ConfigurationError:缺少认证信息。检查.env和环境变量。
  • AuthenticationError:令牌过期。重新运行OAuth流程。
  • RateLimitExceeded:达到速率限制。参考reset_at进行退避。
  • MediaProcessingTimeout/Failed:等待视频处理完成超时。调整timeout和视频质量。

故障排除

  • 缺少凭证:使用echo $X_API_KEY检查环境变量。检查.env是否保存为0o600
  • 无效令牌:重新运行OAuth流程以更新认证信息。
  • 视频超时:延长upload_videotimeout或使用ffmpeg重新编码。

作为库使用

也可以直接从Python代码调用。

安装

uv add pyx-mcp

如何获取认证信息

要使用此库,您需要从您的X(Twitter)开发者账户获取以下四条认证信息。

  1. 访问X开发者门户

  2. 选择或创建应用程序

    • 选择现有应用程序或创建新应用程序。
  3. 查看密钥和令牌

    • 在应用程序仪表板上,转到“密钥和令牌”标签页。
  4. 生成并设置权限

    • API密钥和密钥:在“消费者密钥”部分检查或重新生成。
    • 访问令牌和密钥:在“身份验证令牌”部分,生成具有读写权限的访问令牌和密钥。

将这些检索到的值设置在环境变量或下面描述的.env文件中。

设置认证信息

使用环境变量或.env文件设置认证信息:

export X_API_KEY="your_api_key"
export X_API_SECRET="your_api_secret"
export X_ACCESS_TOKEN="your_access_token"
export X_ACCESS_TOKEN_SECRET="your_access_token_secret"
export X_BEARER_TOKEN="your_bearer_token"  # 用于v2 API(可选)

或在项目根目录中的.env文件中:

X_API_KEY=your_api_key
X_API_SECRET=your_api_secret
X_ACCESS_TOKEN=your_access_token
X_ACCESS_TOKEN_SECRET=your_access_token_secret
X_BEARER_TOKEN=your_bearer_token

.env自动设置为0o600(仅限所有者读写)。.env*.gitignore忽略。


基础用法

from x_client.config import ConfigManager
from x_client.factory import XClientFactory
from x_client.services.post_service import PostService
from x_client.services.media_service import MediaService

# 1. 加载认证信息
config = ConfigManager()
client = XClientFactory.create_from_config(config)

# 2. 初始化服务层
post_service = PostService(client)
media_service = MediaService(client)

# 3. 创建帖子
post = post_service.create_post(text="来自x_client的问候!")
print(f"帖子已创建:{post.id}")

# 4. 带有图片的帖子
from pathlib import Path
media_result = media_service.upload_image(Path("image.png"))
post = post_service.create_post(
    text="看看这张图片!",
    media_ids=[media_result.media_id]
)

# 5. 创建长线程
thread = post_service.create_thread(
    '''Python 3.11亮点...(长文本)''',
    chunk_limit=200,
)
for idx, segment_post in enumerate(thread.posts, start=1):
    print(f"段落 {idx}:{segment_post.id}")
if not thread.succeeded:
    print("线程失败", thread.error)

# 6. 转发操作
repost_state = post_service.repost_post(post.id)
print("已转发:", repost_state.reposted)

undo_state = post_service.undo_repost(post.id)
print("取消转发:", not undo_state.reposted)

# 7. 带有作者信息的搜索
search_results = post_service.search_recent(
    "from:twitterdev",
    expansions=["author_id"],
    user_fields=["username", "verified"],
    post_fields=["created_at"],
)
for item in search_results:
    author = item.author.username if item.author else "未知"
    print(author, item.text)

通过MCP适配器使用(上述API的简化版)

也可以直接从非MCP客户端调用:

from x_client.integrations.mcp_adapter import XMCPAdapter

adapter = XMCPAdapter()  # 认证信息由ConfigManager自动加载

post = adapter.create_post({"text": "来自MCP的问候!"})
print(post)

media = adapter.upload_image({"path": "/path/to/image.png"})
adapter.create_post({"text": "图片帖子", "media_ids": [media["media_id"]]})

日志记录和可观测性

PostService内置了结构化日志记录和事件钩子:

import logging
from x_client.config import ConfigManager
from x_client.factory import XClientFactory
from x_client.services.post_service import PostService

logging.basicConfig(level=logging.INFO)

client = XClientFactory.create_from_config(ConfigManager())

def metrics_hook(event: str, payload: dict[str, object]) -> None:
    # 与Prometheus / OpenTelemetry等的集成点
    print("指标", event, payload)

post_service = PostService(client, event_hook=metrics_hook)
post_service.create_post("可观测性就绪!")

事件钩子将成功和失败统一到一个回调中,便于发送指标并与分布式跟踪集成。


开发环境中的使用

设置

uv sync

这将在.venv/bin/中创建x-mcp-server命令。

使用本地路径运行MCP服务器

要在开发中直接运行MCP服务器:

{
  "mcpServers": {
    "x-client": {
      "command": "/绝对路径/to/twitter/.venv/bin/x-mcp-server",
      "env": {
        "X_API_KEY": "your-api-key",
        "X_API_SECRET": "your-api-secret",
        "X_ACCESS_TOKEN": "your-access-token",
        "X_ACCESS_TOKEN_SECRET": "your-access-token-secret"
      }
    }
  }
}
<details> <summary>其他方法(点击展开)</summary>

方法2:直接使用uv

{
  "mcpServers": {
    "x-client": {
      "command": "uv",
      "args": ["run", "--directory", "/绝对路径/to/twitter", "python", "-m", "x_client.integrations.mcp_server"],
      "env": {
        "X_API_KEY": "your-api-key",
        "X_API_SECRET": "your-api-secret",
        "X_ACCESS_TOKEN": "your-access-token",
        "X_ACCESS_TOKEN_SECRET": "your-access-token-secret"
      }
    }
  }
}

方法3:启动脚本

{
  "mcpServers": {
    "x-client": {
      "command": "/绝对路径/to/twitter/scripts/run_mcp_server.sh",
      "env": { "X_API_KEY": "...", "X_API_SECRET": "...", "X_ACCESS_TOKEN": "...", "X_ACCESS_TOKEN_SECRET": "..." }
    }
  }
}
</details>

重要:将/绝对路径/to/twitter替换为实际项目路径。


命令行使用

您可以轻松地从命令行使用examples/create_post.py发布。

基础用法

# 只有文本
python examples/create_post.py "来自x_client的问候!"

# 带有图片
python examples/create_post.py "看看这张图片!" --image 路径/to/image.png

# 带有视频(最大512MB,支持分块上传)
python examples/create_post.py "看看这段视频!" --video 路径/to/video.mp4

# 使用不同路径的.env
python examples/create_post.py "自定义env的问候" --dotenv /安全路径/.env

线程发布

# 长线程帖子(自动拆分,chunk_limit=180)
python examples/create_post.py "长形式更新..." --thread --chunk-limit 180

# 从文件发布线程(假设UTF-8文本)
python examples/create_post.py --thread-file docs/thread_draft.txt

# 示例长日本线程(适当断行不超过280字符)
python examples/create_post.py --thread-file examples/long_thread_ja.txt --chunk-limit 180

# 示例长英文线程(保持句子断开)
python examples/create_post.py --thread-file examples/long_thread_en.txt --chunk-limit 240

# 每次发布之间等待8秒以避免速率限制
python examples/create_post.py --thread-file examples/long_thread_en.txt --segment-pause 8

# 选择拆分策略(简单 | 句子 | 段落)
python examples/create_post.py --thread-file examples/long_thread_en.txt \
  --chunk-limit 240 --split-strategy 句子

其他操作

# 删除失败线程的第一个推文(用于解决重复错误)
python examples/create_post.py --delete 1234567890123456789

# 转发 / 取消转发
python examples/create_post.py --repost 1234567890
python examples/create_post.py --undo-repost 1234567890

特定语言注意事项

  • 日语:如果有很多全角字符,填满280字符限制可能会难以阅读,因此将--chunk-limit保持在大约150-200字符以维持每句话的块。此外,由于在标点符号后立即拆分可能会破坏上下文,因此在文本文件侧为每个段落插入空白行是安全的。
  • 英语:当包含URL或表情符号时,Twitter将其计为23个字符,因此设置--chunk-limit时应留有一定的余量。如果为每个句子添加换行符,则在拆分后会更容易阅读。

注意事项

  • 当重新发布线程时,如果在24小时内发布相同的正文文本,由于X的规定,您将收到重复内容错误。请删除之前发布的线程或在文本中添加唯一短语,例如时间戳。
  • 如果在短时间内连续发布,X API可能会返回HTTP 429(请求过多)。此库检测RateLimitExceeded并在响应头中的x-rate-limit-reset指定的时间后重试,但如果发生429,请等待2-3分钟后再重新执行命令。将--segment-pause设置为大约5-10秒有助于提前避免429。

测试

# 测试MCP服务器操作
uv run python scripts/test_mcp_server.py

# 单元测试
uv run pytest tests/unit/test_mcp_adapter.py -v

# 运行所有测试
uv run pytest

# 带覆盖率运行
uv run pytest --cov=x_client --cov-report=html

# 详细模式
uv run pytest -v

# 特定