Pydantic AI

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.

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
• Un modelo compatible con Pydantic AI con llamadas a herramientas
• Una cookie de Twitter o auth_token para acciones de escritura

Instalación

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

Guarda los secretos fuera del control de versiones.

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.

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

Mantén las API keys fuera de la salida del agente. Consulta Agent MCP Handoff.

Reutilizar la conexión MCP

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

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.

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 --dry-run, luego ejecuta a través de REST o el Python SDK después de la aprobación.

Manejo de errores

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

Elegir Pydantic AI MCP o REST

Versiones de paquetes

Próximos pasos

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