feat: add Xquik toolset

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,60 @@
+"""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.",
+  "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,102 @@
+"""Search X posts with Xquik."""
+
+from decimal import Decimal
+from typing import Any, Literal
+
+from langchain_core.tools import ArgsSchema
+from pydantic import BaseModel, Field
+
+from intentkit.tools.xquik.base import XquikBaseTool
+
+
+class XquikSearchTweetsInput(BaseModel):
+    """Input parameters for searching 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 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 | None = None
+    next_cursor: str | None = None
+    raw: dict[str, Any] = Field(default_factory=dict)
+
+
+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 XquikSearchTweetsOutput(
+            tweets=[
+                XquikTweet.model_validate(item)
+                for item in payload.get("tweets", [])
+                if isinstance(item, dict)
+            ],
+            has_next_page=payload.get("has_next_page"),
+            next_cursor=payload.get("next_cursor"),
+            raw=payload,
+        )