Haystack Twitter Search API & RAG Python Guide

Haystack is an open source framework for Python AI applications. Compare frameworks for building RAG and agent pipelines. Use xquik-haystack for current tweets. This Haystack AI framework gives you typed tweet Document objects. Each includes text, authors, metrics, and URLs. This Haystack AI API integration supports RAG pipelines and agent workflows. You can follow this guide when building AI applications with Haystack and current tweets. The integration never grants write access. The integration provides two read-only components:

Search Public Tweets

XquikTweetSearch searches keywords, hashtags, accounts, conversations, and exact phrases.

Fetch User Timelines

XquikUserTweetsFetcher retrieves one public account’s tweets and optional replies.

Use these components for search, timelines, monitoring, and retrieval-augmented generation. They can retrieve relevant tweets for Haystack AI agents and Haystack AI RAG pipelines. These components are building blocks for pipelines and agents. Wrap XquikTweetSearch with ComponentTool when an agent needs a search tool called search_current_tweets. Use the followers API for follower exports. Use the write API for approved publishing. Those actions stay outside this integration.

Install Haystack and Xquik

Use Python 3.10 or newer. Pin both packages for repeatable pipeline builds.

python -m pip install "xquik-haystack==0.1.3" "haystack-ai==3.0.0"

Install inside a virtual environment. Release 0.1.3 is published on PyPI. Haystack 3.0.0 supports sync and async runs. The pip install command pins both packages for repeatable builds. Create an Xquik API key, then export it locally.

export XQUIK_API_KEY="xq_..."

Never embed production keys in pipeline YAML, notebooks, or source control. Load the environment variable through a Haystack Secret object.

Search Twitter Tweets in Python

XquikTweetSearch calls the GET /x/tweets/search search endpoint. It accepts standard X search syntax and structured filters. Use it for each Twitter API keyword search or exact phrase query.

from haystack import Pipeline
from haystack.utils import Secret
from haystack_integrations.components.websearch.xquik import XquikTweetSearch

search = XquikTweetSearch(
    api_key=Secret.from_env_var("XQUIK_API_KEY"),
    top_k=20,
    query_type="Latest",
)

pipeline = Pipeline()
pipeline.add_component("twitter_search", search)

result = pipeline.run(
    {
        "twitter_search": {
            "query": '"retrieval augmented generation" lang:en -filter:retweets'
        }
    }
)

documents = result["twitter_search"]["documents"]
links = result["twitter_search"]["links"]

Use Latest for recent monitoring. Use Top for engagement-ranked discovery. Store tweet IDs because rankings can change. Save every search query beside its tweet IDs. Set top_k to cap the number of tweets in each run.

Build Focused Tweet Searches

Search intent	Query example
Exact phrase	`"retrieval augmented generation"`
Account posts	`from:deepset_ai haystack`
Hashtag search	`#haystack #rag`
Date window	`haystack since:2026-07-01 until:2026-08-01`
Exclude reposts	`haystack -filter:retweets`

The API also supports structured search filters. Pass them through extra_params during initialization.

search = XquikTweetSearch(
    top_k=50,
    query_type="Latest",
    extra_params={
        "language": "en",
        "mediaType": "links",
        "minFaves": 10,
        "replies": "exclude",
    },
)

See the tweet search API for author, reply, quote, URL, and conversation filters. Keep timestamps and cursors outside extra_params.

Choose a Search Window

Bound every retrieval with timestamps.

result = search.run(
    query="haystack agents",
    since_time="2026-07-31T00:00:00Z",
    until_time="2026-08-01T00:00:00Z",
)

Save each window. Deduplicate overlaps by Document.meta["id"].

Fetch a Twitter User Timeline

Use XquikUserTweetsFetcher for GET /x/users/{id}/tweets. Pass a username or numeric X user ID.

from haystack.utils import Secret
from haystack_integrations.components.websearch.xquik import (
    XquikUserTweetsFetcher,
)

timeline = XquikUserTweetsFetcher(
    api_key=Secret.from_env_var("XQUIK_API_KEY"),
    include_replies=False,
    include_parent_tweet=False,
)

result = timeline.run(user_id="deepset_ai")
documents = result["documents"]

Set include_replies=True for replies. Enable parent tweets only when reply context matters. Choose search for many accounts and timelines for one account.

Understand Haystack Document Fields

Each tweet becomes one Haystack Document. Tweet text becomes Document.content. Stable fields become metadata.

Document field	Stored tweet value
`content`	Tweet text
`meta.endpoint`	Search or timeline route
`meta.id`, `meta.url`	Tweet ID and canonical URL
`meta.created_at`, `meta.lang`	Timestamp and language
`meta.conversation_id`, `meta.in_reply_to_*`	Conversation and parent fields
`meta.is_reply`, `meta.is_quote_status`	Reply and quote flags
`meta.like_count`, `meta.retweet_count`, `meta.reply_count`	Likes, reposts, and replies
`meta.quote_count`, `meta.view_count`, `meta.bookmark_count`	Quotes, views, and bookmarks
`meta.author`	Author ID, username, name, and verified flag

Missing fields stay absent. Never treat missing metrics as zero. The links output contains each available meta.url.

Build Reliable RAG Citations

Store tweet IDs and canonical URLs before embedding tweet text. This preserves evidence after ranking or joining.

citation_rows = []

for document in documents:
    tweet_id = document.meta.get("id")
    tweet_url = document.meta.get("url")
    if tweet_id and tweet_url:
        citation_rows.append(
            {
                "tweet_id": tweet_id,
                "url": tweet_url,
                "created_at": document.meta.get("created_at"),
                "author": document.meta.get("author"),
            }
        )

Require supplied URLs for citations. Reject URLs absent from retrieved documents. Preserve conversation_id for reply threads.

Treat Tweet Text as Untrusted Context

Tweets can contain prompt injection and unsafe URLs. Never treat tweet text as a system instruction. Keep tweets separate from instructions and tool permissions. Limit retrieval by topic and time. Preserve tweet IDs, authors, timestamps, and URLs. Require citations. Review sensitive conclusions. Likes and reposts rank results. They do not prove accuracy. Treat outputs from large language models (LLMs) as proposals, not evidence.

Index Tweets or Retrieve Them Live

Choose the pipeline pattern that matches freshness requirements.

Live Retrieval

Search during each question for recent tweets and active events.

Indexed Corpus

Store embeddings for repeated research across stable windows.

Expect current results, not guaranteed real time delivery. Keep tweet records separate from embeddings. Use meta.id for deduplication.

Paginate Without Duplicate Tweets

Both components return has_more and next_cursor. Keep the request unchanged. Treat cursors as opaque strings.

search = XquikTweetSearch(top_k=100, query_type="Latest")
page = search.run(query="haystack ai")

documents_by_tweet_id = {}

while True:
    for document in page["documents"]:
        tweet_id = document.meta.get("id")
        if tweet_id:
            documents_by_tweet_id[tweet_id] = document

    if not page["has_more"] or not page["next_cursor"]:
        break

    page = search.run(
        query="haystack ai",
        cursor=page["next_cursor"],
    )

documents = list(documents_by_tweet_id.values())

Save each cursor after its documents. Never edit a cursor. Deduplicate on Document.meta["id"].

Run Haystack Pipelines Asynchronously

Haystack 3 uses one Pipeline class. Both Xquik components expose run_async().

import asyncio

from haystack import Pipeline
from haystack_integrations.components.websearch.xquik import XquikTweetSearch


async def search_tweets():
    pipeline = Pipeline()
    pipeline.add_component(
        "twitter_search",
        XquikTweetSearch(top_k=25, query_type="Latest"),
    )
    return await pipeline.run_async(
        {"twitter_search": {"query": "haystack agents"}}
    )


result = asyncio.run(search_tweets())

Use async runs in web servers. Use sync runs for scripts and scheduled jobs.

Handle Every Documented Error

The integration raises httpx.HTTPStatusError. Branch on the canonical status before retrying.

Status	Meaning	Action
`400`	Invalid request	Fix it. Do not retry unchanged.
`401`	Invalid API key	Add a valid API key.
`402`	Account action needed	Check the account balance.
`404`	Timeline user not found	Check the username or user ID.
`424`	X retrieval dependency failed	Retry with bounded backoff.
`429`	Rate limit exceeded	Back off before retrying.
`502`	X retrieval error	Retry up to the configured limit.

import httpx

from haystack_integrations.components.websearch.xquik import XquikTweetSearch

search = XquikTweetSearch(top_k=20, max_retries=3)

try:
    result = search.run(query="haystack ai")
except httpx.HTTPStatusError as error:
    status = error.response.status_code
    if status in {400, 401, 402, 404}:
        raise RuntimeError(f"Fix Xquik request before retrying: HTTP {status}") from error
    if status in {424, 429, 502}:
        raise RuntimeError(f"Retry Xquik request with backoff: HTTP {status}") from error
    raise

Record status codes, never credentials. Cap retries to prevent unbounded agent loops.

Pipeline Handoff

Use this shape when Haystack hands results to a vector store, evaluation job, queue, CSV export, or dashboard.

Document Rows

Store content, tweet IDs, URLs, timestamps, authors, and public metrics.

Citation Links

Join each canonical URL to meta.id.

Pagination Checkpoint

Store the request, options, has_more, and next_cursor.

Failure Branch

Store each HTTP status with its pipeline run ID.

Keep handoff records separate from embeddings. Later runs can refresh tweets without rebuilding the pipeline.

Haystack Component or Direct REST API

Does your pipeline already return Document objects? Add these components directly. They normalize tweet text, metadata, URLs, and pagination. Use direct Xquik REST routes for followers, following, replies, quotes, reposts, lists, communities, media, trends, monitors, approved writes, and extra query options. Both approaches use the same Xquik contracts. Components only support tweet search and user timelines.

Migrate to Haystack 3

Haystack 3 replaces AsyncPipeline with Pipeline. Call await pipeline.run_async(...) for concurrent execution. Keep meta.id as the tweet identity. Test with haystack-ai==3.0.0 before upgrading. Check the official migration guide for other changes.

Common Haystack Twitter API Questions

What Is Haystack AI?

Haystack links search, RAG, and AI agents inside Python pipelines.

How Does the Twitter API Search Tweets?

It accepts queries, filters, ordering, and cursors. Xquik returns tweets and citation URLs.

What Is a Twitter Search API Python Workflow?

Install both packages. Run XquikTweetSearch, then process documents, links, and cursors.

How Does an API Twitter Search Workflow Preserve Cursors?

Save next_cursor after each page. Reuse every filter.

How Does the Twitter API Search Tweets by Keyword?

Pass keywords, phrases, hashtags, accounts, or filters. Choose Latest or Top.

Haystack AI vs LangChain: Which Fits Twitter RAG?

Choose Haystack for Document pipelines. Choose LangChain for its retrievers and tools.

Where Is the Haystack AI GitHub Integration?

The Xquik Haystack repository contains the package and offline tests.

Can a Haystack AI Agent Publish Tweets?

No. This integration reads searches and timelines. Use the write API for approved tweets.

Does This Require a Haystack Enterprise Platform?

No. Run the open-source components in your Haystack setup.

Search Public Tweets

Fetch User Timelines

​Install Haystack and Xquik

​Search Twitter Tweets in Python

​Build Focused Tweet Searches

​Choose a Search Window

​Fetch a Twitter User Timeline

​Understand Haystack Document Fields

​Build Reliable RAG Citations

​Treat Tweet Text as Untrusted Context

​Index Tweets or Retrieve Them Live

Live Retrieval

Indexed Corpus

​Paginate Without Duplicate Tweets

​Run Haystack Pipelines Asynchronously

​Handle Every Documented Error

​Pipeline Handoff

Document Rows

Citation Links

Pagination Checkpoint

Failure Branch

​Haystack Component or Direct REST API

​Migrate to Haystack 3

​Common Haystack Twitter API Questions

​What Is Haystack AI?

​How Does the Twitter API Search Tweets?

​What Is a Twitter Search API Python Workflow?

​How Does an API Twitter Search Workflow Preserve Cursors?

​How Does the Twitter API Search Tweets by Keyword?

​Haystack AI vs LangChain: Which Fits Twitter RAG?

​Where Is the Haystack AI GitHub Integration?

​Can a Haystack AI Agent Publish Tweets?

​Does This Require a Haystack Enterprise Platform?

​Source and Contracts