feat: add a validated Xquik search toolset

.env.example

@@ -119,7 +119,7 @@ FIRECRAWL_API_KEY=
 #LANGSMITH_API_KEY=
 #LANGSMITH_PROJECT=intentkit
 # Langfuse (set both keys to enable; disables LangSmith). LANGFUSE_BASE_URL is
-# optional — defaults to Langfuse Cloud; set it for self-hosted. (LANGFUSE_HOST
+# optional - defaults to Langfuse Cloud; set it for self-hosted. (LANGFUSE_HOST
 # is accepted as a legacy alias.)
 #LANGFUSE_PUBLIC_KEY=
 #LANGFUSE_SECRET_KEY=
@@ -177,3 +177,4 @@ CRYPTOCOMPARE_API_KEY=
 VENICE_API_KEY=
 CARV_API_KEY=
 OPENSEA_API_KEY=
+XQUIK_API_KEY=

intentkit/config/config.py

@@ -236,6 +236,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")
         # Cloudflare Browser Rendering
         self.cloudflare_account_id: str | None = self.load("CLOUDFLARE_ACCOUNT_ID")
         self.cloudflare_api_token: str | None = self.load("CLOUDFLARE_API_TOKEN")
@@ -248,7 +249,7 @@ def __init__(self) -> None:
         self.langfuse_public_key: str | None = self.load("LANGFUSE_PUBLIC_KEY")
         self.langfuse_secret_key: str | None = self.load("LANGFUSE_SECRET_KEY")
         # LANGFUSE_BASE_URL is the canonical name in the SDK; LANGFUSE_HOST is
-        # its legacy alias, kept as a fallback. Optional — defaults to cloud.
+        # its legacy alias, kept as a fallback. Optional - defaults to cloud.
         self.langfuse_base_url: str | None = self.load(
             "LANGFUSE_BASE_URL"
         ) or self.load("LANGFUSE_HOST")
@@ -323,7 +324,7 @@ def _setup_langfuse(self) -> None:
 
         Unlike LangSmith (env-var driven), Langfuse needs a callback handler;
         ``intentkit.config.tracing`` registers it through LangChain's global
-        configure hook. Langfuse and LangSmith are mutually exclusive — see the
+        configure hook. Langfuse and LangSmith are mutually exclusive - see the
         tracing-backend block in ``__init__``.
         """
         if not self.langfuse_tracing:
@@ -395,7 +396,7 @@ def load(self, key: str, default: str | None = None) -> str | None:
         if value:
             value = value.replace("\\n", "\n")
         # Strip one pair of matching surrounding quotes from either source
-        # (process env or AWS secret) — docker `environment:` blocks and
+        # (process env or AWS secret) - docker `environment:` blocks and
         # pasted secret values carry quotes through literally.
         if (
             value

intentkit/tools/xquik/__init__.py

@@ -0,0 +1,54 @@
+"""Xquik tools."""
+
+import logging
+from typing import TypedDict
+
+from intentkit.config.config import config as system_config
+from intentkit.tools.base import ToolsetConfig, ToolState, filter_enabled_tool_names
+from intentkit.tools.xquik.base import XquikBaseTool
+from intentkit.tools.xquik.search_tweets import XquikSearchTweets
+
+_cache: dict[str, XquikBaseTool] = {}
+
+logger = logging.getLogger(__name__)
+
+
+class ToolStates(TypedDict):
+    """Type definition for Xquik tool states."""
+
+    xquik_search_tweets: ToolState
+
+
+class Config(ToolsetConfig):
+    """Configuration for Xquik tools."""
+
+    states: ToolStates
+
+
+async def get_tools(
+    config: Config,
+    is_private: bool,
+    **_,
+) -> list[XquikBaseTool]:
+    """Get enabled Xquik tools."""
+    return [
+        tool
+        for name in filter_enabled_tool_names(config["states"], is_private)
+        if (tool := get_xquik_tool(name)) is not None
+    ]
+
+
+def get_xquik_tool(name: str) -> XquikBaseTool | None:
+    """Get a Xquik tool by name."""
+    if name == "xquik_search_tweets":
+        if name not in _cache:
+            _cache[name] = XquikSearchTweets()
+        return _cache[name]
+
+    logger.warning("Unknown Xquik tool: %s", name)
+    return None
+
+
+def available() -> bool:
+    """Check if this toolset is available based on system config."""
+    return bool(system_config.xquik_api_key)

intentkit/tools/xquik/base.py

@@ -0,0 +1,62 @@
+"""Base helpers 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"
+
+
+class XquikBaseTool(IntentKitTool):
+    """Base class for Xquik tools."""
+
+    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],
+        timeout: int = 30,
+    ) -> dict[str, Any]:
+        """Make an authenticated GET request to the Xquik API."""
+        api_key = self.get_api_key()
+        headers = {
+            "accept": "application/json",
+            "x-api-key": api_key,
+        }
+        clean_params = {
+            key: value for key, value in params.items() if value is not None
+        }
+
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.get(
+                    f"{XQUIK_BASE_URL}{path}",
+                    headers=headers,
+                    params=clean_params,
+                    timeout=timeout,
+                )
+                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(f"Xquik API request failed: {exc}") 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/schema.json

@@ -0,0 +1,46 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "title": "Xquik",
+  "description": "Search X posts with Xquik using an API key. Xquik is an independent third-party service. Not affiliated with X Corp. \"Twitter\" and \"X\" are trademarks of X Corp.",
+  "x-icon": "/tools/xquik/xquik.svg",
+  "x-tags": [
+    "Search",
+    "Social"
+  ],
+  "properties": {
+    "enabled": {
+      "type": "boolean",
+      "title": "Enabled",
+      "description": "Whether this tool is enabled",
+      "default": false
+    },
+    "states": {
+      "type": "object",
+      "properties": {
+        "xquik_search_tweets": {
+          "type": "string",
+          "title": "Search X Posts",
+          "enum": [
+            "disabled",
+            "public",
+            "private"
+          ],
+          "x-enum-title": [
+            "Disabled",
+            "Agent Owner + All Users",
+            "Agent Owner Only"
+          ],
+          "description": "Search X posts by query, post ID, or X status URL using the Xquik API.",
+          "default": "disabled"
+        }
+      },
+      "description": "States for each Xquik tool (disabled, public, or private)"
+    }
+  },
+  "required": [
+    "states",
+    "enabled"
+  ],
+  "additionalProperties": true
+}

intentkit/tools/xquik/search_tweets.py

@@ -0,0 +1,119 @@
+"""Search X posts with 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, Field, ValidationError
+
+from intentkit.tools.xquik.base import XquikBaseTool
+
+
+class XquikSearchTweetsInput(BaseModel):
+    """Input parameters for searching X posts."""
+
+    q: str = Field(
+        min_length=1,
+        description="Search query, X status URL, or post ID",
+    )
+    query_type: Literal["Latest", "Top"] = Field(
+        default="Latest",
+        description="Sort order for the search results",
+    )
+    limit: int = Field(
+        default=20,
+        ge=1,
+        le=200,
+        description="Maximum number of posts to return",
+    )
+    cursor: str | None = Field(
+        default=None,
+        description="Pagination cursor from a previous Xquik response",
+    )
+
+
+class XquikTweetAuthor(BaseModel):
+    """Author fields returned by Xquik."""
+
+    id: str | None = None
+    username: str | None = None
+    name: str | None = None
+    verified: bool | None = None
+
+
+class XquikTweet(BaseModel):
+    """Post fields returned by Xquik."""
+
+    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):
+    """Output structure for Xquik search results."""
+
+    tweets: list[XquikTweet] = Field(default_factory=list)
+    has_next_page: bool
+    next_cursor: str
+    raw: dict[str, Any] = Field(default_factory=dict)
+
+
+def parse_search_response(payload: dict[str, Any]) -> XquikSearchTweetsOutput:
+    """Validate and convert a Xquik search response."""
+    tweets = payload.get("tweets")
+    has_next_page = payload.get("has_next_page")
+    next_cursor = payload.get("next_cursor")
+    if (
+        not isinstance(tweets, list)
+        or not all(isinstance(item, dict) for item in tweets)
+        or not isinstance(has_next_page, bool)
+        or not isinstance(next_cursor, str)
+    ):
+        raise ToolException("Xquik API returned an unexpected response")
+
+    try:
+        return XquikSearchTweetsOutput(
+            tweets=[XquikTweet.model_validate(item) for item in tweets],
+            has_next_page=has_next_page,
+            next_cursor=next_cursor,
+            raw=payload,
+        )
+    except ValidationError as exc:
+        raise ToolException("Xquik API returned an unexpected response") from exc
+
+
+class XquikSearchTweets(XquikBaseTool):
+    """Search public X posts with Xquik."""
+
+    name: str = "xquik_search_tweets"
+    description: str = "Search X posts by query, post ID, or X status URL using 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:
+        """Execute the Xquik search request."""
+        payload = await self.get(
+            "/api/v1/x/tweets/search",
+            params={
+                "q": q,
+                "queryType": query_type,
+                "limit": limit,
+                "cursor": cursor,
+            },
+        )
+        return parse_search_response(payload)

tests/core/test_origin_provider.py

@@ -80,7 +80,7 @@ async def fake_get(model_id: str) -> LLMModelInfo:
     monkeypatch.setattr(LLMModelInfo, "get", staticmethod(fake_get))
 
     llm = OpenRouterLLM(model_name=info.id, info=info)
-    instance = await llm.create_instance()
+    instance = await llm.create_instance({"api_key": "or-test-key"})
 
     assert isinstance(instance, ChatOpenRouter)
     assert instance.openrouter_provider == {
@@ -100,7 +100,7 @@ async def fake_get(model_id: str) -> LLMModelInfo:
     monkeypatch.setattr(LLMModelInfo, "get", staticmethod(fake_get))
 
     llm = OpenRouterLLM(model_name=info.id, info=info)
-    instance = await llm.create_instance()
+    instance = await llm.create_instance({"api_key": "or-test-key"})
 
     assert isinstance(instance, ChatOpenRouter)
     assert instance.openrouter_provider is None

tests/models/test_llm_cost.py

@@ -7,7 +7,11 @@
 from intentkit.models.llm import LLMModelInfo, _resolve_generation_cost
 
 
-def _info(input_price="0.5", cached="0.05", output="3"):
+def _info(
+    input_price: str = "0.5",
+    cached: str | None = "0.05",
+    output: str = "3",
+) -> LLMModelInfo:
     return LLMModelInfo.model_construct(
         input_price=Decimal(input_price),
         cached_input_price=Decimal(cached) if cached is not None else None,

tests/tools/test_all_toolsets.py

@@ -14,9 +14,6 @@
 
 TOOLS_DIR = Path(__file__).resolve().parents[2] / "intentkit" / "tools"
 
-# MCP-driven toolsets expose their tools dynamically (no module-level get_tools).
-NO_GET_TOOLS = {"mcp_coingecko"}
-
 
 def _toolset_names():
     # Mirror the tool_registry contract: a toolset category is a directory
@@ -39,8 +36,6 @@ def test_toolsets_discovered():
 @pytest.mark.parametrize("name", TOOLSETS)
 def test_toolset_imports_and_exposes_get_tools(name):
     mod = importlib.import_module(f"intentkit.tools.{name}")
-    if name in NO_GET_TOOLS:
-        return
     assert hasattr(mod, "get_tools"), (
         f"intentkit.tools.{name} is missing get_tools "
         f"(the get_skills->get_tools rename must reach every plugin)"