> ## 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 MCP Twitter API agent guide for Python

> Build a Pydantic AI Twitter API agent for typed tweet search, profiles, followers, monitors, exports, and reviewed X actions through MCP. See Python code.

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

Build a Pydantic AI Twitter API agent through Xquik's MCP server. Search
tweets, inspect profiles, export followers, and replay monitor events. Review
every proposed post, reply, like, repost, follow, and direct message. Validate
every durable handoff with a Pydantic model.

## Why use Pydantic AI with a Twitter API?

Pydantic AI combines model tool calls with typed Python output. Xquik supplies
the Twitter API routes through 3 MCP tools: `docs`, `search`, and `execute`.

Each Pydantic AI control covers one boundary.

| Boundary | Pydantic AI control | Twitter agent benefit |
| - | - | - |
| MCP connection | `MCP` capability | Keep credentials and tracing in your process |
| Final response | Pydantic `output_type` | Reject malformed tweet rows and cursors |
| Tool catalog | `.defer_loading()` | Hide unused MCP tools until discovery |
| Write actions | `.approval_required()` | Pause before posting, replying, or following |
| MCP naming | `.prefixed()` | Prevent tool collisions across servers |
| Connection lifecycle | `async with agent` | Reuse one MCP session across related calls |

Use one typed agent for tweet search, profile enrichment, or follower exports.
Use another agent for monitor processing. Keep each agent focused.

## Pydantic AI Twitter API prerequisites

* Python 3.10 or later
* An [Xquik API key](/x-api-quickstart) beginning with `xq_`
* A Pydantic AI-supported model with tool calling
* A connected X account for private reads or write actions

Public X reads do not require X Developer credentials. Authenticate with
Xquik. Connect an X account only when the selected route requires it.

This supplies a Twitter API for Python agents through one MCP connection.
Send the `x-api-key` header. Do not send bearer tokens or access tokens.

## Install Pydantic AI MCP support

Install the current stable Pydantic AI 2.x line. Keep FastMCP below 4 until
Pydantic AI adds MCP SDK v2 support.

```bash theme={null}
python -m pip install --upgrade \
  "pydantic-ai-slim[anthropic,mcp]>=2.22,<2.23" \
  "fastmcp-slim>=3.3,<4" \
  python-dotenv
```

The explicit FastMCP range prevents an MCP SDK v2 prerelease from entering the
environment. Remove that cap only after Pydantic AI adds FastMCP 4 support.

This Pydantic AI MCP Streamable HTTP setup keeps execution inside Python.
This Pydantic AI MCP server connection uses the Xquik endpoint.
Keep the MCP URL HTTPS-only.
It authenticates with the `x-api-key` header.
After setup, run it inside a local Python process.

Store secrets outside source control.

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

```text .gitignore theme={null}
.env
xquik-*-handoff.json
```

## Build a Pydantic AI MCP example for tweet search

Define the final tweet-search handoff before creating the agent. Pydantic AI
validates the model output against this schema.

To import `Agent` from Pydantic AI, use `from pydantic_ai import Agent`.
Use type hints for every durable handoff field.
Start the async example with `import asyncio`.
The `typing` import supplies `Literal` for finite stop reasons.

```python theme={null}
import asyncio
import os
from pathlib import Path
from typing import Literal

from dotenv import load_dotenv
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP


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"]


async def main() -> None:
    load_dotenv()

    xquik_mcp = MCP(
        "https://xquik.com/mcp",
        headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
        allowed_tools=["docs", "search", "execute"],
    )

    agent = Agent(
        "anthropic:claude-sonnet-4-6",
        capabilities=[xquik_mcp],
        output_type=TweetSearchHandoff,
        instructions=(
            "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.run(
        "Search 25 recent tweets about Pydantic AI MCP. "
        "Return the query, route, tweet rows, cursor state, "
        "and an explicit stop reason."
    )

    Path("xquik-pydantic-ai-handoff.json").write_text(
        result.output.model_dump_json(indent=2),
        encoding="utf-8",
    )


asyncio.run(main())
```

`MCP` is Pydantic AI's primary MCP entry point. The connection runs locally
unless you configure another execution mode. Headers, hooks, and traces remain
inside your Python process. The URL selects Streamable HTTP. The allowed list
contains only the public `docs`, `search`, and `execute` tools.

Model Context Protocol (MCP) turns Xquik routes into model-callable tools.
Pydantic builds JSON schemas from `output_type`. Use explicit keyword arguments
for `headers`, `allowed_tools`, `capabilities`, and `output_type`.

Pydantic AI opens and closes the connection for you. Enter
`async with agent` when several runs should share one connection.

The agent discovers `docs`, `search`, and `execute`. Discover route requirements before
selecting an unfamiliar endpoint. The Xquik sandbox invokes
`xquik.request(...)` with validated route values. Xquik injects authentication
separately.

## Validate Twitter API fields before storage

MCP returns normalized snake\_case fields. Date-time values use Unix seconds.
Map REST `createdAt` to `created`, never `created_at`.

Keep these source fields unchanged:

* Tweet rows: `id`, `text`, `author`, `created`, and `url`
* Profile rows: `id`, `username`, `name`, `description`, and `followers`
* Page state: `has_more` and `next_cursor`
* Error fields: `error.type`, `error.code`, and `error.message`
* Write receipts: `tweet_id`, `write_action_id`, and `charged_credits`

Map source `id` to `tweet_id` or `user_id` only in your output model. Never
cast large IDs to floating-point numbers.

Search and list routes return `has_more` and `next_cursor`. Pass `cursor` for
tweet, profile, follower, reply, timeline, community, and list pagination.
Keep the original query and filters unchanged.

Events, draws, and extraction pages use `cursor`. Radar pages use `after`.
Draft pages use `afterCursor`. Treat every cursor as opaque.

Continue through an empty page when `has_more` stays true. Stop when no cursor
exists. Stop after the server repeats a cursor. Return `cursor_stalled` with
the number of collected rows.

MCP output has a 24,000-character limit. Project only required fields. Use
extraction exports when the workflow must store every complete row.

## Build a typed Twitter agent handoff

Conversation text cannot safely resume a Python Twitter API job. Store the
validated fields required by the next worker.

<CardGroup cols={2}>
  <Card title="Tweet search" icon="search">
    Store the query, route, tweet IDs, authors, `created`, URLs, `has_more`,
    `next_cursor`, and stop reason.
  </Card>

  <Card title="Profile lookup" icon="user">
    Store `user_id`, `username`, `name`, `description`, `followers`,
    `verified`, and `profile_picture`.
  </Card>

  <Card title="Follower export" icon="download">
    Store the source user, extraction ID, status, poll URL, requested format,
    and export completion state.
  </Card>

  <Card title="Reply collection" icon="message-circle">
    Store the root tweet ID, reply IDs, parent IDs, cursor state, and coverage
    diagnostics. Keep nested replies separate.
  </Card>

  <Card title="Monitor replay" icon="radio">
    Store `monitor_id`, `event_id`, `type`, `occurred_at`, `has_more`, and
    `next_cursor`. Send the next cursor as `cursor`.
  </Card>

  <Card title="Webhook delivery" icon="webhook">
    Store `webhook_id`, `delivery_id`, and `stream_event_id`. Keep the webhook
    secret in a secret manager.
  </Card>

  <Card title="Write receipt" icon="send">
    Store `tweet_id` or `write_action_id`, `status`, `charged_credits`, `poll`,
    and the idempotency key.
  </Card>

  <Card title="Media attachment" icon="image">
    Use public URLs in `media` for tweets. Reserve uploaded `media_id` values
    for direct messages.
  </Card>
</CardGroup>

Keep API keys, headers, webhook secrets, and raw signatures outside agent
output. Never place private messages in shared traces.

## Reuse the MCP connection safely

Wrap related calls in `async with agent`. One session fetches 2 tweet-search
pages.

```python theme={null}
async def collect_two_search_pages() -> None:
    async with agent:
        first_page = await agent.run(
            "Search 25 tweets about Pydantic AI MCP. Preserve the next cursor."
        )
        cursor = first_page.output.next_cursor
        if not first_page.output.has_more or cursor is None:
            Path("xquik-pydantic-ai-page-1.json").write_text(
                first_page.output.model_dump_json(indent=2),
                encoding="utf-8",
            )
            return

        second_page = await agent.run(
            f"Continue tweet search for {first_page.output.query!r}. "
            f"Use this exact cursor: {cursor!r}. Preserve IDs and cursor state."
        )

    Path("xquik-pydantic-ai-page-2.json").write_text(
        second_page.output.model_dump_json(indent=2),
        encoding="utf-8",
    )
```

One local MCP session connects as one identity. Do not share one session across
users with different Xquik keys. Give every credential its own request-scoped
MCP capability.

## Require approval before X actions

Read-only agents can search tweets automatically. Write-capable agents need a
human decision before every X action. Review posts, replies, likes, reposts,
follows, and direct messages.

Xquik exposes all API calls through one aggregate `execute` tool. Review and
approve each validated Xquik tool invocation. Leave `search` available without
approval.

```python theme={null}
from pydantic import BaseModel
from pydantic_ai import Agent, DeferredToolRequests, DeferredToolResults
from pydantic_ai.mcp import MCPToolset


class ActionReceipt(BaseModel):
    route_used: str
    tweet_id: str | None = None
    write_action_id: str | None = None
    status: str
    charged_credits: int | None = None


xquik = MCPToolset(
    "https://xquik.com/mcp",
    headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
)
reviewed_xquik = xquik.approval_required(
    lambda _ctx, tool_def, _args: tool_def.name == "execute"
)

write_agent = Agent(
    "anthropic:claude-sonnet-4-6",
    toolsets=[reviewed_xquik],
    output_type=[ActionReceipt, DeferredToolRequests],
    instructions="Never execute an X action without explicit approval.",
)

pending = await write_agent.run(
    "Draft a reply to tweet 123. Do not post before approval."
)

if isinstance(pending.output, DeferredToolRequests):
    approvals: dict[str, bool] = {}
    for call in pending.output.approvals:
        decision = input(f"Approve {call.tool_name} with {call.args}? [y/N] ")
        approvals[call.tool_call_id] = decision.strip().lower() == "y"

    completed = await write_agent.run(
        message_history=pending.all_messages(),
        deferred_tool_results=DeferredToolResults(approvals=approvals),
    )
```

Inspect the sandbox code before approval. Confirm the route, method, account,
target ID, text, media, and idempotency key. Reject unexpected arguments.
The CLI displays every validated tool request. Replace it with your
application's review screen when needed.

Do not use `approve_all=True` when the agent can write to X.
Store the approval decision and tool call ID with the resulting action receipt.

## Build Twitter API error handling

Treat error messages as typed error handling inputs.

| Status | Meaning | Pydantic AI decision |
| - | - | - |
| `400` | Invalid parameters or unsupported fields | Correct the request before retrying |
| `401` | Invalid Xquik authentication | Stop and replace the credential |
| `402` | Subscription or credit action required | Report choices and request confirmation |
| `424` | Dependency failure or incomplete replies | Keep partial rows and inspect retryability |
| `429` | Twitter API rate limit reached | Respect `error.retry_after` and back off |
| `502` | Temporary upstream failure | Retry safe reads with a bounded policy |

A `402` never authorizes a purchase. Show the available payment choices. Wait
for explicit confirmation before any supported account checkout action.

Never recreate a pending write after a timeout. Poll the returned action ID.
Retry only safe reads when `error.retryable` permits it.

## Defer and prefix MCP tools

Xquik exposes only 3 tools. Deferred loading is optional for one
Xquik connection. Use it when several servers create a larger tool catalog.

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

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

Prefix tools when another server also exposes `docs`, `search`, or `execute`.

```python theme={null}
twitter_tools = MCPToolset(
    "https://xquik.com/mcp",
    headers={"x-api-key": os.environ["XQUIK_API_KEY"]},
).prefixed("twitter")

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    toolsets=[twitter_tools],
)
```

The resulting names become `twitter_docs`, `twitter_search`, and `twitter_execute`. Update any
approval predicate after adding a prefix.

## Choose Pydantic AI MCP or the REST API

MCP and REST solve different integration problems. MCP adds a discoverable tool
layer over documented Twitter API routes.

| Requirement | Choose | Reason |
| - | - | - |
| A model selects tweet or profile operations | Pydantic AI MCP | The agent discovers `docs`, `search`, and `execute` |
| Application code calls one known route | REST API | The request stays deterministic |
| A human must review an X action | Pydantic AI MCP | Deferred tool approval pauses execution |
| A scheduled follower export runs without a model | REST API | The job needs no model decision |
| A typed agent hands work to Python | Pydantic AI MCP | `output_type` validates the handoff |

Xquik validates the sandbox route, method, query, and body. Pydantic validates
the final typed handoff. Keep both layers. Each layer enforces a separate
boundary.

## Tested Pydantic AI compatibility

Xquik checked these versions on August 3, 2026.

| Package | Checked version | Supported range in this guide |
| - | - | - |
| Python | 3.10 or later | `>=3.10` |
| `pydantic-ai-slim` | 2.22.0 | `>=2.22,<2.23` |
| `fastmcp-slim` | 3.4.5 | `>=3.3,<4` through the `mcp` extra |

[Pydantic AI issue 6661](https://github.com/pydantic/pydantic-ai/issues/6661)
tracks FastMCP 4 and MCP SDK v2 support. Use the stable FastMCP 3.x line until
that compatibility work ships.

## Pydantic AI Twitter API questions

### Does Pydantic AI support MCP?

Yes. Install the `mcp` extra and create an `MCP` capability. Xquik uses the
recommended Streamable HTTP transport.

### What is the difference between Pydantic AI tools and MCP?

Your Python application registers local Pydantic AI tools.
MCP tools come from a connected server and can change independently. Xquik
publishes `docs`, `search`, and `execute` through MCP.

### Is Pydantic AI better than LangChain for a Twitter agent?

Choose Pydantic AI for typed Python handoffs. Choose
[LangChain and LangGraph](/guides/langchain) for durable graph orchestration.
Both can call the same Xquik MCP tools.

### Can Pydantic AI call the Twitter API?

Yes. MCP exposes eligible routes for tweets, profiles, followers, monitors,
extractions, and X actions. Authenticate with an Xquik API key.

### Can Pydantic AI scrape tweets with Python?

Yes. Call `GET /api/v1/x/tweets/search` through the MCP tools. Keep tweet
IDs, authors, `created`, URLs, `has_more`, and `next_cursor`.

### How do I validate Twitter API JSON with Pydantic?

Pass a `BaseModel` as the agent's `output_type`. Call `model_dump_json()` before
storing each validated output. Never print `result.output` into logs or files.
Validate `result.output` before storage. Never save conversational text as JSON.

### How can I use Pydantic with AI model validation effectively?

Model tweet IDs and cursors as strings. Mark a field optional only when its
route omits that field. Use `Literal` for finite stop reasons. Prefer typed
Pydantic models. Validate before storage, queues, exports, or handoff.

### Why avoid a str return?

A structured result keeps tweet IDs, cursors, stop reasons, and errors.
A plain string cannot prove field completeness or types.

### How do I import agent from Pydantic AI?

Use `from pydantic_ai import Agent`. Import `BaseModel` from `pydantic` and
`Literal` from `typing`. Keep these imports separate from application tools.

### Does a Pydantic AI Twitter agent need X developer keys?

No. Use an Xquik API key. Some private reads and write actions also require a
connected X account.

### How do I post a tweet with Pydantic AI?

Use the matching X write route through `execute`.
Validate and approve each requested X write operation.
Store the idempotency key and returned action ID.

### How do I handle Twitter API rate limits in Python?

Read `error.retry_after` from a `429` response. Wait for that interval. Apply
bounded backoff when retrying safe read requests.

### How do I paginate tweet search in Pydantic AI?

Store `has_more` and `next_cursor` in the output model. Reuse unchanged
filters. Stop on completion, limits, or a stalled cursor.

### Can a Pydantic AI agent export Twitter followers?

Yes. Create an extraction, store its ID, and poll its status. Export only
after completion. Save the source user and format together.

### Can Pydantic AI monitor Twitter keywords?

Yes. Create a monitor and webhook. Store both identifiers together. Replay
missed events through `GET /api/v1/events` with `cursor` pagination. Monitor
delivery is asynchronous and cannot guarantee real-time events.

## Next steps

* Read the [official Pydantic AI MCP client guide](https://pydantic.dev/docs/ai/mcp/client/).
* Review the [MCP tool contract](/mcp/tools).
* Follow the [agent handoff checklist](/mcp/agent-handoff).
* Build a [tweet search workflow](/guides/tweet-scraper-csv-export).
* Add [monitor and webhook delivery](/guides/brand-monitoring-workflow).
* Compare [Twitter API alternatives](/twitter-api-alternatives).

Xquik is an independent third-party service. Not affiliated with X Corp.
"Twitter" and "X" are trademarks of X Corp.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.