LangChain - Xquik

Build a LangChain agent that can search tweets, hand off IDs and cursors, post tweets, replay stored monitor events, and run extraction jobs - all through Xquik’s MCP server.

Prerequisites

Python 3.10+
Xquik API key (xq_...)
An LLM API key (Anthropic, OpenAI, or any LangChain-supported provider)

Install

pip install langchain-mcp-adapters langchain langchain-anthropic

Full Example

import asyncio
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent


async def main():
    client = MultiServerMCPClient({
        "xquik": {
            "transport": "streamable_http",
            "url": "https://xquik.com/mcp",
            "headers": {
                "x-api-key": "xq_YOUR_KEY_HERE",
            },
        },
    })

    tools = await client.get_tools()

    agent = create_agent(
        "anthropic:claude-sonnet-4-20250514",
        tools,
        system_prompt="You help users interact with X (Twitter) via the Xquik API.",
    )

    prompt = (
        "Search for the latest tweets about AI agents. Return compact JSON "
        "with query, route_used, tweets[{tweet_id,text,author_username,created_at}], "
        "has_more, next_cursor, and key influencers."
    )
    response = await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]})

    Path("xquik-langchain-handoff.json").write_text(
        str(response["messages"][-1].content),
        encoding="utf-8",
    )


asyncio.run(main())

That’s it. The agent auto-discovers all Xquik tools (explore + xquik) and can call any of the 120 API endpoints. LangChain’s MCP adapter loads tools with MultiServerMCPClient. The client is stateless by default, so persist returned IDs, cursors, and write-action status in your job state instead of relying on the next tool call to remember them. The MCP runtime returns normalized snake_case fields through xquik.request(), so keep prompts aligned with tweet_id, has_more, next_cursor, and returned job IDs.

Handoff Checklist

Tweet search rows

Store tweet_id, text, author_username, created_at, has_more, next_cursor, and the original q.

User profile rows

Store source id as user_id, plus username, name, followers, verified, profile_picture, has_more, next_cursor, and the source lookup or search query.

Trend rows

Store each trend name, rank, query, and description. Keep response count, woeid, and the requested region with the run checkpoint.

Monitor and webhook setup

Store the returned monitor id as monitor_id, event_types, next_billing_at, the returned webhook id as webhook_id, url, and the one-time secret in a secret manager. On production deliveries, store delivery_id for receiver retry de-dupe and stream_event_id when one monitor event should process once across endpoint changes.

Stored event replay

Store event_id, type, monitor_id, monitor_type, occurred_at, has_more, next_cursor, and the after query for the next page.

Extraction jobs

Store extraction_id, status, poll, and export_after_complete; poll before loading CSV, JSON, or XLSX rows.

Writes

Store tweet_id or write_action_id, reply_to_tweet_id, status, charged_credits, and poll; do not resend pending writes.

Media attachments

For tweets or replies, pass public URLs in media and store tweet_id or write_action_id. For DMs, upload first, pass one media_id in media_ids, store message_id, and leave reply_to_message_id unset.

Using LangGraph Directly

If you prefer building the graph manually instead of using create_agent:

import asyncio
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition


async def main():
    model = init_chat_model("anthropic:claude-sonnet-4-20250514")

    client = MultiServerMCPClient({
        "xquik": {
            "transport": "streamable_http",
            "url": "https://xquik.com/mcp",
            "headers": {"x-api-key": "xq_YOUR_KEY_HERE"},
        },
    })

    tools = await client.get_tools()

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

    prompt = (
        "Look up @xquikcom's public profile. Return compact JSON with "
        "username, name, user_id, description, followers_count, and route_used."
    )
    result = await graph.ainvoke({"messages": [{"role": "user", "content": prompt}]})
    Path("xquik-langgraph-handoff.json").write_text(
        str(result["messages"][-1].content),
        encoding="utf-8",
    )


asyncio.run(main())

Environment Variables

Store your API key in a .env file instead of hardcoding it:

.env

XQUIK_API_KEY=xq_YOUR_KEY_HERE
ANTHROPIC_API_KEY=sk-ant-...

import os
from dotenv import load_dotenv

load_dotenv()

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

Multiple MCP Servers

Prefix tool names when connecting multiple servers to avoid collisions:

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

Package Versions

Package	Version
`langchain-mcp-adapters`	0.2.2+
`langchain`	1.0.8+
`mcp`	1.9.2+

Documentation Index

​Prerequisites

​Install

​Full Example

​Handoff Checklist