From db3992e11b05b56ef610db440381f026e4d85cdd Mon Sep 17 00:00:00 2001 From: kriptoburak Date: Thu, 18 Jun 2026 22:12:28 +0300 Subject: [PATCH] feat: add Xquik toolset --- intentkit/config/config.py | 7 +- intentkit/tools/xquik/__init__.py | 54 ++++++++++ intentkit/tools/xquik/base.py | 60 +++++++++++ intentkit/tools/xquik/schema.json | 46 +++++++++ intentkit/tools/xquik/search_tweets.py | 102 +++++++++++++++++++ intentkit/tools/xquik/xquik.svg | 1 + tests/tools/test_xquik.py | 132 +++++++++++++++++++++++++ 7 files changed, 399 insertions(+), 3 deletions(-) create mode 100644 intentkit/tools/xquik/__init__.py create mode 100644 intentkit/tools/xquik/base.py create mode 100644 intentkit/tools/xquik/schema.json create mode 100644 intentkit/tools/xquik/search_tweets.py create mode 100644 intentkit/tools/xquik/xquik.svg create mode 100644 tests/tools/test_xquik.py diff --git a/intentkit/config/config.py b/intentkit/config/config.py index efe6479f..b165380f 100644 --- a/intentkit/config/config.py +++ b/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 diff --git a/intentkit/tools/xquik/__init__.py b/intentkit/tools/xquik/__init__.py new file mode 100644 index 00000000..f77291a7 --- /dev/null +++ b/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) diff --git a/intentkit/tools/xquik/base.py b/intentkit/tools/xquik/base.py new file mode 100644 index 00000000..419e4d12 --- /dev/null +++ b/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 diff --git a/intentkit/tools/xquik/schema.json b/intentkit/tools/xquik/schema.json new file mode 100644 index 00000000..b1f6cc18 --- /dev/null +++ b/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 +} diff --git a/intentkit/tools/xquik/search_tweets.py b/intentkit/tools/xquik/search_tweets.py new file mode 100644 index 00000000..a390524a --- /dev/null +++ b/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, + ) diff --git a/intentkit/tools/xquik/xquik.svg b/intentkit/tools/xquik/xquik.svg new file mode 100644 index 00000000..966e2c61 --- /dev/null +++ b/intentkit/tools/xquik/xquik.svg @@ -0,0 +1 @@ + diff --git a/tests/tools/test_xquik.py b/tests/tools/test_xquik.py new file mode 100644 index 00000000..605b54d6 --- /dev/null +++ b/tests/tools/test_xquik.py @@ -0,0 +1,132 @@ +"""Tests for Xquik tools.""" + +from decimal import Decimal +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from langchain_core.tools.base import ToolException + +from intentkit.tools.xquik import Config, ToolStates, available, get_tools +from intentkit.tools.xquik.search_tweets import ( + XquikSearchTweets, + XquikSearchTweetsInput, +) + + +def _mock_response(payload: dict, status_code: int = 200) -> httpx.Response: + import json + + return httpx.Response( + status_code=status_code, + content=json.dumps(payload).encode(), + headers={"content-type": "application/json"}, + request=httpx.Request("GET", "https://xquik.com/api/v1/x/tweets/search"), + ) + + +def test_tool_metadata(): + """Test tool names, prices, and categories.""" + tool = XquikSearchTweets() + + assert tool.name == "xquik_search_tweets" + assert tool.price == Decimal("15") + assert tool.category == "xquik" + + +def test_search_tweets_input_validation(): + """Validate search input bounds.""" + inp = XquikSearchTweetsInput(q="agent frameworks") + + assert inp.query_type == "Latest" + assert inp.limit == 20 + + with pytest.raises(Exception): + XquikSearchTweetsInput(q="agent frameworks", limit=0) + + with pytest.raises(Exception): + XquikSearchTweetsInput(q="agent frameworks", limit=201) + + +def test_available_with_key(): + """Test available() returns True when API key is set.""" + with patch("intentkit.tools.xquik.system_config") as mock_config: + mock_config.xquik_api_key = "test-key" + assert available() is True + + +def test_available_without_key(): + """Test available() returns False when API key is not set.""" + with patch("intentkit.tools.xquik.system_config") as mock_config: + mock_config.xquik_api_key = "" + assert available() is False + + +@pytest.mark.asyncio +async def test_get_tools_filters_by_state(): + """Test get_tools respects state configuration.""" + config: Config = { + "enabled": True, + "states": ToolStates(xquik_search_tweets="private"), + } + + public_tools = await get_tools(config, is_private=False) + assert public_tools == [] + + private_tools = await get_tools(config, is_private=True) + assert [tool.name for tool in private_tools] == ["xquik_search_tweets"] + + +@pytest.mark.asyncio +async def test_search_tweets_calls_xquik_api(): + """Search uses the public Xquik REST route and API key header.""" + payload = { + "tweets": [ + { + "id": "123", + "text": "IntentKit adds a new data source.", + "author": {"username": "example"}, + } + ], + "has_next_page": True, + "next_cursor": "cursor-1", + } + + with patch("intentkit.tools.xquik.base.config") as mock_config: + mock_config.xquik_api_key = "test-key" + with patch("intentkit.tools.xquik.base.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.get = AsyncMock(return_value=_mock_response(payload)) + mock_client_cls.return_value = mock_client + + output = await XquikSearchTweets()._arun( + q="intentkit", + query_type="Top", + limit=5, + cursor="start", + ) + + assert output.tweets[0].id == "123" + assert output.tweets[0].author is not None + assert output.tweets[0].author.username == "example" + assert output.next_cursor == "cursor-1" + mock_client.get.assert_awaited_once() + _, kwargs = mock_client.get.await_args + assert kwargs["headers"]["x-api-key"] == "test-key" + assert kwargs["params"] == { + "q": "intentkit", + "queryType": "Top", + "limit": 5, + "cursor": "start", + } + + +@pytest.mark.asyncio +async def test_search_tweets_requires_api_key(): + """Search fails when Xquik API key is missing.""" + with patch("intentkit.tools.xquik.base.config") as mock_config: + mock_config.xquik_api_key = "" + with pytest.raises(ToolException, match="Xquik API key"): + await XquikSearchTweets()._arun(q="intentkit")