Add Xquik search MCP server

README.md

@@ -1,6 +1,6 @@
 # mcp-server-toolkit
 
-> Reusable framework for building **Model Context Protocol (MCP) servers** for Claude, plus three production-shaped example servers — filesystem-stats, github-issues, sqlite-query. In-memory test harness with **p99 tool round-trip = 8.2 ms** (target was 50 ms).
+> Reusable framework for building **Model Context Protocol (MCP) servers** for Claude, plus four production-shaped example servers: filesystem-stats, github-issues, sqlite-query, xquik-search. In-memory test harness with **p99 tool round-trip = 8.2 ms** (target was 50 ms).
 
 [![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE) [![Tests](https://img.shields.io/badge/tests-12%20passing-brightgreen)](#tests) [![Python](https://img.shields.io/badge/python-3.10%2B-blue)]()
 
@@ -10,10 +10,11 @@ MCP is Anthropic's standard for letting LLMs call tools running on your machine.
 
 * A thin `ToolkitServer` wrapper with a tool description registry + uniform `ToolError` error-handling pattern
 * An in-memory test harness so your tools are unit-testable via a real `ClientSession` — **no subprocess, no JSON-over-stdio flakiness**
-* **Three example servers** ready to install:
-  * `mcp-filesystem-stats` — sandboxed `list_directory` / `file_summary` / `find_files`
-  * `mcp-github-issues` — read-only public GitHub issue search via REST
-  * `mcp-sqlite-query` — read-only SQLite with write-statement rejection
+* **Four example servers** ready to install:
+  * `mcp-filesystem-stats` - sandboxed `list_directory` / `file_summary` / `find_files`
+  * `mcp-github-issues` - read-only public GitHub issue search via REST
+  * `mcp-sqlite-query` - read-only SQLite with write-statement rejection
+  * `mcp-xquik-search` - X post, user, and trend research via Xquik
 
 ## Hero benchmark
 
@@ -85,7 +86,7 @@ Tool authors:
 * Raise `ToolError("...")` for user-visible failures (the SDK emits `isError: true`).
 * Let any other exception bubble — FastMCP turns it into a generic server error.
 
-## Three example servers
+## Four example servers
 
 ### `filesystem-stats`
 
@@ -117,6 +118,16 @@ Strict read-only SQLite. Only `SELECT` / `WITH ... SELECT` accepted; statement c
 | `describe_table` | `name` | columns + types + PK marker |
 | `query` | `sql` | TSV of up to 100 rows |
 
+### `xquik-search`
+
+Read X posts, users, and trends through Xquik's public REST API. Set `XQUIK_API_KEY` before calling tools.
+
+| Tool | Args | What it returns |
+|---|---|---|
+| `search_tweets` | `query`, `query_type`, `limit`, `cursor` | matching posts plus pagination cursor |
+| `search_users` | `query`, `cursor` | matching profiles plus pagination cursor |
+| `get_trends` | `woeid`, `count` | trending topics for a region |
+
 ## Tests
 
 ```bash
@@ -145,7 +156,7 @@ Each server has an in-memory test suite that exercises:
 │       ├── filesystem_stats.py
 │       ├── github_issues.py
 │       └── sqlite_query.py
-├── tests/                  # 12 pytest-asyncio cases
+├── tests/                  # pytest-asyncio cases
 └── bench/
     ├── latency.py
     └── latency_results.json

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "mcp-server-toolkit"
 version = "0.1.0"
-description = "Reusable Model Context Protocol server framework + 3 example servers (filesystem-stats, github-issues, sqlite-query). In-memory test harness with sub-millisecond tool round-trip."
+description = "Reusable Model Context Protocol server framework + 4 example servers (filesystem-stats, github-issues, sqlite-query, xquik-search). In-memory test harness with sub-millisecond tool round-trip."
 readme = "README.md"
 requires-python = ">=3.10"
 license = { text = "MIT" }
@@ -32,6 +32,7 @@ dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "ruff>=0.6"]
 mcp-filesystem-stats = "mcp_server_toolkit.servers.filesystem_stats:main"
 mcp-github-issues = "mcp_server_toolkit.servers.github_issues:main"
 mcp-sqlite-query = "mcp_server_toolkit.servers.sqlite_query:main"
+mcp-xquik-search = "mcp_server_toolkit.servers.xquik_search:main"
 
 [project.urls]
 Homepage = "https://github.com/Tajaddin/mcp-server-toolkit"

src/mcp_server_toolkit/servers/xquik_search.py

@@ -0,0 +1,189 @@
+"""MCP server: Xquik X search and trends over the public REST API."""
+
+from __future__ import annotations
+
+import argparse
+import os
+from collections.abc import Callable
+from typing import Any
+
+import httpx
+
+from mcp_server_toolkit.base import ToolError, ToolkitServer, register_tool
+
+XQUIK_API = os.environ.get("XQUIK_API_URL", "https://xquik.com").rstrip("/")
+
+ClientFactory = Callable[[], httpx.Client]
+
+
+def _client() -> httpx.Client:
+    api_key = os.environ.get("XQUIK_API_KEY")
+    if not api_key:
+        raise ToolError("Set XQUIK_API_KEY before calling Xquik tools.")
+
+    return httpx.Client(
+        timeout=30.0,
+        headers={
+            "Accept": "application/json",
+            "User-Agent": "mcp-server-toolkit/0.1",
+            "x-api-key": api_key,
+        },
+    )
+
+
+def _clean_params(params: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in params.items() if value is not None}
+
+
+def _request_json(client: httpx.Client, path: str, params: dict[str, Any]) -> dict[str, Any]:
+    response = client.get(f"{XQUIK_API}{path}", params=_clean_params(params))
+    if response.status_code != 200:
+        raise ToolError(f"Xquik returned {response.status_code}: {response.text[:200]}")
+    data = response.json()
+    if not isinstance(data, dict):
+        raise ToolError("Xquik returned an unexpected response shape.")
+    return data
+
+
+def _clamp(value: int, minimum: int, maximum: int) -> int:
+    return max(minimum, min(value, maximum))
+
+
+def _short(value: Any, limit: int = 220) -> str:
+    text = "" if value is None else str(value).strip()
+    if len(text) <= limit:
+        return text
+    return f"{text[: limit - 1]}..."
+
+
+def _format_tweet(tweet: dict[str, Any]) -> str:
+    author = tweet.get("author") if isinstance(tweet.get("author"), dict) else {}
+    username = author.get("username") or "unknown"
+    tweet_id = tweet.get("id") or "unknown"
+    created_at = tweet.get("createdAt") or "unknown time"
+    metrics = [
+        f"likes={tweet[key]}"
+        for key in ("likeCount", "retweetCount", "replyCount", "quoteCount")
+        if tweet.get(key) is not None
+    ]
+    url = f"https://x.com/{username}/status/{tweet_id}" if username != "unknown" else ""
+    parts = [
+        f"{tweet_id} @{username} {created_at}",
+        _short(tweet.get("text")),
+    ]
+    if metrics:
+        parts.append(" ".join(metrics))
+    if url:
+        parts.append(url)
+    return "\n".join(part for part in parts if part)
+
+
+def _format_user(user: dict[str, Any]) -> str:
+    username = user.get("username") or "unknown"
+    name = user.get("name") or ""
+    verified = " verified" if user.get("verified") is True else ""
+    followers = f" followers={user['followersCount']}" if user.get("followersCount") is not None else ""
+    return f"@{username} {name}{verified}{followers}".strip()
+
+
+def _format_trend(trend: dict[str, Any]) -> str:
+    rank = trend.get("rank") or "?"
+    name = trend.get("name") or "(unnamed)"
+    description = f" - {trend['description']}" if trend.get("description") else ""
+    query = f" query={trend['query']}" if trend.get("query") else ""
+    return f"{rank}. {name}{description}{query}"
+
+
+def build_server(client_factory: ClientFactory = _client) -> ToolkitServer:
+    server = ToolkitServer(
+        name="xquik-search",
+        description="Read X posts, users, and trends through Xquik's public REST API.",
+    )
+
+    @register_tool(
+        server,
+        name="search_tweets",
+        description="Search X posts with Xquik. Set XQUIK_API_KEY before calling.",
+    )
+    def search_tweets(
+        query: str,
+        query_type: str = "Latest",
+        limit: int = 10,
+        cursor: str | None = None,
+    ) -> str:
+        if not query.strip():
+            raise ToolError("query must be non-empty")
+        if query_type not in {"Latest", "Top"}:
+            raise ToolError("query_type must be Latest or Top")
+
+        with client_factory() as client:
+            data = _request_json(
+                client,
+                "/api/v1/x/tweets/search",
+                {
+                    "q": query,
+                    "queryType": query_type,
+                    "limit": _clamp(limit, 1, 50),
+                    "cursor": cursor,
+                },
+            )
+        tweets = data.get("tweets") or []
+        if not tweets:
+            return "(no tweets)"
+        lines = [_format_tweet(tweet) for tweet in tweets[:50] if isinstance(tweet, dict)]
+        if data.get("has_next_page") and data.get("next_cursor"):
+            lines.append(f"next_cursor={data['next_cursor']}")
+        return "\n\n".join(lines)
+
+    @register_tool(
+        server,
+        name="search_users",
+        description="Search X users with Xquik. Set XQUIK_API_KEY before calling.",
+    )
+    def search_users(query: str, cursor: str | None = None) -> str:
+        if not query.strip():
+            raise ToolError("query must be non-empty")
+
+        with client_factory() as client:
+            data = _request_json(
+                client,
+                "/api/v1/x/users/search",
+                {"q": query, "cursor": cursor},
+            )
+        users = data.get("users") or []
+        if not users:
+            return "(no users)"
+        lines = [_format_user(user) for user in users if isinstance(user, dict)]
+        if data.get("has_next_page") and data.get("next_cursor"):
+            lines.append(f"next_cursor={data['next_cursor']}")
+        return "\n".join(lines)
+
+    @register_tool(
+        server,
+        name="get_trends",
+        description="Get X trends by WOEID with Xquik. Set XQUIK_API_KEY before calling.",
+    )
+    def get_trends(woeid: int = 1, count: int = 10) -> str:
+        with client_factory() as client:
+            data = _request_json(
+                client,
+                "/api/v1/x/trends",
+                {"woeid": woeid, "count": _clamp(count, 1, 50)},
+            )
+        trends = data.get("trends") or []
+        if not trends:
+            return "(no trends)"
+        return "\n".join(_format_trend(trend) for trend in trends if isinstance(trend, dict))
+
+    return server
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="MCP server: xquik-search")
+    parser.parse_args()
+    server = build_server()
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

tests/test_xquik_search.py

@@ -0,0 +1,117 @@
+"""xquik-search server tests via mocked HTTP."""
+
+from __future__ import annotations
+
+import httpx
+
+from mcp_server_toolkit import call_tool, in_memory_session, list_tools
+from mcp_server_toolkit.servers.xquik_search import build_server
+
+
+def _factory(handler):
+    def create_client() -> httpx.Client:
+        return httpx.Client(transport=httpx.MockTransport(handler))
+
+    return create_client
+
+
+async def test_list_tools_returns_three() -> None:
+    server = build_server(_factory(lambda request: httpx.Response(500)))
+    async with in_memory_session(server) as session:
+        tools = await list_tools(session)
+        names = {tool.name for tool in tools}
+        assert names == {"search_tweets", "search_users", "get_trends"}
+
+
+async def test_search_tweets_formats_results() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/api/v1/x/tweets/search"
+        assert request.url.params["q"] == "mcp"
+        return httpx.Response(
+            200,
+            json={
+                "tweets": [
+                    {
+                        "id": "123",
+                        "text": "MCP example",
+                        "createdAt": "2026-06-28T00:00:00Z",
+                        "likeCount": 4,
+                        "author": {"username": "xquik"},
+                    }
+                ],
+                "has_next_page": True,
+                "next_cursor": "cursor-1",
+            },
+        )
+
+    server = build_server(_factory(handler))
+    async with in_memory_session(server) as session:
+        out = await call_tool(session, "search_tweets", query="mcp", limit=5)
+        text = out.content[0].text
+        assert "123 @xquik" in text
+        assert "MCP example" in text
+        assert "next_cursor=cursor-1" in text
+
+
+async def test_search_users_formats_results() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/api/v1/x/users/search"
+        assert request.url.params["q"] == "xquik"
+        return httpx.Response(
+            200,
+            json={
+                "users": [
+                    {
+                        "username": "xquik",
+                        "name": "Xquik",
+                        "verified": True,
+                        "followersCount": 100,
+                    }
+                ],
+                "has_next_page": False,
+                "next_cursor": None,
+            },
+        )
+
+    server = build_server(_factory(handler))
+    async with in_memory_session(server) as session:
+        out = await call_tool(session, "search_users", query="xquik")
+        assert "@xquik Xquik verified followers=100" in out.content[0].text
+
+
+async def test_get_trends_formats_results() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/api/v1/x/trends"
+        assert request.url.params["woeid"] == "1"
+        return httpx.Response(
+            200,
+            json={
+                "trends": [
+                    {
+                        "rank": 1,
+                        "name": "#AI",
+                        "description": "Artificial intelligence discussions",
+                        "query": "%23AI",
+                    }
+                ],
+                "count": 1,
+                "woeid": 1,
+            },
+        )
+
+    server = build_server(_factory(handler))
+    async with in_memory_session(server) as session:
+        out = await call_tool(session, "get_trends", count=1)
+        text = out.content[0].text
+        assert "1. #AI" in text
+        assert "query=%23AI" in text
+
+
+async def test_tool_errors_are_user_visible() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(401, json={"error": "unauthenticated"})
+
+    server = build_server(_factory(handler))
+    async with in_memory_session(server) as session:
+        out = await call_tool(session, "search_tweets", query="mcp")
+        assert out.isError is True or "Xquik returned 401" in (out.content[0].text or "")