Pydantic AI MCP Twitter API agent guide for Python

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

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=["explore", "xquik"],
    )

    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())

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",
    )

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 == "xquik"
)

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),
    )

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()],
)

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],
)