Pydantic AI

Construisez des agents Twitter API Pydantic AI pour la recherche de tweets typée, les profils, les abonnés et les actions X revues via TwexAPI MCP.

Construisez un agent Twitter API Pydantic AI via le serveur MCP de TwexAPI. Recherchez des tweets, inspectez des profils et validez chaque handoff durable avec un modèle Pydantic. Revoyez chaque publication, réponse, like, follow ou message direct proposé.

Pourquoi utiliser Pydantic AI avec TwexAPI ?

Pydantic AI combine les appels d’outils du modèle avec une sortie Python typée. TwexAPI fournit les routes Twitter API via les outils MCP explore et twexapi_request.

Utilisez un agent typé pour la recherche de tweets ou l’enrichissement de profil. Gardez chaque agent focalisé.

Prérequis

• Python 3.10 ou plus récent
• Une clé API TwexAPI
• Un modèle supporté par Pydantic AI avec appels d’outils
• Un cookie Twitter ou auth_token pour les actions d’écriture

Installation

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

Stockez les secrets en dehors du contrôle de version.

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

Construire un agent de recherche de tweets typé

Définissez le handoff final avant de créer l’agent. Pydantic AI valide la sortie du modèle contre ce schéma.

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

L’agent découvre explore et twexapi_request. Appelez explore en premier lorsque l’agent ne connaît pas une route ou la forme des paramètres.

Valider les champs avant stockage

Gardez ces champs source inchangés dans votre modèle de sortie :

• Lignes de tweet : tweet_id, text, author_username, created_at, url
• Lignes de profil : user_id, username, name, description, compteurs d’abonnés
• État de page : has_more, next_cursor, ou noms de curseur spécifiques au langage
• Reçus d’écriture : nom de route, ID de tweet retourné, statut de confirmation

Ne jamais caster de grands ID en nombres flottants. Mappez les ID source vers tweet_id ou user_id uniquement dans votre modèle de sortie.

Continuez sur une page vide lorsque has_more reste vrai. Arrêtez lorsqu’aucun curseur existe ou le serveur répète un curseur. Retournez cursor_stalled avec le nombre de lignes collectées.

Construire un handoff typé

Gardez les clés API hors de la sortie agent. Consultez Agent MCP Handoff.

Réutiliser la connexion MCP

Enveloppez les appels liés lorsque plusieurs exécutions doivent partager une connexion.

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

Exiger une approbation avant les actions X

Les agents en lecture seule peuvent rechercher des tweets automatiquement. Les agents avec capacité d’écriture ont besoin d’une décision humaine avant chaque action X.

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

Arrêtez l’agent avant les routes read_only: false. Prévisualisez les payloads avec le CLI --dry-run, puis exécutez via REST ou le SDK Python après approbation.

Gestion des erreurs

Ne jamais recréer une écriture en attente après un timeout sans vérifier le statut retourné.

Choisir Pydantic AI MCP ou REST

Versions des packages

Prochaines étapes

• Outils MCP
• Agent MCP Handoff
• LangChain
• SDK Python