Prefect

Build scheduled Twitter search, profile, timeline, and trend workflows in Python with Prefect and TwexAPI read-only tasks.

Prefect is an open-source workflow orchestrator for Python code. Use prefect-x-api-scraper for repeatable tweet searches, profile lookups, timeline refreshes, and trend checks.

The collection is read-only. It provides six asynchronous Prefect tasks. Each task returns the canonical TwexAPI JSON response as a Python dictionary.

Use this collection for research, enrichment, dashboards, alerts, and indexing. Use direct REST, the Python SDK, or MCP for writes, follower pagination, and exports.

Install

Use Python 3.10 or newer.

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

Register the credentials block after installing Prefect:

prefect block register -m prefect_x_api_scraper

Store the TwexAPI API key

Create an API key from the TwexAPI dashboard. Store it inside a TwexApiCredentials block.

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)

Prefect blocks store typed configuration across flows and deployments. Never place API keys in deployment YAML, flow parameters, logs, or repositories.

Choose the right Prefect task

Build a Twitter automation flow in Python

This flow searches recent posts and normalizes tweet rows. It preserves IDs, authors, timestamps, metrics, URLs, and pagination state.

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"),
    }

Run it with asyncio:

import asyncio

result = asyncio.run(social_signal_flow())

Write focused tweet search queries

Use sort_by="Latest" for chronological monitoring. Use sort_by="Top" for engagement-ranked discovery. Persist tweet IDs because rankings can change.

Schedule the Twitter search pipeline

Use .serve() for a long-running local process.

from prefect.schedules import Cron


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

Use a work-pool deployment for Docker, Kubernetes, or serverless workers.

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

Paginate tweet and profile results

Cursor pagination continues large searches without guessing page numbers. Keep the original request unchanged. Pass only the returned cursor.

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

Do not decode cursors. Treat them as opaque strings. Store each cursor after persisting its page.

Make scheduled runs idempotent

Retry only transient failures

Prefect supports retry delays, jitter, and conditional retries. Do not retry invalid inputs unchanged.

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],
)

Route documented errors

See Error Handling and Rate Limits for full recovery guidance. | 5xx | Retry transient failures with a cap. |

MCP planning alternative

Use Twexapi MCP during planning to discover endpoints, then codify the selected route in Prefect tasks or REST calls.

Suggested planning prompt:

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.

Prefect collection or direct TwexAPI API

Choose prefect-x-api-scraper for its six supported reads. It provides blocks, async calls, validation, and task metadata.

Use direct REST, the Python SDK, or MCP for:

• Tweet, reply, quote, like, follow, or DM actions
• Follower and following exports
• Lists and communities

For recurring checks, schedule Prefect flows or REST pagination instead of native monitor endpoints. TwexAPI does not expose async CSV/JSON extraction jobs as a separate product.

Keep writes behind explicit approval. Store confirmed IDs before retrying any write action.

Source and next steps

• prefect-x-api-scraper source
• Prefect schedules
• Prefect task retries
• Advanced Twitter Search
• Global Trending Tweets
• Agent MCP Handoff