---
title: "Prefect"
description: "Crea flujos de trabajo programados de búsqueda de Twitter, perfiles, líneas de tiempo y tendencias en Python con tareas de solo lectura de Prefect y TwexAPI."
---

Prefect es un orquestador de flujos de trabajo de código abierto para Python. Usa [`prefect-x-api-scraper`](https://github.com/twexapi-dev/prefect-x-api-scraper) para búsquedas de tweets repetibles, consultas de perfiles, actualizaciones de líneas de tiempo y verificaciones de tendencias.

La colección es de solo lectura. Proporciona seis tareas asíncronas de Prefect. Cada tarea devuelve la respuesta JSON canónica de TwexAPI como un diccionario de Python.

<CardGroup cols={3}>
  <Card title="Buscar tweets" icon="search">
    Busca palabras clave, hashtags, cuentas, fechas y operadores de consulta de X.
  </Card>

  <Card title="Consultar tweets" icon="message-square">
    Obtén un tweet público a partir de su ID numérico.
  </Card>

  <Card title="Buscar perfiles" icon="users">
    Encuentra cuentas públicas de X por nombre, nombre de usuario o tema.
  </Card>

  <Card title="Consultar perfiles" icon="user-round">
    Recupera un perfil público por screen name.
  </Card>

  <Card title="Actualizar líneas de tiempo" icon="list">
    Obtén tweets recientes de una página de línea de tiempo de usuario.
  </Card>

  <Card title="Rastrear tendencias" icon="trending-up">
    Recupera tweets en tendencia global por país, tema y etiqueta de contenido.
  </Card>
</CardGroup>

Usa esta colección para investigación, enriquecimiento, dashboards, alertas e indexación. Usa REST directo, el [Python SDK](/sdks/python) o [MCP](/mcp/overview) para escrituras, paginación de seguidores y exportaciones.

## Instalación

Usa Python 3.10 o posterior.

```bash
python -m pip install "prefect>=3.0.0" "prefect-x-api-scraper"
```

Registra el bloque de credenciales después de instalar Prefect:

```bash
prefect block register -m prefect_x_api_scraper
```

## Guarda la API key de TwexAPI

Crea una API key desde el [panel de TwexAPI](https://twexapi.io/dashboard). Guárdala dentro de un bloque `TwexApiCredentials`.

```python
from prefect_x_api_scraper import TwexApiCredentials

credentials = TwexApiCredentials(
    api_key="YOUR_API_KEY",
    base_url="https://api.twexapi.io",
    timeout_seconds=30,
)
credentials.save("twexapi", overwrite=True)
```

Los bloques de Prefect almacenan configuración tipada entre flujos y despliegues. Nunca coloques API keys en YAML de despliegue, parámetros de flujo, logs o repositorios.

## Elige la tarea de Prefect correcta

<CardGroup cols={2}>
  <Card title="search_tweets" icon="search">
    Llama a `POST /twitter/advanced_search/page`. Acepta `query`, `cursor`, `sort_by` y campos de paginación.
  </Card>

  <Card title="get_tweet" icon="message-square">
    Llama a `POST /v2/tweet/detail`. Pasa un ID numérico de tweet.
  </Card>

  <Card title="search_users" icon="users">
    Llama a `GET /twitter/search-user/{keyword}/{target_count}`. Acepta una consulta de perfil.
  </Card>

  <Card title="get_user" icon="user-round">
    Llama a `GET /twitter/{screen_name}/about`. Acepta screen names con o sin `@`.
  </Card>

  <Card title="get_user_tweets" icon="list">
    Llama a `POST /twitter/{screen_name}/timeline/page`. Soporta cursores y tamaño de página.
  </Card>

  <Card title="get_trends" icon="trending-up">
    Llama a `GET /twitter/global-trending/tweets`. Acepta `country`, `topic`, `content` y `count`.
  </Card>
</CardGroup>

## Construye un flujo de automatización de Twitter en Python

Este flujo busca publicaciones recientes y normaliza filas de tweets. Conserva IDs, autores, marcas de tiempo, métricas, URLs y estado de paginación.

```python
from __future__ import annotations

from typing import Any

from prefect import flow, task
from prefect_x_api_scraper import TwexApiCredentials, search_tweets


@task
async def normalize_tweet_page(page: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for tweet in page.get("data", {}).get("tweets", page.get("tweets", [])):
        if not isinstance(tweet, dict):
            continue
        author = tweet.get("author") or tweet.get("user")
        rows.append(
            {
                "tweet_id": tweet.get("id") or tweet.get("tweet_id"),
                "text": tweet.get("text") or tweet.get("full_text"),
                "created_at": tweet.get("created_at") or tweet.get("createdAt"),
                "author": author if isinstance(author, dict) else {},
                "like_count": tweet.get("like_count") or tweet.get("likeCount"),
                "repost_count": tweet.get("retweet_count") or tweet.get("retweetCount"),
                "reply_count": tweet.get("reply_count") or tweet.get("replyCount"),
            }
        )
    return rows


@flow(name="TwexAPI Twitter Search")
async def social_signal_flow() -> dict[str, Any]:
    credentials = TwexApiCredentials.load("twexapi")
    page = await search_tweets(
        credentials,
        '"workflow orchestration" lang:en -filter:retweets',
        sort_by="Latest",
    )
    rows = await normalize_tweet_page(page)
    return {
        "tweet_rows": rows,
        "has_more": page.get("has_more") or page.get("has_next_page", False),
        "next_cursor": page.get("next_cursor") or page.get("nextCursor"),
    }
```

Ejecútalo con asyncio:

```python
import asyncio

result = asyncio.run(social_signal_flow())
```

## Escribe consultas de búsqueda de tweets enfocadas

<CardGroup cols={2}>
  <Card title="Frase exacta" icon="quote">
    Usa `"workflow orchestration"` para una frase exacta.
  </Card>

  <Card title="Filtro de cuenta" icon="at-sign">
    Usa `from:PrefectIO` para tweets de una cuenta.
  </Card>

  <Card title="Búsqueda por hashtag" icon="hash">
    Usa `#prefect #python` para hashtags coincidentes.
  </Card>

  <Card title="Ventana de fechas" icon="calendar-range">
    Usa fechas `since:` y `until:` dentro de una consulta.
  </Card>

  <Card title="Excluir reposts" icon="repeat-2">
    Usa `-filter:retweets` cuando importen las publicaciones originales.
  </Card>
</CardGroup>

Usa `sort_by="Latest"` para monitoreo cronológico. Usa `sort_by="Top"` para descubrimiento ordenado por engagement. Persiste los IDs de tweet porque los rankings pueden cambiar.

## Programa el pipeline de búsqueda de Twitter

Usa `.serve()` para un proceso local de larga duración.

```python
from prefect.schedules import Cron


if __name__ == "__main__":
    social_signal_flow.serve(
        name="twexapi-social-signals",
        schedule=Cron("0 * * * *", timezone="UTC"),
    )
```

Usa un despliegue con work pool para workers de Docker, Kubernetes o serverless.

```bash
prefect deploy social_signal_flow.py:social_signal_flow \
  --name twexapi-social-signals \
  --pool production
```

## Pagina resultados de tweets y perfiles

La paginación por cursor continúa búsquedas grandes sin adivinar números de página. Mantén la solicitud original sin cambios. Pasa solo el cursor devuelto.

```python
from typing import Any, Optional

from prefect import flow
from prefect_x_api_scraper import TwexApiCredentials, search_tweets


@flow
async def collect_tweet_pages(query: str) -> list[dict[str, Any]]:
    credentials = TwexApiCredentials.load("twexapi")
    rows_by_id: dict[str, dict[str, Any]] = {}
    cursor: Optional[str] = None

    while True:
        page = await search_tweets(
            credentials,
            query=query,
            sort_by="Latest",
            cursor=cursor,
        )

        tweets = page.get("data", {}).get("tweets", page.get("tweets", []))
        for tweet in tweets:
            if isinstance(tweet, dict):
                tweet_id = tweet.get("id") or tweet.get("tweet_id")
                if tweet_id:
                    rows_by_id[str(tweet_id)] = tweet

        cursor_value = page.get("next_cursor") or page.get("nextCursor")
        has_more = bool(page.get("has_more") or page.get("has_next_page", False))
        if not has_more or not cursor_value:
            break

        cursor = str(cursor_value)

    return list(rows_by_id.values())
```

No decodifiques cursores. Trátalos como cadenas opacas. Guarda cada cursor después de persistir su página.

## Haz que las ejecuciones programadas sean idempotentes

<CardGroup cols={2}>
  <Card title="Identidad de tweet" icon="message-square">
    Haz upsert de filas de tweets por `tweet_id`. No uses el texto como clave.
  </Card>

  <Card title="Identidad de perfil" icon="user-round">
    Haz upsert de filas de perfil por ID numérico de usuario.
  </Card>

  <Card title="Checkpoint de cursor" icon="list-tree">
    Guarda `has_more` y `next_cursor` después de cada página confirmada.
  </Card>
</CardGroup>

## Reintenta solo fallos transitorios

Prefect soporta retrasos de reintento, jitter y reintentos condicionales. No reintentes entradas inválidas sin cambios.

```python
from prefect_x_api_scraper import search_tweets

search_recent_tweets = search_tweets.with_options(
    name="Search Recent Tweets",
    retries=4,
    retry_delay_seconds=[5, 15, 45, 120],
)
```

## Enruta errores documentados

| Estado | Acción |
| ------ | ------ |
| `400` | Corrige consultas faltantes o entrada malformada. No reintentes sin cambios. |
| `401` | Carga una API key válida desde el bloque de credenciales. |
| `403` | Resuelve el acceso a la cuenta o los créditos antes de la siguiente ejecución. |
| `404` | Verifica IDs de tweet, nombres de usuario y screen names. |
| `429` | Reduce la frecuencia del horario y reintenta con retrasos con jitter. |

Consulta [Error Handling](/guides/error-handling) y [Rate Limits](/guides/rate-limits) para orientación completa de recuperación.
| `5xx` | Reintenta fallos transitorios con un límite. |

## Alternativa de planificación con MCP

Usa [Twexapi MCP](/mcp/overview) durante la planificación para descubrir endpoints, luego codifica la ruta seleccionada en tareas de Prefect o llamadas REST.

Prompt de planificación sugerido:

```txt
Use Twexapi MCP explore to choose the best endpoint for daily AI trend collection.
Return the exact method, relative path, query parameters, pagination fields, expected response fields, and retry guidance.
Do not execute write endpoints.
```

## Colección de Prefect o API directa de TwexAPI

Elige `prefect-x-api-scraper` por sus seis lecturas compatibles. Proporciona bloques, llamadas asíncronas, validación y metadatos de tareas.

Usa REST directo, el [Python SDK](/sdks/python) o MCP para:

* Acciones de tweet, respuesta, cita, like, follow o DM
* Exportaciones de seguidores y following
* Listas y comunidades

Para verificaciones recurrentes, programa flujos de Prefect o paginación REST en lugar de endpoints de monitor nativos. TwexAPI no expone trabajos de extracción CSV/JSON asíncronos como un producto separado.

Mantén las escrituras detrás de aprobación explícita. Guarda IDs confirmados antes de reintentar cualquier acción de escritura.

## Fuente y próximos pasos

* [Código fuente de prefect-x-api-scraper](https://github.com/twexapi-dev/prefect-x-api-scraper)
* [Horarios de Prefect](https://docs.prefect.io/v3/concepts/schedules)
* [Reintentos de tareas de Prefect](https://docs.prefect.io/v3/how-to-guides/workflows/retries)
* [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)
* [Global Trending Tweets](/api-reference/trending-endpoints/get-global-trending-tweets-api-twitter-global-trending-tweets-get)
* [Agent MCP Handoff](/mcp/agent-handoff)