这是一个用于与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
角色澄清:
换句话说,这个库有两个方面:
也可以不使用MCP作为库来使用X API(参见README.md底部)。
您可以从AI助手(如Claude Code、Claude Desktop、codex-cli、Gemini等)操作X API。
通过在所有环境中使用uvx,可以实现自动依赖管理和始终保持最新。
在每个AI工具的MCP配置文件中描述以下内容:
TOML格式(Codex-CLI等):
[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"
[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等):
{
"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"
}
}
}
}
{
"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的问候!'"
--from pyx-mcp自动从PyPI获取最新版本以下工具可通过MCP使用:
您:"发布'来自Claude的MCP问候!'"
Claude:使用create_post工具...
发布完成!帖子ID:1234567890
您:"搜索关于'MCP协议'的近期帖子"
Claude:使用search_recent_posts工具...
找到3篇帖子:
1. @user1:我尝试使用MCP...
2. @user2:模型上下文协议是...
AI助手 ↔ MCP服务器(stdio) ↔ XMCPAdapter ↔ 服务层 ↔ X API
.env和环境变量。reset_at进行退避。timeout和视频质量。echo $X_API_KEY检查环境变量。检查.env是否保存为0o600。upload_video的timeout或使用ffmpeg重新编码。也可以直接从Python代码调用。
uv add pyx-mcp
要使用此库,您需要从您的X(Twitter)开发者账户获取以下四条认证信息。
访问X开发者门户:
选择或创建应用程序:
查看密钥和令牌:
生成并设置权限:
将这些检索到的值设置在环境变量或下面描述的.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客户端调用:
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服务器:
{
"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
--chunk-limit保持在大约150-200字符以维持每句话的块。此外,由于在标点符号后立即拆分可能会破坏上下文,因此在文本文件侧为每个段落插入空白行是安全的。--chunk-limit时应留有一定的余量。如果为每个句子添加换行符,则在拆分后会更容易阅读。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
# 特定