--- title: "LangChain" description: "通过 TwexAPI MCP 构建 LangChain 与 LangGraph Twitter API AI Agent，实现推文搜索、资料查询、粉丝列表与经审核的 X 写操作。" --- 通过 TwexAPI 的 MCP 服务器构建 LangChain Twitter API AI Agent。搜索推文、查看资料、分页粉丝列表并审核写操作。保留 tweet ID、时间戳、游标，并将错误路由为类型化值。 ## 为何将 LangChain 与 TwexAPI 搭配使用？ LangChain 将 TwexAPI 工具连接到模型、检索器、数据库与应用服务。LangGraph 增加持久状态、可恢复任务与人机审批。 | AI Agent任务 | TwexAPI 路由 | 为下一步保留 | | --------------------- | --------------------------------------------- | ------------------------------------------------- | | 搜索推文 | `POST /twitter/advanced_search/page` | 查询、tweet ID、作者、`created_at`、游标 | | 查看资料 | `GET /twitter/{screen_name}/about` | 用户 ID、用户名、简介、粉丝数 | | 列出粉丝 | `POST /v3/twitter/users/followers` | 用户名、粉丝行、`next_cursor` | | 读取趋势 | `GET /twitter/global-trending/tweets` | 国家、主题、内容标签、推文行 | | 发帖或回复 | `POST /twitter/tweets/create` | Tweet ID、路由、状态、cookie 确认 | 短对话用 LangChain；失败、审批或进程重启后需恢复的工作用 LangGraph。两者共用相同 MCP 工具与规范化 handoff 契约。 ## 前置条件 * Python 3.10 或更高版本 * [TwexAPI API 密钥](https://twexapi.io/dashboard) * 支持工具调用与结构化输出的 LangChain 兼容模型 * 写操作所需的 Twitter cookie 或 `auth_token` 公开 X 读取无需 X Developer 凭证，使用 TwexAPI 认证即可。 ## 安装 安装兼容的次版本范围以保证可重复构建。 ```bash python -m pip install --upgrade \ "langchain>=1.0" \ "langchain-mcp-adapters>=0.2" \ langchain-anthropic \ langgraph \ python-dotenv ``` 将密钥存放在源码控制之外。 ```bash export TWEXAPI_API_KEY="YOUR_API_KEY" export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY" ``` ## 连接 TwexAPI MCP LangChain 运行 MCP 客户端。TwexAPI 在 `https://api.twexapi.io/mcp` 提供 MCP 服务器，暴露 `explore` 用于发现、`twexapi_request` 用于认证调用。 ```python import asyncio import os from pathlib import Path from typing import Literal from dotenv import load_dotenv from langchain.agents import create_agent from langchain_mcp_adapters.client import MultiServerMCPClient from pydantic import BaseModel class TweetRow(BaseModel): tweet_id: str text: str author_username: str | None = None created_at: str | None = None url: str | None = None class TweetSearchHandoff(BaseModel): query: str route_used: str tweets: list[TweetRow] has_more: bool next_cursor: str | None = None stop_reason: Literal[ "complete", "requested_limit", "cursor_stalled", "page_cap", ] async def main() -> None: load_dotenv() client = MultiServerMCPClient( { "twexapi": { "transport": "streamable_http", "url": "https://api.twexapi.io/mcp", "headers": {"x-api-key": os.environ["TWEXAPI_API_KEY"]}, }, } ) tools = await client.get_tools() agent = create_agent( model="anthropic:claude-sonnet-4-20250514", tools=tools, response_format=TweetSearchHandoff, system_prompt=( "Use TwexAPI for Twitter API requests. Call explore before twexapi_request. " "Preserve exact IDs and cursors. Never invent missing tweet fields. " "Ask for confirmation before read_only: false actions." ), ) result = await agent.ainvoke( { "messages": [ { "role": "user", "content": ( "Search 25 recent tweets about LangChain MCP. " "Return the query, route, tweet rows, cursor state, " "and an explicit stop reason." ), } ] } ) handoff = result["structured_response"] Path("twexapi-langchain-handoff.json").write_text( handoff.model_dump_json(indent=2), encoding="utf-8", ) asyncio.run(main()) ``` `MultiServerMCPClient` 加载远程 MCP 工具。在外部保存每个游标、路由与写状态。客户端默认无状态。 ## 保留 MCP 响应契约 MCP 通过 `explore` 返回端点路径、方法与响应字段。向 `twexapi_request` 仅传入文档化的 `query` 与 `body` 字段。 分页路由返回 `next_cursor`、`has_next_page` 或 `hasMore` 等游标字段。每页复用相同查询与筛选，将游标视为不透明值。 满足以下任一条件时停止分页： * AI Agent已收集请求总量。 * `has_more` 或 `has_next_page` 为 false。 * `next_cursor` 缺失或重复。 * 达到配置的页面上限。 按稳定的 `tweet_id` 或 `user_id` 对推文与用户去重。 ## 保持可恢复的AI Agent handoff 对话历史不是可靠的作业数据库。持久化重试、分页与下游工具所需的值。 存储 `tweet_id`、`text`、`author_username`、`created_at`、`has_more`、`next_cursor` 与原始查询。 存储 `user_id`、`username`、`name`、`description`、粉丝数与查找输入。 存储源用户名、粉丝行、`next_cursor` 与页索引。 存储路由、预览文本、cookie 要求，发帖前需人工审批。 完整清单见 [Agent MCP Handoff](/mcp/agent-handoff)。 ## 构建错误处理 | 状态 | 含义 | AI Agent决策 | | ------ | ----------------------------- | ----------------------------------- | | `400` | 无效路由或参数 | 重试前修正请求 | | `401` | API 密钥缺失或无效 | 停止并更换凭证 | | `403` | 访问被拒或额度不足 | 暂停直至账号访问恢复 | | `429` | 达到速率限制 | 退避后从游标继续 | | `5xx` | 临时服务故障 | 对安全读取应用有界退避 | 将状态码与作业一并存储。未经明确审批切勿重试写操作。 ## 为 X 操作添加人工审批 只读AI Agent可自动搜索推文。启用写的AI Agent在发帖或回复前需审核边界。 ```python from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( model="anthropic:claude-sonnet-4-20250514", tools=tools, middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "twexapi_request": { "allowed_decisions": ["approve", "reject"], } } ) ], checkpointer=InMemorySaver(), ) ``` 生产环境使用持久 LangGraph checkpointer。拒绝路由、账号、目标、文本或媒体与预期不符的操作。 ## 构建 durable LangGraph 工作流 分离发现、审核、执行与存储。每次外部调用后持久化最后完成的节点、路由、响应 ID、游标与重试次数。 ```python from langgraph.graph import START, MessagesState, StateGraph from langgraph.prebuilt import ToolNode, tools_condition def call_model(state: MessagesState): return {"messages": model.bind_tools(tools).invoke(state["messages"])} builder = StateGraph(MessagesState) builder.add_node(call_model) builder.add_node(ToolNode(tools)) builder.add_edge(START, "call_model") builder.add_conditional_edges("call_model", tools_condition) builder.add_edge("tools", "call_model") graph = builder.compile() ``` ## 连接多个 MCP 服务器 AI Agent连接多个 MCP 提供商时为服务器名加前缀。 ```python client = MultiServerMCPClient( { "twexapi": { "transport": "streamable_http", "url": "https://api.twexapi.io/mcp", "headers": {"x-api-key": os.environ["TWEXAPI_API_KEY"]}, }, "docs": { "transport": "streamable_http", "url": "https://docs.twexapi.io/mcp", }, }, tool_name_prefix=True, ) ``` 仅向 TwexAPI AI Agent提供当前任务所需的工具。 ## 包版本 | 包 | 支持范围 | | ------------------------ | --------------- | | Python | `>=3.10` | | `langchain-mcp-adapters` | `>=0.2` | | `langchain` | `>=1.0` | | `langgraph` | `>=0.6` | ## 下一步 * [MCP Tools](/mcp/tools) * [Agent MCP Handoff](/mcp/agent-handoff) * * [Python SDK](/sdks/python) * [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)