Pydantic AI

Build Pydantic AI Twitter API agents for typed tweet search, profiles, followers, and reviewed X actions through TwexAPI MCP.

Build a Pydantic AI Twitter API agent through TwexAPI’s MCP server. Search tweets, inspect profiles, and validate every durable handoff with a Pydantic model. Review every proposed post, reply, like, follow, or direct message.

Why use Pydantic AI with TwexAPI?

Pydantic AI combines model tool calls with typed Python output. TwexAPI supplies Twitter API routes through MCP tools explore and twexapi_request.

Use one typed agent for tweet search or profile enrichment. Keep each agent focused.

Prerequisites

• Python 3.10 or later
• A TwexAPI API key
• A Pydantic AI-supported model with tool calling
• A Twitter cookie or auth_token for write actions

Install

python -m pip install "pydantic-ai[mcp]" python-dotenv

Store secrets outside source control.

export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"

Build a typed tweet search agent

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

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.mcp import MCPServerStreamableHTTP


class TweetRow(BaseModel):
    tweet_id: str
    text: str
    author_username: str | None = None
    created_at: str | 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()

    server = MCPServerStreamableHTTP(
        "https://api.twexapi.io/mcp",
        headers={"x-api-key": os.environ["TWEXAPI_API_KEY"]},
    )

    agent = Agent(
        "anthropic:claude-sonnet-4-20250514",
        toolsets=[server],
        output_type=TweetSearchHandoff,
        instructions=(
            "Use TwexAPI MCP for Twitter API requests. Call explore before twexapi_request. "
            "Preserve exact IDs and cursors. Never invent missing tweet fields. "
            "Ask for confirmation before read_only: false actions."
        ),
    )

    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("twexapi-pydantic-ai-handoff.json").write_text(
        result.output.model_dump_json(indent=2),
        encoding="utf-8",
    )


asyncio.run(main())

The agent discovers explore and twexapi_request. Call explore first when the agent does not know a route or parameter shape.

Validate fields before storage

Keep these source fields unchanged in your output model:

• Tweet rows: tweet_id, text, author_username, created_at, url
• Profile rows: user_id, username, name, description, follower counts
• Page state: has_more, next_cursor, or language-specific cursor names
• Write receipts: route name, returned tweet ID, confirmation status

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

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

Build a typed handoff

Keep API keys outside agent output. See Agent MCP Handoff.

Reuse the MCP connection

Wrap related calls when several runs should share one connection.

async def collect_two_search_pages(agent: Agent) -> 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:
            return

        second_page = await agent.run(
            f"Continue tweet search for {first_page.output.query!r}. "
            f"Use explore, then twexapi_request with cursor {cursor!r}."
        )
        _ = second_page

Require approval before X actions

Read-only agents can search tweets automatically. Write-capable agents need a human decision before every X action.

class WritePlan(BaseModel):
    action: str
    endpoint: str
    preview_text: str
    requires_human_confirmation: bool = True

Stop the agent before calling read_only: false routes. Preview payloads with the CLI --dry-run, then execute through REST or the Python SDK after approval.

Error handling

Never recreate a pending write after a timeout without checking the returned status.

Choose Pydantic AI MCP or REST

Package versions

Next steps

• MCP Tools
• Agent MCP Handoff
• LangChain
• Python SDK