> ## 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.

# Pydantic AI

> Connect Xquik's X (Twitter) tools to Pydantic AI agents via MCP

<blockquote className="agent-llms-directive">
  For the complete documentation index, see <a href="/llms.txt">llms.txt</a>.
</blockquote>

Build a Pydantic AI agent that can search tweets, hand off IDs and cursors, post tweets, replay stored monitor events, and run extraction jobs - connected to Xquik's MCP server with 3 lines of config.

## Prerequisites

* Python 3.10+
* [Xquik API key](/quickstart) (`xq_...`)
* An LLM API key (Anthropic, OpenAI, or any Pydantic AI-supported provider)

## Install

```bash theme={null}
pip install "pydantic-ai[mcp]"
```

## Full Example

```python theme={null}
import asyncio
from pathlib import Path
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP(
    "https://xquik.com/mcp",
    headers={"x-api-key": "xq_YOUR_KEY_HERE"},
)

agent = Agent(
    "anthropic:claude-sonnet-4-20250514",
    toolsets=[server],
    system_prompt="You help users interact with X (Twitter) via the Xquik API.",
)


async def main():
    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."
    )
    result = await agent.run(prompt)
    Path("xquik-pydantic-ai-handoff.json").write_text(
        result.output,
        encoding="utf-8",
    )


asyncio.run(main())
```

The agent auto-discovers all Xquik tools and can call any of the 120 API endpoints.

Pydantic AI registers `MCPServerStreamableHTTP` as an agent `toolsets` entry. 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

<CardGroup cols={2}>
  <Card title="Tweet search rows" icon="search">
    Store `tweet_id`, `text`, `author_username`, `created_at`, `has_more`, `next_cursor`, and the original `q`.
  </Card>

  <Card title="User profile rows" icon="users">
    Store source `id` as `user_id`, plus `username`, `name`, `followers`, `verified`, `profile_picture`, `has_more`, `next_cursor`, and the source lookup or search query.
  </Card>

  <Card title="Trend rows" icon="trending-up">
    Store each trend `name`, `rank`, `query`, and `description`. Keep response `count`, `woeid`, and the requested region with the run checkpoint.
  </Card>

  <Card title="Monitor and webhook setup" icon="radio">
    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.
  </Card>

  <Card title="Stored event replay" icon="activity">
    Store `event_id`, `type`, `monitor_id`, `monitor_type`, `occurred_at`, `has_more`, `next_cursor`, and the `after` query for the next page.
  </Card>

  <Card title="Extraction jobs" icon="database">
    Store `extraction_id`, `status`, `poll`, and `export_after_complete`; poll before loading CSV, JSON, or XLSX rows.
  </Card>

  <Card title="Writes" icon="send">
    Store `tweet_id` or `write_action_id`, `reply_to_tweet_id`, `status`, `charged_credits`, and `poll`; do not resend pending writes.
  </Card>

  <Card title="Media attachments" icon="image">
    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.
  </Card>
</CardGroup>

## Reusing Connections

Wrap multiple calls in `async with agent` to keep the MCP connection open across requests:

```python theme={null}
from pathlib import Path

async def main():
    async with agent:
        profile = await agent.run("Look up @xquikcom's public profile")
        tweets = await agent.run(
            "Get @xquikcom's latest 5 tweets. Return compact JSON with "
            "tweet_id, author_username, text, created_at, has_more, next_cursor, "
            "and route_used."
        )

    Path("xquik-pydantic-ai-profile.json").write_text(
        profile.output,
        encoding="utf-8",
    )
    Path("xquik-pydantic-ai-tweets.json").write_text(
        tweets.output,
        encoding="utf-8",
    )
```

## Deferred Tool Loading

For large tool sets, use `DeferredLoadingToolset` to hide tools from the model until they're discovered via tool search. This reduces prompt token usage:

```python theme={null}
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP(
    "https://xquik.com/mcp",
    headers={"x-api-key": "xq_YOUR_KEY_HERE"},
)

agent = Agent(
    "anthropic:claude-sonnet-4-20250514",
    toolsets=[server.defer_loading()],
)
```

## Loading from JSON Config

If you manage MCP servers via a config file:

```json mcp_config.json theme={null}
{
  "mcpServers": {
    "xquik": {
      "transport": "streamable-http",
      "url": "https://xquik.com/mcp",
      "headers": {
        "x-api-key": "xq_YOUR_KEY_HERE"
      }
    }
  }
}
```

```python theme={null}
from pydantic_ai import Agent
from pydantic_ai.mcp import load_mcp_servers

servers = load_mcp_servers("mcp_config.json")
agent = Agent("anthropic:claude-sonnet-4-20250514", toolsets=servers)
```

## Tool Prefixes

Avoid naming conflicts when connecting multiple MCP servers:

```python theme={null}
from pydantic_ai.mcp import MCPServerStreamableHTTP

xquik = MCPServerStreamableHTTP(
    "https://xquik.com/mcp",
    headers={"x-api-key": "xq_YOUR_KEY_HERE"},
    tool_prefix="xquik",
)
# Tools become: xquik_explore, xquik_xquik, etc.
```

## Environment Variables

```bash .env theme={null}
XQUIK_API_KEY=xq_YOUR_KEY_HERE
ANTHROPIC_API_KEY=sk-ant-...
```

```python theme={null}
import os
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP(
    "https://xquik.com/mcp",
    headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
)
```

## Package Versions

| Package       | Version |
| ------------- | ------- |
| `pydantic-ai` | 1.78.0+ |
| `mcp`         | 1.25.0+ |