Haystack

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.

Tweet Search

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 Timeline

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="username")
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:username")
documents = list(page["documents"])

while page["has_more"] and page["next_cursor"]:
    page = search.run(query="from:username", cursor=page["next_cursor"])
    documents.extend(page["documents"])

Use the same pattern with XquikUserTweetsFetcher.run(user_id="username", cursor=...).

Runtime Options

Component	Init option	Default	Xquik mapping
`XquikTweetSearch`	`top_k`	`20`	`limit`
`XquikTweetSearch`	`query_type`	`Latest`	`queryType`
`XquikTweetSearch`	`extra_params`	`None`	Merged into search query parameters
`XquikUserTweetsFetcher`	`include_replies`	`False`	`includeReplies`
`XquikUserTweetsFetcher`	`include_parent_tweet`	`False`	`includeParentTweet`
Both	`base_url`	`https://xquik.com/api/v1`	Xquik API origin
Both	`timeout`	`10`	HTTP timeout in seconds
Both	`max_retries`	`3`	Haystack 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 field	Value
`content`	Tweet text. Empty string if no text is present.
`meta.endpoint`	`x.tweets.search` or `x.users.tweets`
`meta.id`	Tweet ID when present
`meta.url`	Tweet URL when present
`meta.created_at`	Tweet creation time when present
`meta.lang`	Language code when present
`meta.conversation_id`	Conversation ID when present
`meta.in_reply_to_id`	Parent tweet ID when present
`meta.in_reply_to_user_id`	Parent user ID when present
`meta.in_reply_to_username`	Parent username when present
`meta.is_reply`	Reply flag when present
`meta.is_quote_status`	Quote flag when present
`meta.like_count`	Like count when present
`meta.retweet_count`	Repost count when present
`meta.reply_count`	Reply count when present
`meta.quote_count`	Quote count when present
`meta.view_count`	View count when present
`meta.bookmark_count`	Bookmark count when present
`meta.author`	Author `id`, `username`, `name`, and `verified` fields when present

The links output contains each non-empty tweet URL found in Document.meta["url"].

Pipeline Handoff

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

Document Rows

Store each Document.content, meta.endpoint, meta.id, meta.url, meta.created_at, meta.author.id, meta.author.username, meta.author.name, meta.author.verified, and public metrics before embedding or export.

Citation Links

Store links as the canonical tweet URLs returned by the component. Join them to meta.id when a citation, audit row, or UI card needs a source link.

Pagination Checkpoint

Store request query or user_id, component options, has_more, and next_cursor. Resume with cursor=next_cursor; do not decode cursors.

Failure Branch

Catch httpx.HTTPStatusError, branch on response.status_code, and store the status with the pipeline run ID instead of retrying bad inputs unchanged.

Keep document_rows, citation_rows, and pagination_checkpoints separate from embeddings so later reruns can refresh X context without rebuilding the whole pipeline.

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

Getting Started

SDKs

Guides

Webhooks

MCP Server

Agent Plugins

Framework Guides

OAuth

MPP

Resources

Prerequisites

Install

Components

XquikTweetSearch

XquikUserTweetsFetcher

Pagination Handoff

Pipeline Ready

Tweet Search

User Timeline

Runtime Options

Document Mapping

Pipeline Handoff

Document Rows

Citation Links

Pagination Checkpoint

Failure Branch

Async Usage

Error Handling

Source

​Prerequisites

​Install

​Components

XquikTweetSearch

XquikUserTweetsFetcher

Pagination Handoff

Pipeline Ready

​Tweet Search

​User Timeline

​Pagination

​Runtime Options

​Document Mapping

​Pipeline Handoff

Document Rows

Citation Links

Pagination Checkpoint

Failure Branch

​Async Usage

​Error Handling

​Source

Prerequisites

Install

Components

Tweet Search

User Timeline

Pagination

Runtime Options

Document Mapping

Pipeline Handoff

Async Usage

Error Handling

Source