diff --git a/.env.example b/.env.example index 0d06d7f3..86ed3e6e 100644 --- a/.env.example +++ b/.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. diff --git a/intentkit/config/config.py b/intentkit/config/config.py index 4fffceeb..1358b0ce 100644 --- a/intentkit/config/config.py +++ b/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") diff --git a/intentkit/tools/xquik/__init__.py b/intentkit/tools/xquik/__init__.py new file mode 100644 index 00000000..87c5f1d9 --- /dev/null +++ b/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) diff --git a/intentkit/tools/xquik/base.py b/intentkit/tools/xquik/base.py new file mode 100644 index 00000000..18d3dd2b --- /dev/null +++ b/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 diff --git a/intentkit/tools/xquik/search_tweets.py b/intentkit/tools/xquik/search_tweets.py new file mode 100644 index 00000000..af28b8bb --- /dev/null +++ b/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) 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/core/test_origin_provider.py b/tests/core/test_origin_provider.py index 7dad59ba..dba387fd 100644 --- a/tests/core/test_origin_provider.py +++ b/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() diff --git a/tests/tools/test_xquik.py b/tests/tools/test_xquik.py new file mode 100644 index 00000000..83f3883b --- /dev/null +++ b/tests/tools/test_xquik.py @@ -0,0 +1,204 @@ +"""Tests for the Xquik toolset.""" + +import json +from decimal import Decimal +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from langchain_core.tools.base import ToolException +from pydantic import ValidationError + +from intentkit.tools.xquik import available, get_tools, toolset +from intentkit.tools.xquik.search_tweets import ( + XquikSearchTweets, + XquikSearchTweetsInput, + parse_search_response, +) + + +def _response(payload: object, status_code: int = 200) -> httpx.Response: + """Build an HTTP response with a request for raise_for_status().""" + 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_toolset_and_tool_metadata() -> None: + """Expose repository-native catalog metadata and tool fields.""" + tool = XquikSearchTweets() + + assert toolset.title == "Xquik" + assert toolset.tags == ["Search", "Social"] + assert toolset.icon == "/tools/xquik/xquik.svg" + assert "Not affiliated with X Corp." in toolset.description + assert tool.name == "xquik_search_tweets" + assert tool.title == "Search X Posts" + assert tool.price == Decimal("15") + assert tool.category == "xquik" + + +@pytest.mark.parametrize("q", ["", " "]) +def test_input_rejects_empty_queries(q: str) -> None: + """Reject empty and whitespace-only searches before an API call.""" + with pytest.raises(ValidationError): + XquikSearchTweetsInput(q=q) + + +def test_input_applies_defaults_and_bounds() -> None: + """Keep request size bounded while preserving the API sort contract.""" + request = XquikSearchTweetsInput(q=" agent frameworks ") + + assert request.q == "agent frameworks" + assert request.query_type == "Latest" + assert request.limit == 20 + + with pytest.raises(ValidationError): + XquikSearchTweetsInput(q="agents", limit=0) + with pytest.raises(ValidationError): + XquikSearchTweetsInput(q="agents", limit=201) + + +def test_available_reflects_api_key_configuration() -> None: + """Hide the toolset unless its hosted credential is configured.""" + with patch("intentkit.tools.xquik.system_config") as system_config: + system_config.xquik_api_key = "test-key" + assert available() is True + + system_config.xquik_api_key = None + assert available() is False + + +@pytest.mark.asyncio +async def test_get_tools_returns_known_names_only() -> None: + """Resolve requested tools through the current toolset entrypoint.""" + tools = await get_tools(["unknown", "xquik_search_tweets"]) + + assert [tool.name for tool in tools] == ["xquik_search_tweets"] + + +def test_search_response_requires_contract_fields() -> None: + """Reject malformed payloads instead of returning partial results.""" + invalid_payloads = [ + {"tweets": "not-a-list", "has_next_page": False, "next_cursor": ""}, + {"tweets": [], "has_next_page": False}, + {"tweets": [{"text": "missing id"}], "has_next_page": False, "next_cursor": ""}, + ] + + for payload in invalid_payloads: + with pytest.raises(ToolException, match="unexpected response"): + parse_search_response(payload) + + +@pytest.mark.asyncio +async def test_search_calls_current_xquik_contract() -> None: + """Send the documented route, authentication header, and query fields.""" + 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 config: + config.xquik_api_key = "test-key" + with patch("intentkit.tools.xquik.base.httpx.AsyncClient") as client_class: + client = AsyncMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=False) + client.get = AsyncMock(return_value=_response(payload)) + client_class.return_value = 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" + client_class.assert_called_once_with(timeout=30.0) + client.get.assert_awaited_once() + args, kwargs = client.get.await_args + assert args == ("https://xquik.com/api/v1/x/tweets/search",) + assert kwargs["headers"] == { + "accept": "application/json", + "x-api-key": "test-key", + } + assert kwargs["params"] == { + "q": "intentkit", + "queryType": "Top", + "limit": 5, + "cursor": "start", + } + + +@pytest.mark.asyncio +async def test_search_requires_api_key() -> None: + """Fail before networking when the hosted credential is absent.""" + with patch("intentkit.tools.xquik.base.config") as config: + config.xquik_api_key = None + with pytest.raises(ToolException, match="API key is not configured"): + await XquikSearchTweets()._arun(q="intentkit") + + +@pytest.mark.asyncio +async def test_http_errors_are_sanitized() -> None: + """Report status without returning an upstream response body.""" + with patch("intentkit.tools.xquik.base.config") as config: + config.xquik_api_key = "test-key" + with patch("intentkit.tools.xquik.base.httpx.AsyncClient") as client_class: + client = AsyncMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=False) + client.get = AsyncMock(return_value=_response({"secret": "body"}, 402)) + client_class.return_value = client + + with pytest.raises(ToolException, match="returned HTTP 402") as error: + await XquikSearchTweets()._arun(q="intentkit") + + assert "secret" not in str(error.value) + + +@pytest.mark.asyncio +async def test_request_and_json_errors_are_stable() -> None: + """Convert transport and decoding failures into stable tool errors.""" + with patch("intentkit.tools.xquik.base.config") as config: + config.xquik_api_key = "test-key" + with patch("intentkit.tools.xquik.base.httpx.AsyncClient") as client_class: + client = AsyncMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=False) + client_class.return_value = client + + client.get = AsyncMock( + side_effect=httpx.ConnectError( + "connection details", + request=httpx.Request("GET", "https://xquik.com"), + ) + ) + with pytest.raises(ToolException, match="request failed") as error: + await XquikSearchTweets()._arun(q="intentkit") + assert "connection details" not in str(error.value) + + invalid_json = httpx.Response( + status_code=200, + content=b"not-json", + request=httpx.Request( + "GET", "https://xquik.com/api/v1/x/tweets/search" + ), + ) + client.get = AsyncMock(return_value=invalid_json) + with pytest.raises(ToolException, match="invalid JSON"): + await XquikSearchTweets()._arun(q="intentkit")