Compare commits

...
3 Commits
Author SHA1 Message Date
MervinPraison 0237c6f4ce Release v0.2.45 2026-10-08 05:19:04 +00:00
aa8e7079f3 feat: add Keenable keyless web search tool (#105)
* feat: add Keenable keyless web search tool

Keenable is a web-search service (api.keenable.ai). Per AGENTS.md routing,
vendor search services belong in PraisonAI-Tools as standalone tool modules
(mirroring tavily_tool.py / brave_search_tool.py), not inline in the core
praisonaiagents web_search.py aggregator.

Keyless by default via the public endpoint (stdlib urllib, no install);
KEENABLE_API_KEY switches to the authenticated endpoint for higher limits.
Redirects are refused so X-API-Key is never resent to another host; results
are bounded (50 max, 1 MB body cap, title/URL/snippet length caps).

Auto-discovered by praisonai_tools.tools (no manual registration). 13 unit
tests cover endpoint selection, result mapping/bounds, and error handling.

Relates to MervinPraison/PraisonAI#5671

Co-authored-by: Mervin Praison <[email protected]>

* test: cover Keenable redirect refusal and bound limits

Address Greptile P2: the protective limits had no focused coverage and
build_opener was fully mocked, so _NoRedirect, the result cap, the
oversized title/url/snippet caps, and the response read limit could be
removed with the suite staying green.

- Lift _NoRedirect to module level so the redirect refusal is importable
  and unit-testable (was a closure inside search()).
- _FakeResponse.read now honours the requested byte count, so the
  oversized-response test exercises the actual read cap.
- Add TestProtections: result-count cap in the request body, title and
  snippet truncation, oversized-URL skip, read-limit enforcement, and
  _NoRedirect dropping a 302 (no follow-up request).
- Add the missing __init__ docstring.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

---------

Co-authored-by: MervinPraison <[email protected]>
Co-authored-by: Mervin Praison <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
2026-10-08 06:18:03 +01:00
Mervin PraisonandCursor 4e6a092548 chore: bump github-automation-template hub_ref to v1.1.11 (#103)
Picks up the 403 label fix in the hub callable workflows.

Co-authored-by: Cursor <[email protected]>
2026-09-30 08:38:13 +01:00
11 changed files with 466 additions and 18 deletions

No files matched your search

+1 -1
View File
@@ -3,7 +3,7 @@
"stack": "claude",
"profile": "tools",
"mode": "hub",
"hubRef": "v1.1.10",
"hubRef": "v1.1.11",
"templateVersion": "1.1.6",
"installedAt": "2026-09-07T12:54:25Z"
}
+2 -2
View File
@@ -27,7 +27,7 @@ jobs:
permissions:
issues: write
pull-requests: write
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]0
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]1
secrets: inherit
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
+2 -2
View File
@@ -16,8 +16,8 @@ permissions:
jobs:
recover:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]0
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]1
secrets: inherit
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
pr_number: ${{ github.event.inputs.pr_number || '' }}
+2 -2
View File
@@ -25,8 +25,8 @@ permissions:
jobs:
detect-and-trigger:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]0
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]1
secrets: inherit
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
pr_number: ${{ github.event.inputs.pr_number || '' }}
+4 -4
View File
@@ -43,9 +43,9 @@ jobs:
id-token: write
steps:
- uses: actions/checkout@v4
- uses: MervinPraison/github-automation-template/.github/actions/[email protected]0
- uses: MervinPraison/github-automation-template/.github/actions/[email protected]1
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
- name: Generate GitHub App token
id: app-token
uses: actions/create-github-app-token@v1
@@ -87,9 +87,9 @@ jobs:
issues: write
actions: read
id-token: write
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]0
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]1
secrets: inherit
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
pr_number: ${{ github.event.inputs.pr_number || github.event.client_payload.pr_number || '' }}
merge_method: ${{ github.event.inputs.merge_method || 'merge' }}
+2 -2
View File
@@ -18,7 +18,7 @@ jobs:
issues: write
actions: read
id-token: write
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]0
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]1
secrets: inherit
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
+2 -2
View File
@@ -25,9 +25,9 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: MervinPraison/github-automation-template/.github/actions/[email protected]0
- uses: MervinPraison/github-automation-template/.github/actions/[email protected]1
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
- name: Detect conflicting PRs and trigger Claude
uses: actions/github-script@v7
with:
+2 -2
View File
@@ -13,7 +13,7 @@ permissions:
jobs:
sync:
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]0
uses: MervinPraison/github-automation-template/.github/workflows/[email protected]1
secrets: inherit
with:
hub_ref: v1.1.10
hub_ref: v1.1.11
@@ -0,0 +1,175 @@
"""Keenable Search Tool for PraisonAI Agents.
Keyless (or keyed) web search using Keenable.
Keenable works with no API key and no extra install: it calls the keyless
public endpoint with the Python standard library (``urllib``). Setting
``KEENABLE_API_KEY`` switches to the authenticated endpoint for higher limits.
Usage:
from praisonai_tools import KeenableSearchTool
keenable = KeenableSearchTool()
results = keenable.search("AI news 2024")
Environment Variables:
KEENABLE_API_KEY: Optional. When set, the authenticated endpoint is used.
"""
import json
import logging
import os
from typing import Any, Dict, List, Optional, Union
from urllib.request import HTTPRedirectHandler
from praisonai_tools.tools.base import BaseTool
logger = logging.getLogger(__name__)
KEENABLE_SEARCH_URL = "https://api.keenable.ai/v1/search"
KEENABLE_PUBLIC_SEARCH_URL = "https://api.keenable.ai/v1/search/public"
KEENABLE_APP_TITLE = "PraisonAI"
KEENABLE_SEARCH_TIMEOUT_SECONDS = 30
KEENABLE_MAX_RESPONSE_BYTES = 1_000_000
KEENABLE_MAX_RESULTS = 50
KEENABLE_MAX_TITLE_CHARS = 256
KEENABLE_MAX_URL_CHARS = 2048
KEENABLE_MAX_SNIPPET_CHARS = 1200
def _truncate_text(value: Any, limit: int) -> str:
"""Trim whitespace and cap ``value`` to ``limit`` characters."""
text = str(value or "").strip()
if len(text) > limit:
return text[:limit - 1] + "\u2026"
return text
class _NoRedirect(HTTPRedirectHandler):
"""Redirect handler that refuses to follow any redirect.
Keenable never redirects. Following a 3xx would resend ``X-API-Key`` to
whatever host it points at, so the redirect is dropped (urllib then raises
the 3xx as an error, surfaced to the caller) and no second request is sent.
"""
def redirect_request(self, *args, **kwargs):
"""Return ``None`` so urllib does not issue a follow-up request."""
return None
class KeenableSearchTool(BaseTool):
"""Tool for keyless (or keyed) web search using Keenable."""
name = "keenable_search"
description = "Search the web using Keenable (keyless; an API key raises limits)."
def __init__(self, api_key: Optional[str] = None):
"""Create the tool, resolving the API key from the argument or env.
Args:
api_key: Optional Keenable API key. Falls back to the
``KEENABLE_API_KEY`` environment variable; keyless if unset.
"""
self.api_key = (api_key or os.getenv("KEENABLE_API_KEY") or "").strip()
super().__init__()
def run(
self,
action: str = "search",
query: Optional[str] = None,
max_results: int = 5,
**kwargs,
) -> Union[str, Dict[str, Any], List[Dict[str, Any]]]:
"""Dispatch a Keenable action; only ``search`` is supported."""
action = action.lower().replace("-", "_")
if action == "search":
return self.search(query=query, max_results=max_results)
return {"error": f"Unknown action: {action}"}
def search(self, query: str, max_results: int = 5) -> List[Dict[str, Any]]:
"""Search the web with Keenable.
Args:
query: Search query string.
max_results: Maximum results to return (capped at 50).
Returns:
List of dicts with ``title``, ``url``, ``snippet`` and ``provider``,
or a single-item list with an ``error`` key on failure.
"""
if not query:
return [{"error": "query is required"}]
result_limit = max(0, min(int(max_results), KEENABLE_MAX_RESULTS))
if result_limit == 0:
return []
from urllib.error import URLError
from urllib.request import Request, build_opener
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "praisonai-tools",
# Required on keyless calls; names the app, carries no user identifier.
"X-Keenable-Title": KEENABLE_APP_TITLE,
}
if self.api_key:
headers["X-API-Key"] = self.api_key
body = json.dumps({
"query": query,
"max_results": result_limit,
"snippet_max_length": KEENABLE_MAX_SNIPPET_CHARS,
}).encode("utf-8")
request = Request(
KEENABLE_SEARCH_URL if self.api_key else KEENABLE_PUBLIC_SEARCH_URL,
data=body,
headers=headers,
method="POST",
)
try:
opener = build_opener(_NoRedirect)
with opener.open(request, timeout=KEENABLE_SEARCH_TIMEOUT_SECONDS) as response:
raw = response.read(KEENABLE_MAX_RESPONSE_BYTES + 1)
except URLError as exc:
logger.error(f"Keenable search error: {exc}")
return [{"error": str(exc)}]
except Exception as exc:
logger.error(f"Keenable search error: {exc}")
return [{"error": str(exc)}]
if len(raw) > KEENABLE_MAX_RESPONSE_BYTES:
return [{"error": "Keenable returned an oversized response"}]
try:
payload = json.loads(raw)
except ValueError as exc:
return [{"error": f"Keenable returned invalid JSON: {exc}"}]
if not isinstance(payload, dict) or not isinstance(payload.get("results"), list):
return [{"error": "Keenable returned an invalid response shape"}]
results = []
for result in payload["results"]:
if not isinstance(result, dict):
continue
url = result.get("url")
if not isinstance(url, str) or not url or len(url) > KEENABLE_MAX_URL_CHARS:
continue
# The page text is in `snippet`; `description` is usually empty.
snippet = " ".join(str(result.get("snippet") or result.get("description") or "").split())
results.append({
"title": _truncate_text(result.get("title", ""), KEENABLE_MAX_TITLE_CHARS),
"url": url,
"snippet": _truncate_text(snippet, KEENABLE_MAX_SNIPPET_CHARS),
"provider": "keenable",
})
if len(results) >= result_limit:
break
return results
def keenable_search(query: str, max_results: int = 5) -> List[Dict[str, Any]]:
"""Search the web with Keenable (keyless; KEENABLE_API_KEY raises limits)."""
return KeenableSearchTool().search(query=query, max_results=max_results)
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-tools"
version = "0.2.44"
version = "0.2.45"
description = "Extended tools for PraisonAI Agents"
authors = [
{name = "Mervin Praison"}
@@ -0,0 +1,273 @@
"""Unit tests for KeenableSearchTool."""
import json
import os
from unittest.mock import MagicMock, patch
from praisonai_tools.tools.keenable_search_tool import (
KEENABLE_MAX_RESULTS,
KEENABLE_MAX_SNIPPET_CHARS,
KEENABLE_MAX_TITLE_CHARS,
KEENABLE_MAX_URL_CHARS,
KEENABLE_PUBLIC_SEARCH_URL,
KEENABLE_SEARCH_URL,
KeenableSearchTool,
_NoRedirect,
keenable_search,
)
class _FakeResponse:
"""Minimal stand-in for an ``http.client.HTTPResponse`` context manager."""
def __init__(self, raw: bytes):
self._raw = raw
def read(self, amt=None):
"""Return at most ``amt`` bytes, honouring the caller's read cap."""
if amt is None:
return self._raw
return self._raw[:amt]
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def _payload(results):
return json.dumps({"results": results}).encode("utf-8")
# ── Import + schema (CI smoke) ──────────────────────────────────────
class TestImportAndSchema:
def test_import_via_package(self):
from praisonai_tools import KeenableSearchTool as PkgTool
assert PkgTool is KeenableSearchTool
def test_schema_generation(self):
tool = KeenableSearchTool()
schema = tool.get_schema()
assert schema["function"]["name"] == "keenable_search"
assert "description" in schema["function"]
assert "action" in schema["function"]["parameters"]["properties"]
# ── Endpoint selection ──────────────────────────────────────────────
class TestEndpointSelection:
def test_public_endpoint_without_key(self):
with patch.dict(os.environ, {}, clear=True):
tool = KeenableSearchTool()
assert tool.api_key == ""
captured = {}
def fake_open(request, timeout=None):
captured["url"] = request.full_url
captured["headers"] = dict(request.header_items())
return _FakeResponse(_payload([]))
opener = MagicMock()
opener.open.side_effect = fake_open
with patch("urllib.request.build_opener", return_value=opener):
tool.search("python", max_results=3)
assert captured["url"] == KEENABLE_PUBLIC_SEARCH_URL
header_names = {k.lower() for k in captured["headers"]}
assert "x-api-key" not in header_names
assert "X-keenable-title" in captured["headers"] or "X-Keenable-Title" in captured["headers"]
def test_authenticated_endpoint_with_key(self):
tool = KeenableSearchTool(api_key="secret-key")
captured = {}
def fake_open(request, timeout=None):
captured["url"] = request.full_url
captured["headers"] = dict(request.header_items())
return _FakeResponse(_payload([]))
opener = MagicMock()
opener.open.side_effect = fake_open
with patch("urllib.request.build_opener", return_value=opener):
tool.search("python", max_results=3)
assert captured["url"] == KEENABLE_SEARCH_URL
# urllib title-cases header keys.
assert captured["headers"].get("X-api-key") == "secret-key"
# ── Result mapping + bounds ─────────────────────────────────────────
class TestResultMapping:
def _run(self, tool, results):
opener = MagicMock()
opener.open.return_value = _FakeResponse(_payload(results))
with patch("urllib.request.build_opener", return_value=opener):
return tool.search("q", max_results=10)
def test_maps_fields(self):
out = self._run(
KeenableSearchTool(),
[{"title": "T", "url": "https://example.com", "snippet": "hello world"}],
)
assert out == [{
"title": "T",
"url": "https://example.com",
"snippet": "hello world",
"provider": "keenable",
}]
def test_skips_entries_without_url(self):
out = self._run(
KeenableSearchTool(),
[{"title": "no url"}, {"title": "ok", "url": "https://a.com"}],
)
assert len(out) == 1
assert out[0]["url"] == "https://a.com"
def test_respects_max_results(self):
tool = KeenableSearchTool()
many = [{"url": f"https://a.com/{i}"} for i in range(10)]
opener = MagicMock()
opener.open.return_value = _FakeResponse(_payload(many))
with patch("urllib.request.build_opener", return_value=opener):
out = tool.search("q", max_results=3)
assert len(out) == 3
def test_zero_max_results_short_circuits(self):
out = KeenableSearchTool().search("q", max_results=0)
assert out == []
def test_empty_query_errors(self):
out = KeenableSearchTool().search("", max_results=5)
assert out == [{"error": "query is required"}]
# ── Error handling ──────────────────────────────────────────────────
class TestErrorHandling:
def test_invalid_shape_errors(self):
tool = KeenableSearchTool()
opener = MagicMock()
opener.open.return_value = _FakeResponse(b'{"not_results": 1}')
with patch("urllib.request.build_opener", return_value=opener):
out = tool.search("q")
assert out and "error" in out[0]
def test_oversized_response_errors(self):
tool = KeenableSearchTool()
oversized = b"x" * (1_000_001 + 1)
opener = MagicMock()
opener.open.return_value = _FakeResponse(oversized)
with patch("urllib.request.build_opener", return_value=opener):
out = tool.search("q")
assert out == [{"error": "Keenable returned an oversized response"}]
def test_http_error_is_captured(self):
tool = KeenableSearchTool()
opener = MagicMock()
opener.open.side_effect = RuntimeError("boom")
with patch("urllib.request.build_opener", return_value=opener):
out = tool.search("q")
assert out == [{"error": "boom"}]
# ── Protective bounds + redirect refusal ────────────────────────────
class TestProtections:
def _capture_body(self, tool, requested):
"""Run a search and return the decoded JSON request body sent."""
captured = {}
def fake_open(request, timeout=None):
captured["body"] = json.loads(request.data.decode("utf-8"))
return _FakeResponse(_payload([]))
opener = MagicMock()
opener.open.side_effect = fake_open
with patch("urllib.request.build_opener", return_value=opener):
tool.search("q", max_results=requested)
return captured["body"]
def test_caps_requested_results_at_limit(self):
"""max_results above the hard cap is clamped before the request."""
body = self._capture_body(KeenableSearchTool(), requested=1000)
assert body["max_results"] == KEENABLE_MAX_RESULTS
def test_truncates_oversized_title(self):
"""Titles longer than the cap are truncated with an ellipsis."""
opener = MagicMock()
opener.open.return_value = _FakeResponse(
_payload([{"title": "T" * 5000, "url": "https://a.com"}])
)
with patch("urllib.request.build_opener", return_value=opener):
out = KeenableSearchTool().search("q")
assert len(out[0]["title"]) == KEENABLE_MAX_TITLE_CHARS
assert out[0]["title"].endswith("…")
def test_truncates_oversized_snippet(self):
"""Snippets longer than the cap are truncated with an ellipsis."""
opener = MagicMock()
opener.open.return_value = _FakeResponse(
_payload([{"snippet": "s " * 5000, "url": "https://a.com"}])
)
with patch("urllib.request.build_opener", return_value=opener):
out = KeenableSearchTool().search("q")
assert len(out[0]["snippet"]) == KEENABLE_MAX_SNIPPET_CHARS
assert out[0]["snippet"].endswith("…")
def test_skips_oversized_url(self):
"""Results whose URL exceeds the length cap are dropped."""
long_url = "https://a.com/" + "x" * KEENABLE_MAX_URL_CHARS
opener = MagicMock()
opener.open.return_value = _FakeResponse(
_payload([{"url": long_url}, {"url": "https://ok.com"}])
)
with patch("urllib.request.build_opener", return_value=opener):
out = KeenableSearchTool().search("q")
assert out == [{
"title": "",
"url": "https://ok.com",
"snippet": "",
"provider": "keenable",
}]
def test_read_is_capped_at_limit(self):
"""The body read is bounded, so an oversized stream still errors out."""
tool = KeenableSearchTool()
# Far larger than the cap; _FakeResponse.read honours the byte count,
# so this verifies the read limit rather than a pre-sized buffer.
opener = MagicMock()
opener.open.return_value = _FakeResponse(b"x" * 5_000_000)
with patch("urllib.request.build_opener", return_value=opener):
out = tool.search("q")
assert out == [{"error": "Keenable returned an oversized response"}]
def test_refuses_redirects(self):
"""The redirect handler drops 3xx responses so no follow-up is sent."""
handler = _NoRedirect()
# Returning None is urllib's signal to NOT build/send a second request.
assert handler.redirect_request(
MagicMock(), MagicMock(), 302, "Found", {}, "https://evil.example"
) is None
# ── Module-level convenience function ───────────────────────────────
class TestConvenienceFunction:
def test_keenable_search_function(self):
opener = MagicMock()
opener.open.return_value = _FakeResponse(_payload([{"url": "https://a.com"}]))
with patch("urllib.request.build_opener", return_value=opener):
out = keenable_search("q", max_results=2)
assert out[0]["provider"] == "keenable"