> ## Documentation Index
> Fetch the complete documentation index at: https://docs.xquik.com/llms.txt
> Use this file to discover all available pages before exploring further.
# LangChain
> Connect Xquik's X (Twitter) tools to LangChain agents via MCP
For the complete documentation index, see llms.txt.
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](/quickstart) (`xq_...`)
* An LLM API key (Anthropic, OpenAI, or any LangChain-supported provider)
## Install
```bash theme={null}
pip install langchain-mcp-adapters langchain langchain-anthropic
```
## Full Example
```python theme={null}
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
Store `tweet_id`, `text`, `author_username`, `created_at`, `has_more`, `next_cursor`, and the original `q`.
Store source `id` as `user_id`, plus `username`, `name`, `followers`, `verified`, `profile_picture`, `has_more`, `next_cursor`, and the source lookup or search query.
Store each trend `name`, `rank`, `query`, and `description`. Keep response `count`, `woeid`, and the requested region with the run checkpoint.
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.
Store `event_id`, `type`, `monitor_id`, `monitor_type`, `occurred_at`, `has_more`, `next_cursor`, and the `after` query for the next page.
Store `extraction_id`, `status`, `poll`, and `export_after_complete`; poll before loading CSV, JSON, or XLSX rows.
Store `tweet_id` or `write_action_id`, `reply_to_tweet_id`, `status`, `charged_credits`, and `poll`; do not resend pending writes.
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`:
```python theme={null}
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:
```bash .env theme={null}
XQUIK_API_KEY=xq_YOUR_KEY_HERE
ANTHROPIC_API_KEY=sk-ant-...
```
```python theme={null}
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:
```python theme={null}
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+ |