Add Xquik OpenAPI tool server

README.md

@@ -50,6 +50,7 @@ Reference implementations provided in this repository demonstrate common use-cas
 - [**Get User Info Server**](servers/get-user-info) - Access and return enriched user profile information from authentication providers or internal systems.
 - [**SQL Chat Server**](servers/sql) - Connect to SQL databases and automatically generate, execute, and optimize queries based on your database schema and natural language input. Enables chat-based data exploration, leveraging external Retrieval-Augmented Generation (RAG) for advanced query assistance.
 - [**External RAG Tool Server**](servers/external-rag) - Connect and execute your own Retrieval-Augmented Generation (RAG) pipelines as callable API tools. Easily integrate custom or third-party RAG flows, providing structured access and modular composition for knowledge-intensive applications.
+- [**Xquik Search Server**](servers/xquik) - Search recent X/Twitter posts through Xquik and return structured tweet data for agents and dashboards.
 
 (More examples and reference implementations will be actively developed and continually updated.)
 

servers/xquik/README.md

@@ -0,0 +1,63 @@
+# Xquik Search Tool Server
+
+A FastAPI-based OpenAPI tool server for searching recent X/Twitter posts through Xquik.
+
+Built with:
+FastAPI, OpenAPI, Python, Xquik
+
+---
+
+## Quickstart
+
+```bash
+git clone https://github.com/open-webui/openapi-servers
+cd openapi-servers/servers/xquik
+
+pip install -r requirements.txt
+
+export XQUIK_API_KEY="xqk_your_api_key"
+
+uvicorn main:app --host 0.0.0.0 --reload
+```
+
+---
+
+## Configuration
+
+- `XQUIK_API_KEY`: Required. Xquik API key used when calling Xquik.
+- `XQUIK_BASE_URL`: Optional. Defaults to `https://xquik.com`.
+
+The server keeps the Xquik API key in the local environment and does not expose it through the public OpenAPI schema.
+
+---
+
+## Endpoints
+
+### `GET /search`
+
+Search recent X/Twitter posts.
+
+Required parameters:
+
+- `q`: Search query, tweet ID, X status URL, or account date window.
+
+Optional parameters:
+
+- `queryType`: `Latest` or `Top`. Defaults to `Latest`.
+- `limit`: Maximum tweets to return. Defaults to `20`, maximum `200`.
+- `cursor`: Pagination cursor from a previous response.
+
+### `GET /health`
+
+Returns whether the server is running and whether `XQUIK_API_KEY` is configured.
+
+---
+
+## API Documentation
+
+Once running, explore the generated docs:
+
+- Swagger UI: http://localhost:8000/docs
+- OpenAPI JSON: http://localhost:8000/openapi.json
+
+Made by the Open WebUI community.

servers/xquik/main.py

@@ -0,0 +1,72 @@
+import os
+from typing import Any, Literal, Optional
+
+import httpx
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException, Query
+from fastapi.middleware.cors import CORSMiddleware
+
+load_dotenv()
+
+XQUIK_BASE_URL = os.getenv("XQUIK_BASE_URL", "https://xquik.com").rstrip("/")
+XQUIK_API_KEY = os.getenv("XQUIK_API_KEY")
+
+app = FastAPI(
+    title="Xquik Search API Server",
+    version="1.0.0",
+    description="Search recent X/Twitter posts through Xquik and expose the results as an OpenAPI tool server.",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/health", summary="Health check")
+def health() -> dict[str, bool]:
+    return {"ok": True, "xquik_configured": bool(XQUIK_API_KEY)}
+
+
+@app.get("/search", summary="Search X/Twitter posts")
+async def search_tweets(
+    q: str = Query(..., description="Search query, X status URL, tweet ID, or account date window"),
+    queryType: Literal["Latest", "Top"] = Query(
+        "Latest",
+        description="Sort order. Use Latest for chronological results or Top for engagement-ranked results.",
+    ),
+    limit: int = Query(20, ge=1, le=200, description="Maximum number of tweets to return"),
+    cursor: Optional[str] = Query(None, description="Pagination cursor from a previous response"),
+) -> dict[str, Any]:
+    if not XQUIK_API_KEY:
+        raise HTTPException(status_code=500, detail="Set XQUIK_API_KEY before calling /search.")
+
+    params: dict[str, str | int] = {
+        "q": q,
+        "queryType": queryType,
+        "limit": limit,
+    }
+    if cursor:
+        params["cursor"] = cursor
+
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(
+                f"{XQUIK_BASE_URL}/api/v1/x/tweets/search",
+                params=params,
+                headers={"x-api-key": XQUIK_API_KEY},
+            )
+    except httpx.RequestError as exc:
+        raise HTTPException(status_code=503, detail=f"Could not reach Xquik: {exc}") from exc
+
+    if response.status_code >= 400:
+        try:
+            detail: Any = response.json()
+        except ValueError:
+            detail = response.text
+        raise HTTPException(status_code=response.status_code, detail=detail)
+
+    return response.json()

servers/xquik/requirements.txt

@@ -0,0 +1,4 @@
+fastapi
+uvicorn[standard]
+httpx
+python-dotenv