LangChain

通过 TwexAPI MCP 构建 LangChain 与 LangGraph Twitter API AI Agent，实现推文搜索、资料查询、粉丝列表与经审核的 X 写操作。

通过 TwexAPI 的 MCP 服务器构建 LangChain Twitter API AI Agent。搜索推文、查看资料、分页粉丝列表并审核写操作。保留 tweet ID、时间戳、游标，并将错误路由为类型化值。

为何将 LangChain 与 TwexAPI 搭配使用？

LangChain 将 TwexAPI 工具连接到模型、检索器、数据库与应用服务。LangGraph 增加持久状态、可恢复任务与人机审批。

短对话用 LangChain；失败、审批或进程重启后需恢复的工作用 LangGraph。两者共用相同 MCP 工具与规范化 handoff 契约。

前置条件

• Python 3.10 或更高版本
• TwexAPI API 密钥
• 支持工具调用与结构化输出的 LangChain 兼容模型
• 写操作所需的 Twitter cookie 或 auth_token

公开 X 读取无需 X Developer 凭证，使用 TwexAPI 认证即可。

安装

安装兼容的次版本范围以保证可重复构建。

python -m pip install --upgrade \
  "langchain>=1.0" \
  "langchain-mcp-adapters>=0.2" \
  langchain-anthropic \
  langgraph \
  python-dotenv

将密钥存放在源码控制之外。

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 用于认证调用。

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

对话历史不是可靠的作业数据库。持久化重试、分页与下游工具所需的值。

完整清单见 Agent MCP Handoff。

构建错误处理

将状态码与作业一并存储。未经明确审批切勿重试写操作。

为 X 操作添加人工审批

只读AI Agent可自动搜索推文。启用写的AI Agent在发帖或回复前需审核边界。

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、游标与重试次数。

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 提供商时为服务器名加前缀。

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提供当前任务所需的工具。

包版本

下一步

• MCP Tools
• Agent MCP Handoff
• Python SDK
• Advanced Twitter Search