diff --git a/README.md b/README.md index f64e623..a5cea15 100644 --- a/README.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index bf6c533..55e82d8 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/mcp_server_toolkit/servers/xquik_search.py b/src/mcp_server_toolkit/servers/xquik_search.py new file mode 100644 index 0000000..bf4e200 --- /dev/null +++ b/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() diff --git a/tests/test_xquik_search.py b/tests/test_xquik_search.py new file mode 100644 index 0000000..6aabadb --- /dev/null +++ b/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 "")