feat: add native Xquik public search toolset

.env.example

@@ -207,6 +207,7 @@ CRYPTOCOMPARE_API_KEY=
 VENICE_API_KEY=
 CARV_API_KEY=
 OPENSEA_API_KEY=
+XQUIK_API_KEY=
 
 # Composio (team links: external app accounts + MCP tools for the lead agent).
 # Leave empty to disable the Links feature entirely.

intentkit/config/config.py

@@ -256,6 +256,7 @@ def __init__(self) -> None:
         self.venice_api_key: str | None = self.load("VENICE_API_KEY")
         self.coingecko_api_key: str | None = self.load("COINGECKO_API_KEY")
         self.opensea_api_key: str | None = self.load("OPENSEA_API_KEY")
+        self.xquik_api_key: str | None = self.load("XQUIK_API_KEY")
         # Composio — team links (external app accounts + hosted MCP tools).
         # The feature is disabled entirely when the API key is unset.
         self.composio_api_key: str | None = self.load("COMPOSIO_API_KEY")

intentkit/tools/xquik/__init__.py

@@ -0,0 +1,52 @@
+"""Xquik tools for public X research."""
+
+import logging
+from collections.abc import Callable
+
+from intentkit.config.config import config as system_config
+from intentkit.tools.meta import ToolsetMeta
+from intentkit.tools.xquik.base import XquikBaseTool
+from intentkit.tools.xquik.search_tweets import XquikSearchTweets
+
+toolset = ToolsetMeta(
+    title="Xquik",
+    description=(
+        "Search public X posts through Xquik. Xquik is an independent third-party "
+        'service. Not affiliated with X Corp. "Twitter" and "X" are trademarks '
+        "of X Corp."
+    ),
+    tags=["Search", "Social"],
+    icon="/tools/xquik/xquik.svg",
+)
+
+logger = logging.getLogger(__name__)
+
+_cache: dict[str, XquikBaseTool] = {}
+
+_TOOL_CLASSES: dict[str, Callable[[], XquikBaseTool]] = {
+    "xquik_search_tweets": XquikSearchTweets,
+}
+
+
+async def get_tools(tool_names: list[str], **_) -> list[XquikBaseTool]:
+    """Return requested Xquik tools and skip unknown names."""
+    return [tool for name in tool_names if (tool := get_xquik_tool(name))]
+
+
+def get_xquik_tool(tool_name: str) -> XquikBaseTool | None:
+    """Get a cached Xquik tool by name."""
+    if tool_name in _cache:
+        return _cache[tool_name]
+
+    tool_class = _TOOL_CLASSES.get(tool_name)
+    if tool_class is None:
+        logger.warning("Unknown Xquik tool: %s", tool_name)
+        return None
+
+    _cache[tool_name] = tool_class()
+    return _cache[tool_name]
+
+
+def available() -> bool:
+    """Check whether the hosted Xquik credential is configured."""
+    return bool(system_config.xquik_api_key)

intentkit/tools/xquik/base.py

@@ -0,0 +1,59 @@
+"""Shared HTTP client behavior for Xquik tools."""
+
+from typing import Any
+
+import httpx
+from langchain_core.tools.base import ToolException
+
+from intentkit.config.config import config
+from intentkit.tools.base import IntentKitTool
+
+XQUIK_BASE_URL = "https://xquik.com/api/v1"
+
+
+class XquikBaseTool(IntentKitTool):
+    """Base class for authenticated Xquik reads."""
+
+    category: str = "xquik"
+
+    def get_api_key(self) -> str:
+        """Return the configured Xquik API key."""
+        if not config.xquik_api_key:
+            raise ToolException("Xquik API key is not configured")
+        return config.xquik_api_key
+
+    async def get(
+        self,
+        path: str,
+        params: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Call one authenticated Xquik GET endpoint."""
+        headers = {
+            "accept": "application/json",
+            "x-api-key": self.get_api_key(),
+        }
+        clean_params = {
+            key: value for key, value in params.items() if value is not None
+        }
+
+        try:
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.get(
+                    f"{XQUIK_BASE_URL}{path}",
+                    headers=headers,
+                    params=clean_params,
+                )
+                response.raise_for_status()
+                payload = response.json()
+        except httpx.HTTPStatusError as exc:
+            raise ToolException(
+                f"Xquik API returned HTTP {exc.response.status_code}"
+            ) from exc
+        except httpx.RequestError as exc:
+            raise ToolException("Xquik API request failed") from exc
+        except ValueError as exc:
+            raise ToolException("Xquik API returned invalid JSON") from exc
+
+        if not isinstance(payload, dict):
+            raise ToolException("Xquik API returned an unexpected response")
+        return payload

intentkit/tools/xquik/search_tweets.py

@@ -0,0 +1,117 @@
+"""Search public X posts through Xquik."""
+
+from decimal import Decimal
+from typing import Any, Literal
+
+from langchain_core.tools import ArgsSchema
+from langchain_core.tools.base import ToolException
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
+
+from intentkit.tools.xquik.base import XquikBaseTool
+
+
+class XquikSearchTweetsInput(BaseModel):
+    """Input parameters for searching public X posts."""
+
+    q: str = Field(description="Search query, X status URL, or post ID")
+    query_type: Literal["Latest", "Top"] = Field(
+        default="Latest",
+        description="Sort order for results",
+    )
+    limit: int = Field(
+        default=20,
+        ge=1,
+        le=200,
+        description="Maximum posts to return (1-200)",
+    )
+    cursor: str | None = Field(
+        default=None,
+        description="Pagination cursor returned by a previous search",
+    )
+
+    @field_validator("q")
+    @classmethod
+    def validate_query(cls, value: str) -> str:
+        """Reject empty and whitespace-only searches."""
+        value = value.strip()
+        if not value:
+            raise ValueError("Search query must not be empty")
+        return value
+
+
+class XquikTweetAuthor(BaseModel):
+    """Author fields returned by Xquik."""
+
+    model_config = ConfigDict(extra="allow")
+
+    id: str | None = None
+    username: str | None = None
+    name: str | None = None
+    verified: bool | None = None
+
+
+class XquikTweet(BaseModel):
+    """Public post fields returned by Xquik."""
+
+    model_config = ConfigDict(extra="allow")
+
+    id: str
+    text: str | None = None
+    createdAt: str | None = None
+    likeCount: int | None = None
+    retweetCount: int | None = None
+    replyCount: int | None = None
+    quoteCount: int | None = None
+    viewCount: int | None = None
+    bookmarkCount: int | None = None
+    author: XquikTweetAuthor | None = None
+
+
+class XquikSearchTweetsOutput(BaseModel):
+    """Validated Xquik search response."""
+
+    model_config = ConfigDict(extra="allow")
+
+    tweets: list[XquikTweet] = Field(default_factory=list)
+    has_next_page: bool
+    next_cursor: str
+
+
+def parse_search_response(payload: dict[str, Any]) -> XquikSearchTweetsOutput:
+    """Validate required pagination and post fields."""
+    try:
+        return XquikSearchTweetsOutput.model_validate(payload)
+    except ValidationError as exc:
+        raise ToolException("Xquik API returned an unexpected response") from exc
+
+
+class XquikSearchTweets(XquikBaseTool):
+    """Search public X posts by query, post ID, or status URL."""
+
+    name: str = "xquik_search_tweets"
+    title: str = "Search X Posts"
+    description: str = (
+        "Search public X posts by query, post ID, or status URL through Xquik."
+    )
+    price: Decimal = Decimal("15")
+    args_schema: ArgsSchema | None = XquikSearchTweetsInput
+
+    async def _arun(
+        self,
+        q: str,
+        query_type: Literal["Latest", "Top"] = "Latest",
+        limit: int = 20,
+        cursor: str | None = None,
+        **_,
+    ) -> XquikSearchTweetsOutput:
+        """Run the Xquik search request."""
+        payload = await self.get(
+            "/x/tweets/search",
+            params={
+                "q": q,
+                "queryType": query_type,
+                "limit": limit,
+                "cursor": cursor,
+            },
+        )
+        return parse_search_response(payload)

tests/core/test_origin_provider.py

@@ -83,6 +83,7 @@ async def fake_get(model_id: str) -> LLMModelInfo:
         return info
 
     monkeypatch.setattr(LLMModelInfo, "get", staticmethod(fake_get))
+    monkeypatch.setattr("intentkit.models.llm.config.openrouter_api_key", "or-test-key")
 
     llm = OpenRouterLLM(model_name=info.id, info=info)
     instance = await llm.create_instance()
@@ -103,6 +104,7 @@ async def fake_get(model_id: str) -> LLMModelInfo:
         return info
 
     monkeypatch.setattr(LLMModelInfo, "get", staticmethod(fake_get))
+    monkeypatch.setattr("intentkit.models.llm.config.openrouter_api_key", "or-test-key")
 
     llm = OpenRouterLLM(model_name=info.id, info=info)
     instance = await llm.create_instance()