---
title: "Pydantic AI"
description: "Crea agentes de Twitter API con Pydantic AI para búsqueda tipada de tweets, perfiles, seguidores y acciones en X revisadas a través de TwexAPI MCP."
---

Crea un agente de Twitter API con Pydantic AI a través del servidor MCP de TwexAPI. Busca tweets, inspecciona perfiles y valida cada entrega durable con un modelo Pydantic. Revisa cada publicación, respuesta, like, follow o mensaje directo propuesto.

## ¿Por qué usar Pydantic AI con TwexAPI?

Pydantic AI combina llamadas a herramientas del modelo con salida Python tipada. TwexAPI proporciona rutas de Twitter API a través de las herramientas MCP `explore` y `twexapi_request`.

| Límite | Control de Pydantic AI | Beneficio para el agente de Twitter |
| -------------------- | ---------------------- | -------------------------------------------- |
| Conexión MCP | `MCPServerStreamableHTTP` | Mantén las credenciales en tu proceso |
| Respuesta final | Pydantic `output_type` | Rechaza filas de tweets y cursores mal formados |
| Acciones de escritura | Paso de revisión humana | Pausa antes de publicar, responder o seguir |
| Ciclo de vida de conexión | `async with agent` | Reutiliza una sesión MCP en llamadas relacionadas |

Usa un agente tipado para búsqueda de tweets o enriquecimiento de perfiles. Mantén cada agente enfocado.

## Requisitos previos

* Python 3.10 o posterior
* Una [API key de TwexAPI](https://twexapi.io/dashboard)
* Un modelo compatible con Pydantic AI con llamadas a herramientas
* Una cookie de Twitter o `auth_token` para acciones de escritura

## Instalación

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

Guarda los secretos fuera del control de versiones.

```bash
export TWEXAPI_API_KEY="YOUR_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_KEY"
```

## Construir un agente tipado de búsqueda de tweets

Define la entrega final antes de crear el agente. Pydantic AI valida la salida del modelo contra este esquema.

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

El agente descubre `explore` y `twexapi_request`. Llama a `explore` primero cuando el agente no conoce una ruta o la forma de los parámetros.

## Validar campos antes del almacenamiento

Mantén estos campos de origen sin cambios en tu modelo de salida:

* Filas de tweet: `tweet_id`, `text`, `author_username`, `created_at`, `url`
* Filas de perfil: `user_id`, `username`, `name`, `description`, cantidades de seguidores
* Estado de página: `has_more`, `next_cursor`, o nombres de cursor específicos del idioma
* Recibos de escritura: nombre de ruta, ID de tweet devuelto, estado de confirmación

Nunca conviertas IDs grandes a números de punto flotante. Mapea los IDs de origen solo a `tweet_id` o `user_id` en tu modelo de salida.

Continúa a través de una página vacía cuando `has_more` siga siendo true. Detente cuando no exista cursor o el servidor repita un cursor. Devuelve `cursor_stalled` con el número de filas recopiladas.

## Construir una entrega tipada

<CardGroup cols={2}>
  <Card title="Búsqueda de tweets" icon="search">
    Almacena consulta, ruta, IDs de tweet, autores, URLs, `has_more`, `next_cursor` y razón de parada.
  </Card>

  <Card title="Consulta de perfil" icon="user-round">
    Almacena `user_id`, `username`, `name`, `description` y cantidades de seguidores.
  </Card>

  <Card title="Páginas de seguidores" icon="users">
    Almacena nombre de usuario de origen, filas de seguidores y checkpoint de cursor.
  </Card>

  <Card title="Acciones de escritura" icon="send">
    Almacena ruta, texto de vista previa, requisito de cookie y registro de aprobación.
  </Card>
</CardGroup>

Mantén las API keys fuera de la salida del agente. Consulta [Agent MCP Handoff](/mcp/agent-handoff).

## Reutilizar la conexión MCP

Envuelve llamadas relacionadas cuando varias ejecuciones deban compartir una conexión.

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

## Requerir aprobación antes de acciones en X

Los agentes de solo lectura pueden buscar tweets automáticamente. Los agentes con capacidad de escritura necesitan una decisión humana antes de cada acción en X.

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

Detén el agente antes de llamar rutas con `read_only: false`. Previsualiza cargas con el [CLI](/sdks/cli) `--dry-run`, luego ejecuta a través de REST o el [Python SDK](/sdks/python) después de la aprobación.

## Manejo de errores

| Estado | Decisión de Pydantic AI |
| ------ | ---------------------------------------------- |
| `400` | Corrige la solicitud antes de reintentar |
| `401` | Detente y reemplaza la credencial |
| `403` | Reporta problemas de acceso o créditos |
| `429` | Espera y conserva el cursor |
| `5xx` | Reintenta lecturas seguras con backoff acotado |

Nunca recrees una escritura pendiente después de un timeout sin verificar el estado devuelto.

## Elegir Pydantic AI MCP o REST

| Requisito | Elegir | Razón |
| ------------------------------------------------ | --------------- | --------------------------------------------------- |
| Un modelo selecciona operaciones de tweet o perfil | Pydantic AI MCP | El agente descubre rutas con `explore` |
| El código de aplicación llama una ruta conocida | [Python SDK](/sdks/python) | La solicitud permanece determinista |
| Un humano debe revisar una acción en X | Pydantic AI MCP + REST manual | Pausa antes de la ejecución |
| Una exportación programada corre sin modelo | [Prefect](/guides/prefect) o REST | No se requiere decisión del modelo |

## Versiones de paquetes

| Paquete | Rango compatible |
| ------------------ | --------------- |
| Python | `>=3.10` |
| `pydantic-ai` | `>=0.8` |
| `pydantic` | `>=2.7` |

## Próximos pasos

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