LangChain

Build LangChain and LangGraph Twitter API agents for tweet search, profiles, followers, and reviewed X actions through TwexAPI MCP.

Build a LangChain Twitter API agent through TwexAPI’s MCP server. Search tweets, inspect profiles, paginate follower lists,. Preserve tweet IDs, timestamps, cursors, and route errors as typed values.

Why use LangChain with TwexAPI?

LangChain connects TwexAPI tools to models, retrievers, databases, and application services. LangGraph adds durable state, resumable jobs, and human approval.

Use LangChain for short tool-calling conversations. Use LangGraph when work must resume after failures, approvals, or process restarts. Both use the same MCP tools and normalized handoff contract.

Prerequisites

• Python 3.10 or later
• A TwexAPI API key
• A LangChain-supported model with tool and structured-output support

Public X reads need no X Developer credentials. Authenticate with TwexAPI.

Install

Install compatible minor ranges for repeatable builds.

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

Store secrets outside source control.

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

Connect TwexAPI MCP

LangChain runs the MCP client. TwexAPI runs the MCP server at https://api.twexapi.io/mcp. The server exposes explore for discovery and twexapi_request for authenticated calls.

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 loads remote MCP tools. Save every cursor, route, and write status externally. The client is stateless by default.

Preserve the MCP response contract

MCP returns endpoint paths, methods, and response fields from explore. Pass only documented query and body fields to twexapi_request.

Paginated routes return cursor fields such as next_cursor, has_next_page, or hasMore. Reuse the same query and filters on every page. Treat each cursor as opaque.

Stop pagination when one condition becomes true:

• The agent collects the requested total.
• has_more or has_next_page becomes false.
• next_cursor is missing or repeats.
• The configured page cap is reached.

Deduplicate tweets and users by stable tweet_id or user_id values.

Keep a resumable agent handoff

Conversation history is not a reliable job database. Persist values needed for retries, pagination, and downstream tools.

See Agent MCP Handoff for the full checklist.

Build error handling

Store status codes with the job. Never retry write actions without explicit approval.

Add human approval to X actions

Read-only agents can search tweets automatically. Write-enabled agents need a review boundary before posting or replying.

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

Use a persistent LangGraph checkpointer in production. Reject any action with an unexpected route, account, target, text, or media.

Build durable LangGraph workflows

Separate discovery, review, execution, and storage. Persist the last completed node, route, response IDs, cursor, and retry count after each external call.

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

Connect multiple MCP servers

Prefix server names when your agent connects to more than one MCP provider.

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

Give the TwexAPI agent only the tools required for its current job.

Package versions

Next steps

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