Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.xquik.com/llms.txt

Use this file to discover all available pages before exploring further.

Use xquik-haystack when a Haystack pipeline needs fresh public X context from Xquik’s REST API. The package exposes read-only web search components that return Haystack Document objects for tweet search and user timelines.

​
Prerequisites

  • Python 3.10+
  • Xquik API key (xq_...)
  • Haystack application or pipeline using haystack-ai

​
Install

Install the current GitHub build: 
pip install git+https://github.com/Xquik-dev/xquik-haystack.git
The package is not published on PyPI yet. Use the GitHub install URL until the first package release is available.

​
Components

XquikTweetSearch

Calls GET /x/tweets/search and returns matching public posts as Haystack Document objects.

XquikUserTweetsFetcher

Calls GET /x/users/{id}/tweets and returns recent public posts from one user as Haystack Document objects.

Pagination Handoff

Returns has_more and next_cursor so the next run can fetch the next Xquik page.

Pipeline Ready

Returns documents for direct use in Haystack pipelines, retrievers, joiners, rankers, and generators.
Both components read XQUIK_API_KEY by default, accept haystack.utils.Secret for explicit key injection, support sync and async runs, and send xquik-api-contract: 2026-04-29. Use XquikTweetSearch when the pipeline needs matching tweets for a keyword, hashtag, account filter, or X search operator. 
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=10,
    query_type="Latest",
)

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

result = pipeline.run({"x_search": {"query": "haystack ai"}})
documents = result["x_search"]["documents"]
top_k maps to the Xquik limit query parameter. query_type accepts Latest for chronological results or Top for engagement-ranked results.

​
User Tweets

Use XquikUserTweetsFetcher when the pipeline needs recent public posts from one username or X user ID. 
from haystack.utils import Secret
from haystack_integrations.components.websearch.xquik import XquikUserTweetsFetcher

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

result = fetcher.run(user_id="xquikcom")
documents = result["documents"]
Pass include_replies=True to include replies. Pass include_parent_tweet=True when reply rows should include parent tweet data when available.

​
Pagination

The components normalize Xquik pagination into Haystack-friendly output keys: 
from haystack_integrations.components.websearch.xquik import XquikTweetSearch

search = XquikTweetSearch(top_k=None)

page = search.run(query="from:xquikcom")
documents = list(page["documents"])

while page["has_more"] and page["next_cursor"]:
    page = search.run(query="from:xquikcom", cursor=page["next_cursor"])
    documents.extend(page["documents"])
Use the same pattern with XquikUserTweetsFetcher.run(user_id="xquikcom", cursor=...).

​
Runtime Options

ComponentInit optionDefaultXquik mapping
XquikTweetSearchtop_k20limit
XquikTweetSearchquery_typeLatestqueryType
XquikTweetSearchextra_paramsNoneMerged into search query parameters
XquikUserTweetsFetcherinclude_repliesFalseincludeReplies
XquikUserTweetsFetcherinclude_parent_tweetFalseincludeParentTweet
Bothbase_urlhttps://xquik.com/api/v1Xquik API origin
Bothtimeout10HTTP timeout in seconds
Bothmax_retries3Haystack request retry attempts
Per-run overrides are available for top_k, query_type, cursor, since_time, and until_time on tweet search, and for cursor, include_replies, and include_parent_tweet on user tweets.

​
Document Mapping

Each returned tweet becomes a Haystack Document.
Document fieldValue
contentTweet text. Empty string if no text is present.
meta.endpointx.tweets.search or x.users.tweets
meta.idTweet ID when present
meta.urlTweet URL when present
meta.created_atTweet creation time when present
meta.langLanguage code when present
meta.conversation_idConversation ID when present
meta.in_reply_to_idParent tweet ID when present
meta.in_reply_to_user_idParent user ID when present
meta.in_reply_to_usernameParent username when present
meta.is_replyReply flag when present
meta.is_quote_statusQuote flag when present
meta.like_countLike count when present
meta.retweet_countRepost count when present
meta.reply_countReply count when present
meta.quote_countQuote count when present
meta.view_countView count when present
meta.bookmark_countBookmark count when present
meta.authorAuthor id, username, name, and verified fields when present
The links output contains each non-empty tweet URL found in Document.meta["url"].

​
Async Usage

Both components expose run_async() with the same inputs and outputs as run(). 
import asyncio
from haystack_integrations.components.websearch.xquik import XquikTweetSearch


async def main():
    search = XquikTweetSearch(top_k=5)
    result = await search.run_async(query="haystack ai")
    return result["documents"]


documents = asyncio.run(main())

​
Error Handling

Xquik HTTP errors are raised through httpx.HTTPStatusError after the response status check. Handle authentication, rate limit, payment, and validation errors where your Haystack app normally handles retriever or web search failures. 
import httpx
from haystack_integrations.components.websearch.xquik import XquikTweetSearch

search = XquikTweetSearch(top_k=5)

try:
    result = search.run(query="haystack ai")
except httpx.HTTPStatusError as exc:
    status = exc.response.status_code
    raise RuntimeError(f"Xquik search failed with HTTP {status}") from exc

​
Source

Last modified on May 15, 2026