---
title: "Pydantic AI"
description: "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`.

| Frontière | Contrôle Pydantic AI | Bénéfice agent Twitter |
| -------------------- | ---------------------- | -------------------------------------------- |
| Connexion MCP | `MCPServerStreamableHTTP` | Gardez les identifiants dans votre processus |
| Réponse finale | `output_type` Pydantic | Rejetez les lignes de tweet et curseurs mal formés |
| Actions d'écriture | Étape de revue humaine | Pausez avant publication, réponse ou follow |
| Cycle de vie connexion | `async with agent` | Réutilisez une session MCP sur des appels liés |

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](https://twexapi.io/dashboard)
* Un modèle supporté par Pydantic AI avec appels d'outils
* Un cookie Twitter ou `auth_token` pour les actions d'écriture

## Installation

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

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

```bash
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.

```python
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é

<CardGroup cols={2}>
  <Card title="Recherche de tweets" icon="search">
    Stockez la requête, la route, les ID de tweet, les auteurs, les URLs, `has_more`, `next_cursor` et la raison d'arrêt.
  </Card>

  <Card title="Lookup de profil" icon="user-round">
    Stockez `user_id`, `username`, `name`, `description` et les compteurs d'abonnés.
  </Card>

  <Card title="Pages d'abonnés" icon="users">
    Stockez le username source, les lignes d'abonnés et le checkpoint curseur.
  </Card>

  <Card title="Actions d'écriture" icon="send">
    Stockez la route, le texte de prévisualisation, l'exigence cookie et l'enregistrement d'approbation.
  </Card>
</CardGroup>

Gardez les clés API hors de la sortie agent. Consultez [Agent MCP Handoff](/mcp/agent-handoff).

## Réutiliser la connexion MCP

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

```python
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.

```python
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](/sdks/cli) `--dry-run`, puis exécutez via REST ou le [SDK Python](/sdks/python) après approbation.

## Gestion des erreurs

| Statut | Décision Pydantic AI |
| ------ | ---------------------------------------------- |
| `400` | Corrigez la requête avant de réessayer |
| `401` | Arrêtez et remplacez l'identifiant |
| `403` | Signalez les problèmes d'accès ou de crédits |
| `429` | Ralentissez et préservez le curseur |
| `5xx` | Réessayez les lectures sûres avec backoff borné |

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

## Choisir Pydantic AI MCP ou REST

| Exigence | Choisir | Raison |
| ------------------------------------------------ | --------------- | --------------------------------------------------- |
| Un modèle sélectionne des opérations tweet ou profil | Pydantic AI MCP | L'agent découvre les routes avec `explore` |
| Le code applicatif appelle une route connue | [SDK Python](/sdks/python) | La requête reste déterministe |
| Un humain doit revoir une action X | Pydantic AI MCP + REST manuel | Pausez avant exécution |
| Un export planifié s'exécute sans modèle | [Prefect](/guides/prefect) ou REST | Aucune décision de modèle requise |

## Versions des packages

| Package | Plage supportée |
| ------------------ | --------------- |
| Python | `>=3.10` |
| `pydantic-ai` | `>=0.8` |
| `pydantic` | `>=2.7` |

## Prochaines étapes

* [Outils MCP](/mcp/tools)
* [Agent MCP Handoff](/mcp/agent-handoff)
* [LangChain](/guides/langchain)
* [SDK Python](/sdks/python)