Pydantic AI

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 (xq_...)
An LLM API key (Anthropic, OpenAI, or any Pydantic AI-supported provider)

Install

pip install "pydantic-ai-slim[anthropic,mcp]"

Full Example

import asyncio
from pathlib import Path
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

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

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    toolsets=[toolset],
    instructions="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 discovers both Xquik tools and can call 118 MCP operations. Pydantic AI registers MCPToolset as an agent toolsets entry. A URL that does not end in /sse uses Streamable HTTP automatically. The MCP runtime returns normalized snake_case fields through xquik.request(). 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.

Reusing Connections

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

from pathlib import Path

async def main():
    async with agent:
        profile = await agent.run("Look up @username's public profile")
        tweets = await agent.run(
            "Get @username'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:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

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

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    toolsets=[toolset.defer_loading()],
)

Loading from JSON Config

If you manage MCP servers via a config file:

mcp_config.json

{
  "mcpServers": {
    "xquik": {
      "url": "https://xquik.com/mcp",
      "headers": {
        "x-api-key": "xq_YOUR_KEY_HERE"
      }
    }
  }
}

from pydantic_ai import Agent
from pydantic_ai.mcp import load_mcp_toolsets

toolsets = load_mcp_toolsets("mcp_config.json")
agent = Agent("anthropic:claude-sonnet-4-6", toolsets=toolsets)

Tool Prefixes

Avoid naming conflicts when connecting multiple MCP servers:

from pydantic_ai.mcp import MCPToolset

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

Environment Variables

.env

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

import os
from pydantic_ai.mcp import MCPToolset

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

Verified Package Versions

Package	Version
`pydantic-ai-slim`	2.11.0
`fastmcp-slim`	3.4.4

Getting Started

SDKs

Guides

Webhooks

MCP Server

Agent Plugins

Framework Guides

OAuth

MPP

Resources

Prerequisites

Install

Full Example

Handoff Checklist

Tweet search rows

User profile rows

Trend rows

Monitor and webhook setup

Stored event replay

Extraction jobs

Writes

Media attachments

Reusing Connections

Deferred Tool Loading

Loading from JSON Config

Tool Prefixes

Environment Variables

Verified Package Versions

​Prerequisites

​Install

​Full Example

​Handoff Checklist

Tweet search rows

User profile rows

Trend rows

Monitor and webhook setup

Stored event replay

Extraction jobs

Writes

Media attachments

​Reusing Connections

​Deferred Tool Loading

​Loading from JSON Config

​Tool Prefixes

​Environment Variables

​Verified Package Versions

Prerequisites

Install

Full Example

Handoff Checklist

Reusing Connections

Deferred Tool Loading

Loading from JSON Config

Tool Prefixes

Environment Variables

Verified Package Versions