LangChain

Construisez des agents API Twitter LangChain et LangGraph pour la recherche de tweets, les profils, les abonnés et les actions X revues via TwexAPI MCP.

Construisez un agent API Twitter LangChain via le serveur MCP de TwexAPI. Recherchez des tweets, inspectez des profils, paginez les listes d’abonnés et révisez les actions d’écriture. Conservez les ID de tweets, horodatages, curseurs et erreurs de route en valeurs typées.

Pourquoi utiliser LangChain avec TwexAPI ?

LangChain connecte les outils TwexAPI aux modèles, retrieveurs, bases de données et services applicatifs. LangGraph ajoute un état durable, des tâches reprenables et l’approbation humaine.

Utilisez LangChain pour les conversations courtes avec appels d’outils. Utilisez LangGraph lorsque le travail doit reprendre après des échecs, des approbations ou des redémarrages de processus. Les deux utilisent les mêmes outils MCP et le même contrat de handoff normalisé.

Prérequis

• Python 3.10 ou ultérieur
• Une clé API TwexAPI
• Un modèle pris en charge par LangChain avec appels d’outils et sortie structurée
• Un cookie Twitter ou auth_token pour les actions d’écriture

Les lectures publiques X ne nécessitent pas d’identifiants X Developer. Authentifiez-vous avec TwexAPI.

Installation

Installez des plages de versions mineures compatibles pour des builds reproductibles.

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

Stockez les secrets hors du contrôle de version.

export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"

Connecter TwexAPI MCP

LangChain exécute le client MCP. TwexAPI exécute le serveur MCP sur https://api.twexapi.io/mcp. Le serveur expose explore pour la découverte et twexapi_request pour les appels authentifiés.

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 charge les outils MCP distants. Sauvegardez chaque curseur, route et statut d’écriture en externe. Le client est sans état par défaut.

Préserver le contrat de réponse MCP

MCP renvoie les chemins d’endpoints, méthodes et champs de réponse depuis explore. Passez uniquement les champs query et body documentés à twexapi_request.

Les routes paginées renvoient des champs de curseur tels que next_cursor, has_next_page ou hasMore. Réutilisez la même requête et les mêmes filtres sur chaque page. Traitez chaque curseur comme opaque.

Arrêtez la pagination lorsqu’une condition devient vraie :

• L’agent collecte le total demandé.
• has_more ou has_next_page devient false.
• next_cursor est absent ou se répète.
• Le plafond de pages configuré est atteint.

Dédupliquez tweets et utilisateurs par tweet_id ou user_id stables.

Conserver un handoff agent reprenable

L’historique de conversation n’est pas une base de données de tâches fiable. Persistez les valeurs nécessaires aux retries, à la pagination et aux outils en aval.

Voir Agent MCP Handoff pour la checklist complète.

Gestion des erreurs

Stockez les codes de statut avec la tâche. Ne réessayez jamais les actions d’écriture sans approbation explicite.

Ajouter l’approbation humaine aux actions X

Les agents en lecture seule peuvent rechercher des tweets automatiquement. Les agents avec écriture ont besoin d’une frontière de revue avant publication ou réponse.

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(),
)

Utilisez un checkpointer LangGraph persistant en production. Rejetez toute action avec une route, un compte, une cible, un texte ou un média inattendu.

Construire des workflows LangGraph durables

Séparez découverte, revue, exécution et stockage. Persistez après chaque appel externe le dernier nœud complété, la route, les ID de réponse, le curseur et le compteur de retries.

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()

Connecter plusieurs serveurs MCP

Préfixez les noms de serveur lorsque votre agent se connecte à plus d’un fournisseur 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,
)

Donnez à l’agent TwexAPI uniquement les outils requis pour sa tâche actuelle.

Versions des packages

Étapes suivantes

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