LangChain

Crea agentes de Twitter API con LangChain y LangGraph para búsqueda de tweets, perfiles, seguidores y acciones en X revisadas a través de TwexAPI MCP.

Crea un agente de Twitter API con LangChain a través del servidor MCP de TwexAPI. Busca tweets, inspecciona perfiles, pagina listas de seguidores y revisa acciones de escritura. Conserva los IDs de tweet, marcas de tiempo, cursores y enruta los errores como valores tipados.

¿Por qué usar LangChain con TwexAPI?

LangChain conecta las herramientas de TwexAPI con modelos, recuperadores, bases de datos y servicios de aplicación. LangGraph añade estado durable, trabajos reanudables y aprobación humana.

Usa LangChain para conversaciones cortas con llamadas a herramientas. Usa LangGraph cuando el trabajo deba reanudarse después de fallos, aprobaciones o reinicios del proceso. Ambos usan las mismas herramientas MCP y el mismo contrato de entrega normalizado.

Requisitos previos

• Python 3.10 o posterior
• Una API key de TwexAPI
• Un modelo compatible con LangChain con soporte para herramientas y salida estructurada
• Una cookie de Twitter o auth_token para acciones de escritura

Las lecturas públicas de X no requieren credenciales de X Developer. Autentícate con TwexAPI.

Instalación

Instala rangos de versiones menores compatibles para builds reproducibles.

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

Guarda los secretos fuera del control de versiones.

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

Conectar TwexAPI MCP

LangChain ejecuta el cliente MCP. TwexAPI ejecuta el servidor MCP en https://api.twexapi.io/mcp. El servidor expone explore para descubrimiento y twexapi_request para llamadas autenticadas.

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 carga herramientas MCP remotas. Guarda cada cursor, ruta y estado de escritura externamente. El cliente es sin estado por defecto.

Conservar el contrato de respuesta MCP

MCP devuelve rutas de endpoint, métodos y campos de respuesta desde explore. Pasa solo los campos documentados de query y body a twexapi_request.

Las rutas paginadas devuelven campos de cursor como next_cursor, has_next_page o hasMore. Reutiliza la misma consulta y filtros en cada página. Trata cada cursor como opaco.

Detén la paginación cuando se cumpla una de estas condiciones:

• El agente recopila el total solicitado.
• has_more o has_next_page pasa a ser false.
• next_cursor falta o se repite.
• Se alcanza el límite de páginas configurado.

Elimina duplicados de tweets y usuarios por valores estables de tweet_id o user_id.

Mantener una entrega de agente reanudable

El historial de conversación no es una base de datos de trabajos confiable. Persiste los valores necesarios para reintentos, paginación y herramientas posteriores.

Consulta Agent MCP Handoff para la lista de verificación completa.

Construir manejo de errores

Almacena los códigos de estado con el trabajo. Nunca reintentes acciones de escritura sin aprobación explícita.

Añadir aprobación humana a acciones en X

Los agentes de solo lectura pueden buscar tweets automáticamente. Los agentes con escritura necesitan un límite de revisión antes de publicar o responder.

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

Usa un checkpointer persistente de LangGraph en producción. Rechaza cualquier acción con una ruta, cuenta, destino, texto o medio inesperados.

Construir flujos de trabajo LangGraph durables

Separa descubrimiento, revisión, ejecución y almacenamiento. Persiste el último nodo completado, ruta, IDs de respuesta, cursor y contador de reintentos después de cada llamada externa.

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

Conectar múltiples servidores MCP

Usa prefijos en los nombres de servidor cuando tu agente se conecte a más de un proveedor 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,
)

Dale al agente de TwexAPI solo las herramientas necesarias para su trabajo actual.

Versiones de paquetes

Próximos pasos

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