LangChain Twitter API agent with MCP & LangGraph

python -m pip install --upgrade \
  "langchain>=1.3,<1.4" \
  "langchain-mcp-adapters>=0.3,<0.4" \
  langchain-anthropic \
  python-dotenv

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: int | 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(
        {
            "xquik": {
                "transport": "http",
                "url": "https://xquik.com/mcp",
                "headers": {"x-api-key": os.environ["XQUIK_API_KEY"]},
            },
        }
    )
    tools = await client.get_tools()

    agent = create_agent(
        model="anthropic:claude-sonnet-4-6",
        tools=tools,
        response_format=TweetSearchHandoff,
        system_prompt=(
            "Use Xquik for Twitter API requests. Preserve exact IDs and cursors. "
            "Use GET /api/v1/x/tweets/search. Never invent missing tweet fields."
        ),
    )

    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("xquik-langchain-handoff.json").write_text(
        handoff.model_dump_json(indent=2),
        encoding="utf-8",
    )


asyncio.run(main())

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-6",
    tools=tools,
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "xquik": {
                    "allowed_decisions": ["approve", "reject"],
                }
            }
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {
    "configurable": {
        "thread_id": "xquik-twitter-agent-job-001",
    }
}

pending = await agent.ainvoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Draft a reply, but never post without my approval.",
            }
        ]
    },
    config=config,
)

client = MultiServerMCPClient(
    {
        "xquik": {
            "transport": "http",
            "url": "https://xquik.com/mcp",
            "headers": {"x-api-key": os.environ["XQUIK_API_KEY"]},
        },
        "knowledge": {
            "transport": "http",
            "url": "https://knowledge.example.com/mcp",
        },
    },
    tool_name_prefix=True,
)