Release v0.2.46

feat: add You.com tools (migrated from praisonaiagents core)

.env.example

@@ -0,0 +1,12 @@
+# PraisonAI-Tools environment variables (examples)
+#
+# Copy to `.env` and fill in the values for the tools you use.
+# Never commit real secrets.
+
+# ── Stripe (StripeTool) ─────────────────────────────────────────────
+# Stripe secret key. Prefer least-privilege *restricted* keys (rk_...).
+# Use test keys (sk_test_/rk_test_) for development and live keys
+# (sk_live_/rk_live_) only in production. STRIPE_API_KEY is accepted as a
+# fallback if STRIPE_SECRET_KEY is unset.
+STRIPE_SECRET_KEY=sk_test_your_key_here
+# STRIPE_API_KEY=sk_test_your_key_here

.github/automation.meta.json

@@ -3,7 +3,7 @@
   "stack": "claude",
   "profile": "tools",
   "mode": "hub",
-  "hubRef": "v1.1.8",
+  "hubRef": "v1.1.15",
   "templateVersion": "1.1.6",
   "installedAt": "2026-09-07T12:54:25Z"
 }

.github/workflows/auto-pr-comment.yml

@@ -10,7 +10,7 @@ on:
   pull_request_target:
     types: [opened, synchronize]
   schedule:
-    - cron: '30 3,9,15,21 * * *'
+    - cron: '30 * * * *'
   workflow_dispatch:
 
 env:
@@ -27,7 +27,7 @@ jobs:
     permissions:
       issues: write
       pull-requests: write
-    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].8
+    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].20
     secrets: inherit
     with:
-      hub_ref: v1.1.8
+      hub_ref: v1.1.20

.github/workflows/bot-pr-recovery.yml

@@ -16,8 +16,8 @@ permissions:
 
 jobs:
   recover:
-    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].8
+    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].20
     secrets: inherit
     with:
-      hub_ref: v1.1.8
+      hub_ref: v1.1.20
       pr_number: ${{ github.event.inputs.pr_number || '' }}

.github/workflows/ci-failure-claude.yml

@@ -25,8 +25,8 @@ permissions:
 
 jobs:
   detect-and-trigger:
-    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].8
+    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].20
     secrets: inherit
     with:
-      hub_ref: v1.1.8
+      hub_ref: v1.1.20
       pr_number: ${{ github.event.inputs.pr_number || '' }}

.github/workflows/claude-merge-gate.yml

@@ -43,9 +43,9 @@ jobs:
       id-token: write
     steps:
       - uses: actions/checkout@v4
-      - uses: MervinPraison/github-automation-template/.github/actions/[email protected].8
+      - uses: MervinPraison/github-automation-template/.github/actions/[email protected].20
         with:
-          hub_ref: v1.1.8
+          hub_ref: v1.1.20
       - name: Generate GitHub App token
         id: app-token
         uses: actions/create-github-app-token@v1
@@ -87,9 +87,10 @@ jobs:
       issues: write
       actions: read
       id-token: write
-    uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
+      statuses: write   # the assess job records security/outsider-review
+    uses: MervinPraison/github-automation-template/.github/workflows/[email protected]
     secrets: inherit
     with:
-      hub_ref: v1.1.8
+      hub_ref: v1.1.20
       pr_number: ${{ github.event.inputs.pr_number || github.event.client_payload.pr_number || '' }}
       merge_method: ${{ github.event.inputs.merge_method || 'merge' }}

.github/workflows/claude.yml

@@ -18,7 +18,7 @@ jobs:
       issues: write
       actions: read
       id-token: write
-    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].8
+    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].20
     secrets: inherit
     with:
-      hub_ref: v1.1.8
+      hub_ref: v1.1.20

.github/workflows/merge-conflict-claude.yml

@@ -25,9 +25,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - uses: MervinPraison/github-automation-template/.github/actions/[email protected].8
+      - uses: MervinPraison/github-automation-template/.github/actions/[email protected].20
         with:
-          hub_ref: v1.1.8
+          hub_ref: v1.1.20
       - name: Detect conflicting PRs and trigger Claude
         uses: actions/github-script@v7
         with:

.github/workflows/pipeline-status-sync.yml

@@ -13,7 +13,7 @@ permissions:
 
 jobs:
   sync:
-    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].8
+    uses: MervinPraison/github-automation-template/.github/workflows/[email protected].20
     secrets: inherit
     with:
-      hub_ref: v1.1.8
+      hub_ref: v1.1.20

README.md

@@ -646,6 +646,23 @@ videos = yt.search("python tutorial", limit=5)
 transcript = yt.get_transcript("dQw4w9WgXcQ")
 ```
 
+### Meeting Intelligence Tools
+
+```python
+from praisonai_tools import FirefliesTool
+
+# Fireflies.ai meeting intelligence (curated GraphQL operations)
+fireflies = FirefliesTool()  # Uses FIREFLIES_API_KEY env var
+meetings = fireflies.list_recent_meetings(limit=5)
+meetings = fireflies.list_recent_meetings(limit=5, after="5")  # cursor pagination
+transcript = fireflies.get_transcript(meeting_id="abc123")
+results = fireflies.search(keywords="roadmap")
+```
+
+`FirefliesTool` distinguishes a missing key (`{"error": "FIREFLIES_API_KEY required"}`)
+from a rejected key (HTTP 401/403), and surfaces `retry_after` on rate limits.
+Set `FIREFLIES_API_URL` to override the endpoint (previews/proxies).
+
 ### Data Tools
 
 ```python
@@ -669,6 +686,7 @@ air = weather.get_air_quality("Tokyo")
 | TTSTool | `OPENAI_API_KEY` or `ELEVENLABS_API_KEY` |
 | YouTubeTool | `YOUTUBE_API_KEY` |
 | WeatherTool | `OPENWEATHER_API_KEY` |
+| FirefliesTool | `FIREFLIES_API_KEY` (optional `FIREFLIES_API_URL`) |
 | SwarmScoreTool | None (uses public API) |
 
 ### Using with PraisonAI Agents

praisonai_tools/catalogue.py

@@ -55,6 +55,7 @@ _MODULE_EXTRAS = {
     "langextract_tool": ("langextract",),
     "swarmscore_tool": ("swarmscore",),
     "composio_tool": ("composio",),
+    "stripe_tool": ("stripe",),
 }
 
 

praisonai_tools/tools/fireflies_tool.py

@@ -0,0 +1,268 @@
+"""Fireflies.ai Tool for PraisonAI Agents.
+
+Curated GraphQL operations over the Fireflies.ai meeting-intelligence API:
+list recent meetings, fetch a transcript, and keyword search.
+
+Design (Option B from the issue): each agent-callable method maps to a small,
+pre-approved GraphQL document rather than letting the LLM craft arbitrary
+queries. This keeps the surface predictable and robust.
+
+Usage:
+    from praisonai_tools import FirefliesTool
+
+    fireflies = FirefliesTool()
+    meetings = fireflies.list_recent_meetings(limit=5)
+    transcript = fireflies.get_transcript(meeting_id="abc123")
+
+Environment Variables:
+    FIREFLIES_API_KEY: Fireflies API key (sent as ``Authorization: Bearer <key>``)
+    FIREFLIES_API_URL: Optional endpoint override (defaults to the public API)
+"""
+
+import os
+import logging
+from typing import Any, Dict, List, Optional, Union
+
+from praisonai_tools.tools.base import BaseTool
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_API_URL = "https://api.fireflies.ai/graphql"
+
+
+class FirefliesTool(BaseTool):
+    """Tool for querying Fireflies.ai meeting intelligence via curated GraphQL."""
+
+    name = "fireflies"
+    description = (
+        "Query Fireflies.ai meeting intelligence: list recent meetings, "
+        "fetch transcripts, and search by keyword."
+    )
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        api_url: Optional[str] = None,
+    ):
+        self.api_key = api_key or os.getenv("FIREFLIES_API_KEY")
+        self.api_url = api_url or os.getenv("FIREFLIES_API_URL") or DEFAULT_API_URL
+        super().__init__()
+
+    def _graphql(self, query: str, variables: Dict = None) -> Dict:
+        """Execute a curated GraphQL document, returning normalized results.
+
+        Missing credentials are reported distinctly from HTTP auth failures so
+        callers can tell a misconfiguration (no key) apart from a rejected key
+        (401/403).
+        """
+        if not self.api_key:
+            return {"error": "FIREFLIES_API_KEY required"}
+
+        # Never transmit the API key over a plaintext endpoint.
+        if not self.api_url.lower().startswith("https://"):
+            return {"error": "FIREFLIES_API_URL must use HTTPS"}
+
+        try:
+            import requests
+        except ImportError:
+            return {"error": "requests not installed"}
+
+        try:
+            resp = requests.post(
+                self.api_url,
+                headers={
+                    "Authorization": f"Bearer {self.api_key}",
+                    "Content-Type": "application/json",
+                },
+                json={"query": query, "variables": variables or {}},
+                timeout=30,
+            )
+        except Exception as e:
+            return {"error": str(e)}
+
+        if resp.status_code in (401, 403):
+            return {
+                "error": f"Fireflies authentication failed (HTTP {resp.status_code})",
+                "status_code": resp.status_code,
+            }
+        if resp.status_code == 429:
+            retry_after = resp.headers.get("Retry-After") or self._graphql_retry_after(resp)
+            error = {"error": "Fireflies rate limit exceeded", "status_code": 429}
+            if retry_after:
+                error["retry_after"] = retry_after
+            return error
+
+        try:
+            payload = resp.json()
+        except Exception:
+            return {
+                "error": f"Fireflies returned non-JSON response (HTTP {resp.status_code})",
+                "status_code": resp.status_code,
+            }
+
+        if payload.get("errors"):
+            messages = [e.get("message", "unknown error") for e in payload["errors"]]
+            return {"error": "; ".join(messages)}
+        # Surface any remaining HTTP failure instead of letting an error body be
+        # normalized into an empty/missing result by the operation handlers.
+        if resp.status_code >= 400:
+            return {
+                "error": f"Fireflies request failed (HTTP {resp.status_code})",
+                "status_code": resp.status_code,
+            }
+        return payload
+
+    @staticmethod
+    def _graphql_retry_after(resp) -> Optional[str]:
+        """Extract retryAfter from a GraphQL 429 body when the header is absent."""
+        try:
+            body = resp.json()
+        except Exception:
+            return None
+        for err in body.get("errors") or []:
+            metadata = ((err.get("extensions") or {}).get("metadata")) or {}
+            if metadata.get("retryAfter"):
+                return str(metadata["retryAfter"])
+        return None
+
+    def run(
+        self,
+        action: str = "list_recent_meetings",
+        meeting_id: Optional[str] = None,
+        keywords: Optional[str] = None,
+        **kwargs,
+    ) -> Union[str, Dict[str, Any], List[Dict[str, Any]]]:
+        action = action.lower().replace("-", "_")
+
+        if action == "list_recent_meetings":
+            return self.list_recent_meetings(**kwargs)
+        elif action == "get_transcript":
+            return self.get_transcript(meeting_id=meeting_id)
+        elif action == "search":
+            return self.search(keywords=keywords, **kwargs)
+        else:
+            return {"error": f"Unknown action: {action}"}
+
+    def list_recent_meetings(
+        self, limit: int = 10, after: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """List recent meetings/transcripts.
+
+        Args:
+            limit: Maximum number of meetings to return.
+            after: Cursor (skip offset) for pagination. Explicit paging avoids
+                unbounded scans of the account history.
+        """
+        query = """
+        query($limit: Int, $skip: Int) {
+            transcripts(limit: $limit, skip: $skip) {
+                id
+                title
+                date
+                duration
+                organizer_email
+            }
+        }
+        """
+        limit = max(1, min(limit, 50))  # Fireflies caps page size at 50.
+        variables: Dict[str, Any] = {"limit": limit}
+        if after is not None:
+            try:
+                variables["skip"] = int(after)
+            except (TypeError, ValueError):
+                return [{"error": "after must be an integer cursor"}]
+
+        result = self._graphql(query, variables)
+        if "error" in result:
+            return [result]
+
+        transcripts = result.get("data", {}).get("transcripts") or []
+        return [
+            {
+                "id": t.get("id"),
+                "title": t.get("title"),
+                "date": t.get("date"),
+                "duration": t.get("duration"),
+                "organizer_email": t.get("organizer_email"),
+            }
+            for t in transcripts
+        ]
+
+    def get_transcript(self, meeting_id: str) -> Dict[str, Any]:
+        """Fetch a single meeting transcript with summary and sentences."""
+        if not meeting_id:
+            return {"error": "meeting_id required"}
+
+        query = """
+        query($id: String!) {
+            transcript(id: $id) {
+                id
+                title
+                date
+                duration
+                organizer_email
+                summary { overview }
+                sentences {
+                    text
+                    speaker_name
+                    start_time
+                }
+            }
+        }
+        """
+        result = self._graphql(query, {"id": meeting_id})
+        if "error" in result:
+            return result
+
+        transcript = result.get("data", {}).get("transcript")
+        if not transcript:
+            return {"error": "transcript not found"}
+
+        summary = transcript.get("summary") or {}
+        sentences = transcript.get("sentences") or []
+        return {
+            "id": transcript.get("id"),
+            "title": transcript.get("title"),
+            "date": transcript.get("date"),
+            "duration": transcript.get("duration"),
+            "organizer_email": transcript.get("organizer_email"),
+            "summary": summary.get("overview"),
+            "sentences": sentences,
+            "sentence_count": len(sentences),
+        }
+
+    def search(self, keywords: str, limit: int = 10) -> List[Dict[str, Any]]:
+        """Search meetings whose transcripts match the given keyword phrase."""
+        if not keywords:
+            return [{"error": "keywords required"}]
+
+        limit = max(1, min(limit, 50))  # Fireflies caps page size at 50.
+        query = """
+        query($keyword: String, $limit: Int) {
+            transcripts(keyword: $keyword, limit: $limit) {
+                id
+                title
+                date
+                organizer_email
+            }
+        }
+        """
+        result = self._graphql(query, {"keyword": keywords, "limit": limit})
+        if "error" in result:
+            return [result]
+
+        transcripts = result.get("data", {}).get("transcripts") or []
+        return [
+            {
+                "id": t.get("id"),
+                "title": t.get("title"),
+                "date": t.get("date"),
+                "organizer_email": t.get("organizer_email"),
+            }
+            for t in transcripts
+        ]
+
+
+def list_fireflies_meetings(limit: int = 10) -> List[Dict[str, Any]]:
+    """List recent Fireflies meetings."""
+    return FirefliesTool().list_recent_meetings(limit=limit)

praisonai_tools/tools/keenable_search_tool.py

@@ -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)

praisonai_tools/tools/reddit_tool.py

@@ -17,6 +17,17 @@ Environment Variables:
     REDDIT_CLIENT_ID: Reddit API client ID
     REDDIT_CLIENT_SECRET: Reddit API client secret
     REDDIT_USER_AGENT: User agent string (optional)
+    REDDIT_USERNAME: Reddit username (required for write actions)
+    REDDIT_PASSWORD: Reddit password (required for write actions)
+    REDDIT_ALLOW_WRITE: Set to "1" to enable write actions (post/comment)
+
+Write actions (submit_post, post_comment) are disabled by default to reduce
+spam/abuse surface. They are only permitted when ``REDDIT_ALLOW_WRITE=1`` (or
+``allow_write=True``) and Reddit account credentials are configured. You are
+responsible for complying with Reddit's API terms when enabling writes.
+
+Reddit rate-limits misconfigured clients aggressively. Use a descriptive
+User-Agent referencing your app id, e.g. ``script:my-app:v1.0 (by /u/you)``.
 """
 
 import os
@@ -32,13 +43,20 @@ class RedditTool(BaseTool):
     """Tool for interacting with Reddit."""
     
     name = "reddit"
-    description = "Search Reddit, get posts from subreddits, and read comments."
+    description = (
+        "Search Reddit, get posts from subreddits, and read comments. "
+        "Can also comment on posts and submit new posts when write actions "
+        "are enabled (REDDIT_ALLOW_WRITE=1 with account credentials)."
+    )
     
     def __init__(
         self,
         client_id: Optional[str] = None,
         client_secret: Optional[str] = None,
         user_agent: Optional[str] = None,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        allow_write: Optional[bool] = None,
     ):
         """Initialize RedditTool.
         
@@ -46,13 +64,30 @@ class RedditTool(BaseTool):
             client_id: Reddit API client ID
             client_secret: Reddit API client secret
             user_agent: User agent string
+            username: Reddit username (required for write actions)
+            password: Reddit password (required for write actions)
+            allow_write: Enable write actions. Defaults to the
+                ``REDDIT_ALLOW_WRITE`` env var ("1"/"true" enables).
         """
         self.client_id = client_id or os.getenv("REDDIT_CLIENT_ID")
         self.client_secret = client_secret or os.getenv("REDDIT_CLIENT_SECRET")
-        self.user_agent = user_agent or os.getenv("REDDIT_USER_AGENT", "PraisonAI/1.0")
+        self.user_agent = user_agent or os.getenv("REDDIT_USER_AGENT") or self._default_user_agent()
+        self.username = username or os.getenv("REDDIT_USERNAME")
+        self.password = password or os.getenv("REDDIT_PASSWORD")
+        if allow_write is None:
+            allow_write = os.getenv("REDDIT_ALLOW_WRITE", "").lower() in ("1", "true", "yes")
+        self.allow_write = allow_write
         self._reddit = None
         super().__init__()
     
+    def _default_user_agent(self) -> str:
+        """Build a compliant User-Agent referencing the app id.
+        
+        Reddit recommends ``<platform>:<app id>:<version> (by /u/<user>)``.
+        """
+        app_id = self.client_id or "unknown"
+        return f"script:praisonai-{app_id}:1.0 (by PraisonAI)"
+    
     @property
     def reddit(self):
         """Lazy-load Reddit client."""
@@ -65,13 +100,31 @@ class RedditTool(BaseTool):
             if not self.client_id or not self.client_secret:
                 raise ValueError("REDDIT_CLIENT_ID and REDDIT_CLIENT_SECRET required")
             
-            self._reddit = praw.Reddit(
-                client_id=self.client_id,
-                client_secret=self.client_secret,
-                user_agent=self.user_agent,
-            )
+            kwargs = {
+                "client_id": self.client_id,
+                "client_secret": self.client_secret,
+                "user_agent": self.user_agent,
+            }
+            if self.username and self.password:
+                kwargs["username"] = self.username
+                kwargs["password"] = self.password
+            
+            self._reddit = praw.Reddit(**kwargs)
         return self._reddit
     
+    def _check_write_allowed(self) -> Optional[Dict[str, Any]]:
+        """Return an error dict if write actions are not permitted, else None."""
+        if not self.allow_write:
+            return {
+                "error": "Write actions disabled. Set REDDIT_ALLOW_WRITE=1 to enable "
+                "posting/commenting.",
+            }
+        if not self.username or not self.password:
+            return {
+                "error": "REDDIT_USERNAME and REDDIT_PASSWORD required for write actions.",
+            }
+        return None
+    
     def run(
         self,
         action: str = "search",
@@ -98,6 +151,15 @@ class RedditTool(BaseTool):
             return self.get_comments(post_id=post_id, limit=limit)
         elif action == "subreddit_info":
             return self.get_subreddit_info(subreddit=subreddit)
+        elif action == "post_comment":
+            return self.post_comment(post_id=post_id, body=kwargs.get("body"))
+        elif action == "submit_post":
+            return self.submit_post(
+                subreddit=subreddit,
+                title=kwargs.get("title"),
+                body=kwargs.get("body"),
+                url=kwargs.get("url"),
+            )
         else:
             return {"error": f"Unknown action: {action}"}
     
@@ -284,6 +346,83 @@ class RedditTool(BaseTool):
             logger.error(f"Reddit get_subreddit_info error: {e}")
             return {"error": str(e)}
     
+    def post_comment(self, post_id: str, body: str) -> Dict[str, Any]:
+        """Reply with a comment to a post (write action).
+        
+        Gated behind ``REDDIT_ALLOW_WRITE=1`` and account credentials.
+        
+        Args:
+            post_id: Reddit post ID to comment on
+            body: Comment text
+            
+        Returns:
+            Result with the new comment id, or an error dict
+        """
+        denied = self._check_write_allowed()
+        if denied:
+            return denied
+        if not post_id:
+            return {"error": "post_id is required"}
+        if not body:
+            return {"error": "body is required"}
+        
+        try:
+            submission = self.reddit.submission(id=post_id)
+            comment = submission.reply(body)
+            return {
+                "success": True,
+                "id": comment.id,
+                "permalink": f"https://reddit.com{comment.permalink}",
+            }
+        except Exception as e:
+            logger.error(f"Reddit post_comment error: {e}")
+            return {"error": str(e)}
+    
+    def submit_post(
+        self,
+        subreddit: str,
+        title: str,
+        body: Optional[str] = None,
+        url: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Submit a new post to a subreddit (write action).
+        
+        Gated behind ``REDDIT_ALLOW_WRITE=1`` and account credentials.
+        
+        Args:
+            subreddit: Target subreddit name
+            title: Post title
+            body: Self-post text (mutually exclusive with url)
+            url: Link URL (mutually exclusive with body)
+            
+        Returns:
+            Result with the new post id, or an error dict
+        """
+        denied = self._check_write_allowed()
+        if denied:
+            return denied
+        if not subreddit:
+            return {"error": "subreddit is required"}
+        if not title:
+            return {"error": "title is required"}
+        if url and body:
+            return {"error": "provide either body or url, not both"}
+        
+        try:
+            sub = self.reddit.subreddit(subreddit)
+            if url:
+                submission = sub.submit(title, url=url)
+            else:
+                submission = sub.submit(title, selftext=body or "")
+            return {
+                "success": True,
+                "id": submission.id,
+                "permalink": f"https://reddit.com{submission.permalink}",
+            }
+        except Exception as e:
+            logger.error(f"Reddit submit_post error: {e}")
+            return {"error": str(e)}
+    
     def _format_post(self, post, include_body: bool = False) -> Dict[str, Any]:
         """Format a Reddit post."""
         result = {

praisonai_tools/tools/stripe_tool.py

@@ -0,0 +1,338 @@
+"""Stripe Tool for PraisonAI Agents.
+
+Narrow, audited Stripe operations — retrieve customers/payment intents,
+list payment links, and create payment links with required correlation
+metadata and optional idempotency keys.
+
+Design constraint: prefer narrow, audited operations over a "god tool"
+that dumps the entire Stripe surface. One verb per action.
+
+Usage:
+    from praisonai_tools import StripeTool
+
+    stripe = StripeTool()  # reads STRIPE_SECRET_KEY (or STRIPE_API_KEY)
+    customer = stripe.run(action="get_customer", customer_id="cus_123")
+    link = stripe.run(
+        action="create_payment_link",
+        price_id="price_123",
+        quantity=1,
+        correlation_id="order-42",
+    )
+
+Environment Variables:
+    STRIPE_SECRET_KEY: Stripe secret key. Restricted keys are recommended.
+        Accepts test (``sk_test_``/``rk_test_``) or live
+        (``sk_live_``/``rk_live_``) keys. STRIPE_API_KEY is also accepted
+        as a fallback.
+
+Security notes:
+    * Secrets are read from the environment only and never logged.
+    * Responses include a ``mode`` banner (``test``/``live``) derived from
+      the key prefix so live operations are visible to callers.
+    * Structured logs record only the operation, mode, idempotency key, and
+      the Stripe object id — never full request/response payloads.
+"""
+
+import os
+import logging
+from typing import Any, Dict, List, Optional, Union
+
+from praisonai_tools.tools.base import BaseTool
+
+logger = logging.getLogger(__name__)
+
+API_BASE = "https://api.stripe.com/v1"
+
+_TEST_PREFIXES = ("sk_test_", "rk_test_")
+_LIVE_PREFIXES = ("sk_live_", "rk_live_")
+
+
+class StripeTool(BaseTool):
+    """Tool for narrow, audited Stripe operations."""
+
+    name = "stripe"
+    description = (
+        "Retrieve Stripe customers and payment intents, list payment links, "
+        "and create payment links with required correlation metadata."
+    )
+
+    def __init__(self, api_key: Optional[str] = None):
+        self.api_key = (
+            api_key
+            or os.getenv("STRIPE_SECRET_KEY")
+            or os.getenv("STRIPE_API_KEY")
+        )
+        super().__init__()
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _mode(self) -> str:
+        """Return ``test``, ``live``, or ``unknown`` from the key prefix."""
+        key = self.api_key or ""
+        if key.startswith(_LIVE_PREFIXES):
+            return "live"
+        if key.startswith(_TEST_PREFIXES):
+            return "test"
+        return "unknown"
+
+    def _require_key(self) -> Optional[Dict[str, str]]:
+        if not self.api_key:
+            return {"error": "STRIPE_SECRET_KEY (or STRIPE_API_KEY) is required"}
+        return None
+
+    @staticmethod
+    def _import_requests():
+        try:
+            import requests
+            return requests
+        except ImportError:
+            return None
+
+    def _request(
+        self,
+        method: str,
+        endpoint: str,
+        data: Optional[Dict[str, Any]] = None,
+        idempotency_key: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Make a request to the Stripe REST API.
+
+        Centralises key validation, the optional ``requests`` import, the
+        HTTP call, auditable logging, and error handling.
+        """
+        err = self._require_key()
+        if err:
+            return err
+
+        requests = self._import_requests()
+        if requests is None:
+            return {"error": "requests package is not installed"}
+
+        headers = {"Authorization": f"Bearer {self.api_key}"}
+        if idempotency_key:
+            headers["Idempotency-Key"] = idempotency_key
+
+        try:
+            resp = requests.request(
+                method,
+                f"{API_BASE}/{endpoint}",
+                headers=headers,
+                data=data,
+                timeout=30,
+            )
+            body = resp.json()
+        except requests.exceptions.RequestException as e:
+            logger.error("Stripe request error: %s", e)
+            return {"error": f"API request failed: {e}"}
+        except ValueError as e:
+            logger.error("Stripe JSON decode error: %s", e)
+            return {"error": "Failed to decode API response"}
+
+        if isinstance(body, dict) and "error" in body:
+            stripe_err = body["error"]
+            message = stripe_err.get("message", "Stripe API error")
+            logger.error(
+                "Stripe API error: op=%s mode=%s type=%s",
+                endpoint, self._mode(), stripe_err.get("type"),
+            )
+            return {"error": message}
+
+        # Defensive: a non-2xx status without Stripe's usual ``error`` object
+        # must not be treated as a successful payload.
+        if resp.status_code >= 400:
+            logger.error(
+                "Stripe HTTP error: op=%s mode=%s status=%s",
+                endpoint, self._mode(), resp.status_code,
+            )
+            return {"error": f"Stripe API returned HTTP {resp.status_code}"}
+
+        obj_id = body.get("id") if isinstance(body, dict) else None
+        logger.info(
+            "Stripe op=%s mode=%s idempotency_key=%s object_id=%s",
+            endpoint, self._mode(), idempotency_key, obj_id,
+        )
+        return body
+
+    def _banner(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """Attach a live/test mode banner to a successful response."""
+        if isinstance(payload, dict) and "error" not in payload:
+            payload = {"mode": self._mode(), **payload}
+        return payload
+
+    # ------------------------------------------------------------------
+    # Dispatcher
+    # ------------------------------------------------------------------
+
+    def run(
+        self,
+        action: str = "get_customer",
+        customer_id: Optional[str] = None,
+        payment_intent_id: Optional[str] = None,
+        price_id: Optional[str] = None,
+        quantity: int = 1,
+        correlation_id: Optional[str] = None,
+        idempotency_key: Optional[str] = None,
+        limit: int = 10,
+        **kwargs,
+    ) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
+        """Dispatch to the appropriate Stripe action.
+
+        Actions:
+            get_customer         — Retrieve a customer by id (read).
+            get_payment_intent   — Retrieve a payment intent by id (read).
+            list_payment_links   — List payment links (read).
+            create_payment_link  — Create a payment link (write, guarded).
+        """
+        action = action.lower().replace("-", "_")
+
+        if action == "get_customer":
+            return self.get_customer(customer_id=customer_id)
+        elif action == "get_payment_intent":
+            return self.get_payment_intent(payment_intent_id=payment_intent_id)
+        elif action == "list_payment_links":
+            return self.list_payment_links(limit=limit)
+        elif action == "create_payment_link":
+            return self.create_payment_link(
+                price_id=price_id,
+                quantity=quantity,
+                correlation_id=correlation_id,
+                idempotency_key=idempotency_key,
+            )
+        else:
+            return {"error": f"Unknown action: {action}"}
+
+    # ------------------------------------------------------------------
+    # Phase 1 — Read-mostly safety
+    # ------------------------------------------------------------------
+
+    def get_customer(self, customer_id: Optional[str] = None) -> Dict[str, Any]:
+        """Retrieve a customer by id.
+
+        Args:
+            customer_id: The Stripe customer id (``cus_...``).
+        """
+        if not customer_id:
+            return {"error": "customer_id is required"}
+
+        result = self._request("get", f"customers/{customer_id}")
+        if "error" in result:
+            return result
+
+        return self._banner({
+            "id": result.get("id"),
+            "email": result.get("email"),
+            "name": result.get("name"),
+            "created": result.get("created"),
+            "currency": result.get("currency"),
+            "delinquent": result.get("delinquent"),
+        })
+
+    def get_payment_intent(
+        self, payment_intent_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Retrieve a payment intent by id.
+
+        Args:
+            payment_intent_id: The Stripe payment intent id (``pi_...``).
+        """
+        if not payment_intent_id:
+            return {"error": "payment_intent_id is required"}
+
+        result = self._request("get", f"payment_intents/{payment_intent_id}")
+        if "error" in result:
+            return result
+
+        return self._banner({
+            "id": result.get("id"),
+            "amount": result.get("amount"),
+            "currency": result.get("currency"),
+            "status": result.get("status"),
+            "customer": result.get("customer"),
+            "created": result.get("created"),
+        })
+
+    def list_payment_links(self, limit: int = 10) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
+        """List payment links.
+
+        Args:
+            limit: Maximum number of payment links to return (1–100).
+        """
+        limit = max(1, min(int(limit), 100))
+        result = self._request("get", f"payment_links?limit={limit}")
+        if "error" in result:
+            return result
+
+        links = [
+            {
+                "id": link.get("id"),
+                "url": link.get("url"),
+                "active": link.get("active"),
+                "metadata": link.get("metadata", {}),
+            }
+            for link in result.get("data", [])
+        ]
+        return self._banner({"payment_links": links})
+
+    # ------------------------------------------------------------------
+    # Phase 2 — Mutations with guardrails
+    # ------------------------------------------------------------------
+
+    def create_payment_link(
+        self,
+        price_id: Optional[str] = None,
+        quantity: int = 1,
+        correlation_id: Optional[str] = None,
+        idempotency_key: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a payment link.
+
+        A ``correlation_id`` is required and stored as payment-link metadata
+        so every created link is traceable back to the request that made it.
+        An optional ``idempotency_key`` is surfaced to make retries safe.
+
+        Args:
+            price_id: An existing Stripe Price id (``price_...``).
+            quantity: Quantity for the line item (>= 1).
+            correlation_id: Required correlation id stored as metadata.
+            idempotency_key: Optional Stripe ``Idempotency-Key`` header value.
+        """
+        if not price_id:
+            return {"error": "price_id is required"}
+        if not correlation_id:
+            return {"error": "correlation_id is required (stored as metadata)"}
+        try:
+            quantity = int(quantity)
+        except (TypeError, ValueError):
+            return {"error": "quantity must be an integer"}
+        if quantity < 1:
+            return {"error": "quantity must be >= 1"}
+
+        data = {
+            "line_items[0][price]": price_id,
+            "line_items[0][quantity]": quantity,
+            "metadata[correlation_id]": correlation_id,
+        }
+        result = self._request(
+            "post",
+            "payment_links",
+            data=data,
+            idempotency_key=idempotency_key,
+        )
+        if "error" in result:
+            return result
+
+        return self._banner({
+            "id": result.get("id"),
+            "url": result.get("url"),
+            "active": result.get("active"),
+            "metadata": result.get("metadata", {}),
+        })
+
+
+def get_stripe_customer(
+    customer_id: str, api_key: Optional[str] = None
+) -> Dict[str, Any]:
+    """Retrieve a Stripe customer by id."""
+    return StripeTool(api_key=api_key).get_customer(customer_id=customer_id)

praisonai_tools/tools/youdotcom_tool.py

@@ -0,0 +1,484 @@
+"""You.com search and content tools for PraisonAI Agents.
+
+Migrated from ``praisonaiagents.tools.youdotcom_tools`` (MervinPraison/PraisonAI#6168).
+
+You.com provides AI-powered web search, news search, content extraction, and image search.
+This module requires the YDC_API_KEY environment variable to be set.
+
+Usage:
+    from praisonai_tools.tools.youdotcom_tool import ydc_search, ydc_news, ydc_contents
+    
+    # Search the web (unified web + news)
+    results = ydc_search("AI news 2024")
+    
+    # Get live news
+    news = ydc_news("technology trends")
+    
+    # Extract content from URLs
+    content = ydc_contents(["https://example.com"])
+    
+    # Or use the class directly
+    from praisonai_tools.tools.youdotcom_tool import YouTools
+    you = YouTools()
+    results = you.search("AI news")
+"""
+
+from typing import List, Dict, Any, Optional, Union
+import logging
+import os
+from importlib import util
+
+
+class YouTools:
+    """Comprehensive tools for web search and content using You.com API.
+    
+    You.com provides AI-powered search capabilities optimized for LLM applications.
+    
+    Features:
+    - Unified web and news search with LLM-ready snippets
+    - Live news search from authoritative sources
+    - Content extraction from URLs (HTML/Markdown)
+    - Image search
+    - Advanced search operators support
+    - Geographic and language targeting
+    - Freshness controls
+    
+    Requires:
+    - youdotcom package: pip install youdotcom
+    - YDC_API_KEY environment variable
+    """
+    
+    def __init__(self, api_key: Optional[str] = None):
+        """Initialize YouTools.
+        
+        Args:
+            api_key: Optional API key. If not provided, uses YDC_API_KEY env var.
+        """
+        self._api_key = api_key
+        self._client = None
+    
+    def _get_api_key(self) -> Optional[str]:
+        """Get API key from instance or environment."""
+        return self._api_key or os.environ.get("YDC_API_KEY")
+    
+    def _get_client(self):
+        """Get or create You.com client."""
+        if self._client is None:
+            if util.find_spec("youdotcom") is None:
+                return None, "youdotcom package is not installed. Please install it using: pip install youdotcom"
+            api_key = self._get_api_key()
+            if not api_key:
+                return None, "YDC_API_KEY environment variable is not set. Please set it to use You.com tools."
+            
+            from youdotcom import You
+            self._client = You(api_key_auth=api_key)
+        return self._client, None
+    
+    def search(
+        self,
+        query: str,
+        count: int = 10,
+        freshness: Optional[str] = None,
+        country: Optional[str] = None,
+        language: Optional[str] = None,
+        offset: int = 0,
+        safesearch: str = "moderate",
+        livecrawl: Optional[str] = None,
+        livecrawl_formats: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Search the web using You.com's unified search API.
+        
+        Returns both web and news results in a single call with LLM-ready snippets.
+        
+        Args:
+            query: Search query (supports operators like site:, filetype:, +term, -term)
+            count: Max results per section (default 10, max 100)
+            freshness: Filter by recency - "day", "week", "month", "year" or "YYYY-MM-DDtoYYYY-MM-DD"
+            country: ISO country code (e.g., "US", "GB", "FR")
+            language: BCP 47 language code (e.g., "EN", "JP", "DE")
+            offset: Pagination offset (0-9), results = offset * count
+            safesearch: Content filter - "off", "moderate", "strict"
+            livecrawl: Fetch full content - "web", "news", or "all"
+            livecrawl_formats: Content format - "html" or "markdown"
+            
+        Returns:
+            Dict containing:
+            - results: {web: [...], news: [...]}
+            - metadata: {query, search_uuid, latency}
+            
+        Search Operators:
+            - site:domain.com - Search within specific domain
+            - filetype:pdf - Filter by file type
+            - +term - Must include term
+            - -term - Exclude term
+            - AND, OR, NOT - Boolean operators
+        """
+        client, error = self._get_client()
+        if error:
+            logging.error(error)
+            return {"error": error}
+        
+        try:
+            kwargs = {"query": query, "count": count, "offset": offset}
+            
+            if freshness:
+                kwargs["freshness"] = freshness
+            if country:
+                kwargs["country"] = country
+            if language:
+                kwargs["language"] = language
+            if safesearch:
+                kwargs["safesearch"] = safesearch
+            if livecrawl:
+                kwargs["livecrawl"] = livecrawl
+            if livecrawl_formats:
+                kwargs["livecrawl_formats"] = livecrawl_formats
+            
+            response = client.search.unified(**kwargs)
+            
+            # Convert response to dict if it's a model object
+            if hasattr(response, 'model_dump'):
+                return response.model_dump()
+            elif hasattr(response, 'to_dict'):
+                return response.to_dict()
+            elif hasattr(response, '__dict__'):
+                return self._convert_response(response)
+            return response
+            
+        except Exception as e:
+            error_msg = f"You.com search error: {str(e)}"
+            logging.error(error_msg)
+            return {"error": error_msg}
+    
+    def _convert_response(self, response) -> Dict[str, Any]:
+        """Convert SDK response object to dictionary."""
+        result = {}
+        
+        if hasattr(response, 'results'):
+            results = response.results
+            result['results'] = {}
+            if hasattr(results, 'web') and results.web:
+                result['results']['web'] = [
+                    {
+                        'url': getattr(r, 'url', ''),
+                        'title': getattr(r, 'title', ''),
+                        'description': getattr(r, 'description', ''),
+                        'snippets': getattr(r, 'snippets', []),
+                        'thumbnail_url': getattr(r, 'thumbnail_url', None),
+                        'page_age': getattr(r, 'page_age', None),
+                        'favicon_url': getattr(r, 'favicon_url', None),
+                        'authors': getattr(r, 'authors', []),
+                    }
+                    for r in results.web
+                ]
+            if hasattr(results, 'news') and results.news:
+                result['results']['news'] = [
+                    {
+                        'url': getattr(r, 'url', ''),
+                        'title': getattr(r, 'title', ''),
+                        'description': getattr(r, 'description', ''),
+                        'thumbnail_url': getattr(r, 'thumbnail_url', None),
+                        'page_age': getattr(r, 'page_age', None),
+                    }
+                    for r in results.news
+                ]
+        
+        if hasattr(response, 'metadata'):
+            meta = response.metadata
+            result['metadata'] = {
+                'query': getattr(meta, 'query', ''),
+                'search_uuid': getattr(meta, 'search_uuid', ''),
+                'latency': getattr(meta, 'latency', 0),
+            }
+        
+        return result
+    
+    def get_contents(
+        self,
+        urls: Union[str, List[str]],
+        format: str = "markdown",
+    ) -> Dict[str, Any]:
+        """Fetch content from URLs using You.com's content extraction API.
+        
+        Args:
+            urls: Single URL or list of URLs to fetch content from
+            format: Content format - "html" or "markdown" (default: markdown)
+            
+        Returns:
+            Dict containing array of results with:
+            - url: The fetched URL
+            - title: Page title
+            - html: HTML content (if format="html")
+            - markdown: Markdown content (if format="markdown")
+            - metadata: {site_name, favicon_url}
+        """
+        client, error = self._get_client()
+        if error:
+            logging.error(error)
+            return {"error": error}
+        
+        try:
+            if isinstance(urls, str):
+                urls = [urls]
+            
+            # Use the contents endpoint
+            response = client.search.contents(urls=urls, format=format)
+            
+            # Convert response
+            if hasattr(response, 'model_dump'):
+                return {"results": response.model_dump()}
+            elif isinstance(response, list):
+                return {"results": [
+                    {
+                        'url': getattr(r, 'url', ''),
+                        'title': getattr(r, 'title', ''),
+                        'html': getattr(r, 'html', None),
+                        'markdown': getattr(r, 'markdown', None),
+                        'metadata': {
+                            'site_name': getattr(getattr(r, 'metadata', None), 'site_name', None) if hasattr(r, 'metadata') else None,
+                            'favicon_url': getattr(getattr(r, 'metadata', None), 'favicon_url', None) if hasattr(r, 'metadata') else None,
+                        }
+                    }
+                    for r in response
+                ]}
+            return {"results": response}
+            
+        except Exception as e:
+            error_msg = f"You.com contents error: {str(e)}"
+            logging.error(error_msg)
+            return {"error": error_msg}
+    
+    def live_news(
+        self,
+        query: str,
+        count: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        """Search for live news using You.com's news API.
+        
+        Note: This API may require early access. Contact [email protected] for access.
+        
+        Args:
+            query: News search query
+            count: Maximum number of news results to return
+            
+        Returns:
+            Dict containing:
+            - news: {query, results: [...], metadata}
+            - Each result has: title, description, url, page_age, source_name, thumbnail
+        """
+        client, error = self._get_client()
+        if error:
+            logging.error(error)
+            return {"error": error}
+        
+        try:
+            kwargs = {"q": query}
+            if count:
+                kwargs["count"] = count
+            
+            # Try to access live news endpoint
+            if hasattr(client, 'livenews'):
+                response = client.livenews.get(**kwargs)
+            else:
+                # Fallback: use unified search with news focus
+                return self.search(query=query, count=count or 10)
+            
+            if hasattr(response, 'model_dump'):
+                return response.model_dump()
+            elif hasattr(response, 'news'):
+                news = response.news
+                return {
+                    'news': {
+                        'query': getattr(getattr(news, 'query', None), 'original', query) if hasattr(news, 'query') else query,
+                        'results': [
+                            {
+                                'title': getattr(r, 'title', ''),
+                                'description': getattr(r, 'description', ''),
+                                'url': getattr(r, 'url', ''),
+                                'page_age': getattr(r, 'page_age', None),
+                                'age': getattr(r, 'age', None),
+                                'source_name': getattr(r, 'source_name', None),
+                                'thumbnail': getattr(getattr(r, 'thumbnail', None), 'src', None) if hasattr(r, 'thumbnail') else None,
+                            }
+                            for r in getattr(news, 'results', [])
+                        ],
+                        'metadata': {
+                            'request_uuid': getattr(getattr(news, 'metadata', None), 'request_uuid', None) if hasattr(news, 'metadata') else None,
+                        }
+                    }
+                }
+            return response
+            
+        except Exception as e:
+            error_msg = f"You.com live news error: {str(e)}"
+            logging.error(error_msg)
+            return {"error": error_msg}
+    
+    def images(
+        self,
+        query: str,
+    ) -> Dict[str, Any]:
+        """Search for images using You.com's image search API.
+        
+        Note: This API may require early access. Contact [email protected] for access.
+        
+        Args:
+            query: Image search query
+            
+        Returns:
+            Dict containing:
+            - images: {results: [...]}
+            - metadata: {query, search_uuid}
+            - Each result has: title, page_url, image_url
+        """
+        client, error = self._get_client()
+        if error:
+            logging.error(error)
+            return {"error": error}
+        
+        try:
+            if hasattr(client, 'images'):
+                response = client.images.get(q=query)
+            else:
+                return {"error": "Image search not available in this SDK version"}
+            
+            if hasattr(response, 'model_dump'):
+                return response.model_dump()
+            elif hasattr(response, 'images'):
+                images = response.images
+                return {
+                    'images': {
+                        'results': [
+                            {
+                                'title': getattr(r, 'title', ''),
+                                'page_url': getattr(r, 'page_url', ''),
+                                'image_url': getattr(r, 'image_url', ''),
+                            }
+                            for r in getattr(images, 'results', [])
+                        ]
+                    },
+                    'metadata': {
+                        'query': getattr(getattr(response, 'metadata', None), 'query', query) if hasattr(response, 'metadata') else query,
+                        'search_uuid': getattr(getattr(response, 'metadata', None), 'search_uuid', None) if hasattr(response, 'metadata') else None,
+                    }
+                }
+            return response
+            
+        except Exception as e:
+            error_msg = f"You.com images error: {str(e)}"
+            logging.error(error_msg)
+            return {"error": error_msg}
+    
+    def __enter__(self):
+        """Context manager entry."""
+        return self
+    
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Context manager exit."""
+        if self._client is not None:
+            # Close client if it has a close method
+            if hasattr(self._client, 'close'):
+                self._client.close()
+            elif hasattr(self._client, '__exit__'):
+                self._client.__exit__(exc_type, exc_val, exc_tb)
+        return False
+
+
+# Standalone functions for direct import
+def ydc_search(
+    query: str,
+    count: int = 10,
+    freshness: Optional[str] = None,
+    country: Optional[str] = None,
+    language: Optional[str] = None,
+    offset: int = 0,
+    safesearch: str = "moderate",
+    livecrawl: Optional[str] = None,
+    livecrawl_formats: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Search the web using You.com's unified search API.
+    
+    This is a convenience function that creates a YouTools instance.
+    For repeated calls, consider using YouTools class directly.
+    
+    Args:
+        query: Search query (supports operators like site:, filetype:, +term, -term)
+        count: Max results per section (default 10, max 100)
+        freshness: Filter by recency - "day", "week", "month", "year"
+        country: ISO country code (e.g., "US", "GB")
+        language: BCP 47 language code (e.g., "EN", "JP")
+        offset: Pagination offset (0-9)
+        safesearch: Content filter - "off", "moderate", "strict"
+        livecrawl: Fetch full content - "web", "news", or "all"
+        livecrawl_formats: Content format - "html" or "markdown"
+        
+    Returns:
+        Dict containing search results with web and news sections
+    """
+    tools = YouTools()
+    return tools.search(
+        query=query,
+        count=count,
+        freshness=freshness,
+        country=country,
+        language=language,
+        offset=offset,
+        safesearch=safesearch,
+        livecrawl=livecrawl,
+        livecrawl_formats=livecrawl_formats,
+    )
+
+
+def ydc_contents(
+    urls: Union[str, List[str]],
+    format: str = "markdown",
+) -> Dict[str, Any]:
+    """Fetch content from URLs using You.com's content extraction API.
+    
+    Args:
+        urls: Single URL or list of URLs to fetch content from
+        format: Content format - "html" or "markdown"
+        
+    Returns:
+        Dict containing extracted content for each URL
+    """
+    tools = YouTools()
+    return tools.get_contents(urls=urls, format=format)
+
+
+def ydc_news(
+    query: str,
+    count: Optional[int] = None,
+) -> Dict[str, Any]:
+    """Search for live news using You.com's news API.
+    
+    Args:
+        query: News search query
+        count: Maximum number of results
+        
+    Returns:
+        Dict containing news results
+    """
+    tools = YouTools()
+    return tools.live_news(query=query, count=count)
+
+
+def ydc_images(
+    query: str,
+) -> Dict[str, Any]:
+    """Search for images using You.com's image search API.
+    
+    Args:
+        query: Image search query
+        
+    Returns:
+        Dict containing image results
+    """
+    tools = YouTools()
+    return tools.images(query=query)
+
+
+# Alias for simple usage
+ydc = ydc_search
+
+__all__ = ["YouTools", "ydc_search", "ydc_contents", "ydc_news", "ydc_images", "ydc"]

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "praisonai-tools"
-version = "0.2.42"
+version = "0.2.46"
 description = "Extended tools for PraisonAI Agents"
 authors = [
     {name = "Mervin Praison"}
@@ -85,6 +85,9 @@ meeting = [
     "openai>=1.12.0",
     "chromadb>=0.4.0",
 ]
+stripe = [
+    "requests>=2.28.0",
+]
 
 [project.urls]
 Homepage = "https://docs.praison.ai"

tests/unit/tools/test_fireflies_tool.py

@@ -0,0 +1,279 @@
+"""Unit tests for FirefliesTool."""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from praisonai_tools.tools.fireflies_tool import (
+    FirefliesTool,
+    list_fireflies_meetings,
+    DEFAULT_API_URL,
+)
+
+
+def _mock_response(payload, status_code=200, headers=None):
+    resp = MagicMock()
+    resp.json.return_value = payload
+    resp.status_code = status_code
+    resp.headers = headers or {}
+    return resp
+
+
+# ── Configuration ───────────────────────────────────────────────────
+
+
+class TestConfiguration:
+    def test_api_key_from_arg(self):
+        tool = FirefliesTool(api_key="ff_xyz")
+        assert tool.api_key == "ff_xyz"
+
+    def test_api_key_from_env(self):
+        with patch.dict(os.environ, {"FIREFLIES_API_KEY": "envkey"}, clear=True):
+            tool = FirefliesTool()
+            assert tool.api_key == "envkey"
+
+    def test_default_api_url(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = FirefliesTool(api_key="x")
+            assert tool.api_url == DEFAULT_API_URL
+
+    def test_api_url_override_from_env(self):
+        with patch.dict(
+            os.environ, {"FIREFLIES_API_URL": "https://proxy.local/graphql"}, clear=True
+        ):
+            tool = FirefliesTool(api_key="x")
+            assert tool.api_url == "https://proxy.local/graphql"
+
+
+# ── _graphql wiring ─────────────────────────────────────────────────
+
+
+class TestGraphQL:
+    def test_missing_key_returns_distinct_error(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = FirefliesTool()
+            assert tool._graphql("query { x }") == {"error": "FIREFLIES_API_KEY required"}
+
+    def test_sends_bearer_header(self):
+        tool = FirefliesTool(api_key="ff_xyz")
+        with patch("requests.post") as post:
+            post.return_value = _mock_response({"data": {"ok": True}})
+            tool._graphql("query { x }", {"v": 1})
+            kwargs = post.call_args.kwargs
+            assert kwargs["headers"]["Authorization"] == "Bearer ff_xyz"
+            assert kwargs["json"] == {"query": "query { x }", "variables": {"v": 1}}
+
+    def test_maps_401_to_auth_error(self):
+        tool = FirefliesTool(api_key="bad")
+        with patch("requests.post", return_value=_mock_response({}, status_code=401)):
+            result = tool._graphql("query { x }")
+        assert result["status_code"] == 401
+        assert "authentication failed" in result["error"]
+
+    def test_maps_403_to_auth_error(self):
+        tool = FirefliesTool(api_key="bad")
+        with patch("requests.post", return_value=_mock_response({}, status_code=403)):
+            result = tool._graphql("query { x }")
+        assert result["status_code"] == 403
+
+    def test_rate_limit_surfaces_retry_after(self):
+        tool = FirefliesTool(api_key="x")
+        resp = _mock_response({}, status_code=429, headers={"Retry-After": "12"})
+        with patch("requests.post", return_value=resp):
+            result = tool._graphql("query { x }")
+        assert result["status_code"] == 429
+        assert result["retry_after"] == "12"
+
+    def test_graphql_errors_are_flattened(self):
+        tool = FirefliesTool(api_key="x")
+        payload = {"errors": [{"message": "boom"}, {"message": "bad"}]}
+        with patch("requests.post", return_value=_mock_response(payload)):
+            assert tool._graphql("query { x }") == {"error": "boom; bad"}
+
+    def test_handles_request_exception(self):
+        tool = FirefliesTool(api_key="x")
+        with patch("requests.post", side_effect=RuntimeError("net")):
+            assert tool._graphql("q") == {"error": "net"}
+
+    def test_rejects_non_https_url(self):
+        tool = FirefliesTool(api_key="x", api_url="http://insecure.local/graphql")
+        with patch("requests.post") as post:
+            result = tool._graphql("query { x }")
+        post.assert_not_called()
+        assert result == {"error": "FIREFLIES_API_URL must use HTTPS"}
+
+    def test_rate_limit_from_graphql_payload(self):
+        tool = FirefliesTool(api_key="x")
+        payload = {
+            "errors": [{"extensions": {"metadata": {"retryAfter": 30}}}]
+        }
+        resp = _mock_response(payload, status_code=429)
+        with patch("requests.post", return_value=resp):
+            result = tool._graphql("query { x }")
+        assert result["status_code"] == 429
+        assert result["retry_after"] == "30"
+
+    def test_generic_http_error_surfaces_status(self):
+        tool = FirefliesTool(api_key="x")
+        resp = _mock_response({"data": {"transcripts": None}}, status_code=500)
+        with patch("requests.post", return_value=resp):
+            result = tool._graphql("query { x }")
+        assert result["status_code"] == 500
+        assert "HTTP 500" in result["error"]
+
+
+# ── list_recent_meetings ────────────────────────────────────────────
+
+
+class TestListRecentMeetings:
+    def test_returns_normalised_meetings(self):
+        tool = FirefliesTool(api_key="x")
+        payload = {
+            "data": {
+                "transcripts": [
+                    {
+                        "id": "m1",
+                        "title": "Standup",
+                        "date": "2026-01-01",
+                        "duration": 15,
+                        "organizer_email": "[email protected]",
+                    }
+                ]
+            }
+        }
+        with patch("requests.post", return_value=_mock_response(payload)):
+            meetings = tool.list_recent_meetings(limit=5)
+        assert len(meetings) == 1
+        assert meetings[0]["title"] == "Standup"
+        assert meetings[0]["organizer_email"] == "[email protected]"
+
+    def test_after_cursor_maps_to_skip(self):
+        tool = FirefliesTool(api_key="x")
+        with patch(
+            "requests.post", return_value=_mock_response({"data": {"transcripts": []}})
+        ) as post:
+            tool.list_recent_meetings(limit=5, after="10")
+        sent = post.call_args.kwargs["json"]["variables"]
+        assert sent == {"limit": 5, "skip": 10}
+
+    def test_invalid_after_cursor(self):
+        tool = FirefliesTool(api_key="x")
+        assert tool.list_recent_meetings(after="notint") == [
+            {"error": "after must be an integer cursor"}
+        ]
+
+    def test_propagates_error(self):
+        tool = FirefliesTool(api_key="x")
+        with patch("requests.post", side_effect=RuntimeError("net")):
+            assert tool.list_recent_meetings() == [{"error": "net"}]
+
+    def test_limit_clamped_to_50(self):
+        tool = FirefliesTool(api_key="x")
+        with patch(
+            "requests.post", return_value=_mock_response({"data": {"transcripts": []}})
+        ) as post:
+            tool.list_recent_meetings(limit=500)
+        assert post.call_args.kwargs["json"]["variables"]["limit"] == 50
+
+
+# ── get_transcript ──────────────────────────────────────────────────
+
+
+class TestGetTranscript:
+    def test_requires_meeting_id(self):
+        tool = FirefliesTool(api_key="x")
+        assert tool.get_transcript(meeting_id="") == {"error": "meeting_id required"}
+
+    def test_returns_normalised_transcript(self):
+        tool = FirefliesTool(api_key="x")
+        payload = {
+            "data": {
+                "transcript": {
+                    "id": "m1",
+                    "title": "Standup",
+                    "date": "2026-01-01",
+                    "duration": 15,
+                    "organizer_email": "[email protected]",
+                    "summary": {"overview": "We talked."},
+                    "sentences": [
+                        {"text": "Hi", "speaker_name": "A", "start_time": 0.0}
+                    ],
+                }
+            }
+        }
+        with patch("requests.post", return_value=_mock_response(payload)):
+            transcript = tool.get_transcript(meeting_id="m1")
+        assert transcript["summary"] == "We talked."
+        assert transcript["sentence_count"] == 1
+
+    def test_not_found(self):
+        tool = FirefliesTool(api_key="x")
+        with patch(
+            "requests.post",
+            return_value=_mock_response({"data": {"transcript": None}}),
+        ):
+            assert tool.get_transcript(meeting_id="missing") == {
+                "error": "transcript not found"
+            }
+
+
+# ── search ──────────────────────────────────────────────────────────
+
+
+class TestSearch:
+    def test_requires_keywords(self):
+        tool = FirefliesTool(api_key="x")
+        assert tool.search(keywords="") == [{"error": "keywords required"}]
+
+    def test_returns_results(self):
+        tool = FirefliesTool(api_key="x")
+        payload = {
+            "data": {
+                "transcripts": [
+                    {
+                        "id": "m1",
+                        "title": "Roadmap",
+                        "date": "2026-01-01",
+                        "organizer_email": "[email protected]",
+                    }
+                ]
+            }
+        }
+        with patch("requests.post", return_value=_mock_response(payload)) as post:
+            results = tool.search(keywords="roadmap", limit=3)
+        sent = post.call_args.kwargs["json"]["variables"]
+        assert sent == {"keyword": "roadmap", "limit": 3}
+        assert results[0]["title"] == "Roadmap"
+
+
+# ── run() dispatcher ────────────────────────────────────────────────
+
+
+class TestRunDispatcher:
+    def test_unknown_action(self):
+        tool = FirefliesTool(api_key="x")
+        assert tool.run(action="bogus") == {"error": "Unknown action: bogus"}
+
+    def test_routes_get_transcript(self):
+        tool = FirefliesTool(api_key="x")
+        with patch.object(tool, "get_transcript", return_value={"ok": True}) as m:
+            out = tool.run(action="get-transcript", meeting_id="m1")
+        m.assert_called_once_with(meeting_id="m1")
+        assert out == {"ok": True}
+
+    def test_routes_search(self):
+        tool = FirefliesTool(api_key="x")
+        with patch.object(tool, "search", return_value=[]) as m:
+            tool.run(action="search", keywords="hi", limit=5)
+        m.assert_called_once_with(keywords="hi", limit=5)
+
+
+# ── Module-level helper ─────────────────────────────────────────────
+
+
+class TestListFirefliesMeetingsHelper:
+    def test_delegates_to_tool(self):
+        with patch.object(
+            FirefliesTool, "list_recent_meetings", return_value=["x"]
+        ) as m:
+            assert list_fireflies_meetings(limit=5) == ["x"]
+        m.assert_called_once_with(limit=5)

tests/unit/tools/test_keenable_search_tool.py

@@ -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"

tests/unit/tools/test_reddit_tool.py

@@ -0,0 +1,176 @@
+"""Unit tests for RedditTool."""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from praisonai_tools.tools.reddit_tool import RedditTool
+
+
+# ── Import + schema (CI smoke) ──────────────────────────────────────
+
+
+class TestImportAndSchema:
+    def test_import_via_package(self):
+        from praisonai_tools import RedditTool as PkgRedditTool
+
+        assert PkgRedditTool is RedditTool
+
+    def test_schema_generation(self):
+        tool = RedditTool(client_id="cid", client_secret="secret")
+        schema = tool.get_schema()
+        assert schema["function"]["name"] == "reddit"
+        assert "description" in schema["function"]
+        assert "action" in schema["function"]["parameters"]["properties"]
+
+
+# ── User-Agent ──────────────────────────────────────────────────────
+
+
+class TestUserAgent:
+    def test_default_user_agent_references_app_id(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="myappid", client_secret="secret")
+        assert "myappid" in tool.user_agent
+
+    def test_explicit_user_agent_wins(self):
+        tool = RedditTool(client_id="c", client_secret="s", user_agent="custom:ua:1.0")
+        assert tool.user_agent == "custom:ua:1.0"
+
+    def test_env_user_agent_used(self):
+        with patch.dict(os.environ, {"REDDIT_USER_AGENT": "env:ua:1.0"}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.user_agent == "env:ua:1.0"
+
+
+# ── Write flag gating ───────────────────────────────────────────────
+
+
+class TestWriteGating:
+    def test_write_disabled_by_default(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.allow_write is False
+
+    def test_env_enables_write(self):
+        with patch.dict(os.environ, {"REDDIT_ALLOW_WRITE": "1"}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.allow_write is True
+
+    def test_post_comment_blocked_when_write_disabled(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        result = tool.post_comment(post_id="abc", body="hi")
+        assert "REDDIT_ALLOW_WRITE" in result["error"]
+
+    def test_submit_post_blocked_when_write_disabled(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        result = tool.submit_post(subreddit="test", title="T", body="B")
+        assert "REDDIT_ALLOW_WRITE" in result["error"]
+
+    def test_write_requires_account_credentials(self):
+        tool = RedditTool(client_id="c", client_secret="s", allow_write=True)
+        result = tool.post_comment(post_id="abc", body="hi")
+        assert "REDDIT_USERNAME" in result["error"]
+
+
+# ── Write actions (mocked praw) ─────────────────────────────────────
+
+
+def _writable_tool():
+    return RedditTool(
+        client_id="c",
+        client_secret="s",
+        username="u",
+        password="p",
+        allow_write=True,
+    )
+
+
+class TestPostComment:
+    def test_requires_post_id(self):
+        tool = _writable_tool()
+        assert tool.post_comment(post_id="", body="hi") == {"error": "post_id is required"}
+
+    def test_requires_body(self):
+        tool = _writable_tool()
+        assert tool.post_comment(post_id="abc", body="") == {"error": "body is required"}
+
+    def test_success(self):
+        tool = _writable_tool()
+        comment = MagicMock()
+        comment.id = "c1"
+        comment.permalink = "/r/test/comments/abc/x/c1/"
+        submission = MagicMock()
+        submission.reply.return_value = comment
+        reddit = MagicMock()
+        reddit.submission.return_value = submission
+        with patch.object(RedditTool, "reddit", new=reddit):
+            result = tool.post_comment(post_id="abc", body="hello")
+        submission.reply.assert_called_once_with("hello")
+        assert result["success"] is True
+        assert result["id"] == "c1"
+
+
+class TestSubmitPost:
+    def test_requires_subreddit(self):
+        tool = _writable_tool()
+        assert tool.submit_post(subreddit="", title="T") == {"error": "subreddit is required"}
+
+    def test_requires_title(self):
+        tool = _writable_tool()
+        assert tool.submit_post(subreddit="test", title="") == {"error": "title is required"}
+
+    def test_body_and_url_mutually_exclusive(self):
+        tool = _writable_tool()
+        result = tool.submit_post(subreddit="test", title="T", body="B", url="http://x")
+        assert "not both" in result["error"]
+
+    def test_selftext_success(self):
+        tool = _writable_tool()
+        submission = MagicMock()
+        submission.id = "p1"
+        submission.permalink = "/r/test/comments/p1/"
+        sub = MagicMock()
+        sub.submit.return_value = submission
+        reddit = MagicMock()
+        reddit.subreddit.return_value = sub
+        with patch.object(RedditTool, "reddit", new=reddit):
+            result = tool.submit_post(subreddit="test", title="T", body="B")
+        sub.submit.assert_called_once_with("T", selftext="B")
+        assert result["id"] == "p1"
+
+    def test_link_success(self):
+        tool = _writable_tool()
+        submission = MagicMock()
+        submission.id = "p2"
+        submission.permalink = "/r/test/comments/p2/"
+        sub = MagicMock()
+        sub.submit.return_value = submission
+        reddit = MagicMock()
+        reddit.subreddit.return_value = sub
+        with patch.object(RedditTool, "reddit", new=reddit):
+            result = tool.submit_post(subreddit="test", title="T", url="http://x")
+        sub.submit.assert_called_once_with("T", url="http://x")
+        assert result["id"] == "p2"
+
+
+# ── run() dispatcher ────────────────────────────────────────────────
+
+
+class TestRunDispatcher:
+    def test_unknown_action(self):
+        tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.run(action="bogus") == {"error": "Unknown action: bogus"}
+
+    def test_routes_post_comment(self):
+        tool = RedditTool(client_id="c", client_secret="s")
+        with patch.object(tool, "post_comment", return_value={"ok": True}) as m:
+            tool.run(action="post_comment", post_id="abc", body="hi")
+        m.assert_called_once_with(post_id="abc", body="hi")
+
+    def test_routes_submit_post(self):
+        tool = RedditTool(client_id="c", client_secret="s")
+        with patch.object(tool, "submit_post", return_value={"ok": True}) as m:
+            tool.run(action="submit-post", subreddit="test", title="T", body="B")
+        m.assert_called_once_with(subreddit="test", title="T", body="B", url=None)

tests/unit/tools/test_stripe_tool.py

@@ -0,0 +1,248 @@
+"""Unit tests for StripeTool."""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from praisonai_tools.tools.stripe_tool import StripeTool, get_stripe_customer
+
+
+def _mock_response(payload, status_code=200):
+    resp = MagicMock()
+    resp.json.return_value = payload
+    resp.status_code = status_code
+    return resp
+
+
+# ── Key handling / mode banner ──────────────────────────────────────
+
+
+class TestKeyAndMode:
+    def test_env_secret_key_fallback(self):
+        with patch.dict(os.environ, {"STRIPE_SECRET_KEY": "sk_test_x"}, clear=True):
+            tool = StripeTool()
+            assert tool.api_key == "sk_test_x"
+            assert tool._mode() == "test"
+
+    def test_stripe_api_key_fallback(self):
+        with patch.dict(os.environ, {"STRIPE_API_KEY": "sk_live_x"}, clear=True):
+            tool = StripeTool()
+            assert tool.api_key == "sk_live_x"
+            assert tool._mode() == "live"
+
+    def test_restricted_key_prefixes(self):
+        assert StripeTool(api_key="rk_test_x")._mode() == "test"
+        assert StripeTool(api_key="rk_live_x")._mode() == "live"
+
+    def test_unknown_mode(self):
+        assert StripeTool(api_key="weird")._mode() == "unknown"
+
+    def test_missing_key_returns_error(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = StripeTool()
+            assert tool.get_customer(customer_id="cus_1") == {
+                "error": "STRIPE_SECRET_KEY (or STRIPE_API_KEY) is required"
+            }
+
+
+# ── get_customer ────────────────────────────────────────────────────
+
+
+class TestGetCustomer:
+    def test_requires_customer_id(self):
+        tool = StripeTool(api_key="sk_test_x")
+        assert tool.get_customer(customer_id="") == {"error": "customer_id is required"}
+
+    def test_returns_normalised_customer_with_banner(self):
+        tool = StripeTool(api_key="sk_test_x")
+        payload = {
+            "id": "cus_1",
+            "email": "[email protected]",
+            "name": "Alice",
+            "created": 123,
+            "currency": "usd",
+            "delinquent": False,
+        }
+        with patch("requests.request", return_value=_mock_response(payload)) as req:
+            result = tool.get_customer(customer_id="cus_1")
+
+        args, kwargs = req.call_args
+        assert args[0] == "get"
+        assert args[1].endswith("/customers/cus_1")
+        assert kwargs["headers"]["Authorization"] == "Bearer sk_test_x"
+        assert result["mode"] == "test"
+        assert result["id"] == "cus_1"
+        assert result["email"] == "[email protected]"
+
+    def test_propagates_stripe_error(self):
+        tool = StripeTool(api_key="sk_test_x")
+        payload = {"error": {"message": "No such customer", "type": "invalid_request_error"}}
+        with patch("requests.request", return_value=_mock_response(payload)):
+            assert tool.get_customer(customer_id="cus_x") == {"error": "No such customer"}
+
+    def test_handles_request_exception(self):
+        import requests
+
+        tool = StripeTool(api_key="sk_test_x")
+        with patch(
+            "requests.request",
+            side_effect=requests.exceptions.RequestException("boom"),
+        ):
+            result = tool.get_customer(customer_id="cus_1")
+        assert result["error"].startswith("API request failed")
+
+    def test_non_2xx_without_error_object_is_failure(self):
+        # A non-2xx response whose body lacks Stripe's usual ``error`` object
+        # must not be treated as a successful payload.
+        tool = StripeTool(api_key="sk_test_x")
+        with patch(
+            "requests.request",
+            return_value=_mock_response({"unexpected": "shape"}, status_code=502),
+        ):
+            result = tool.get_customer(customer_id="cus_1")
+        assert result == {"error": "Stripe API returned HTTP 502"}
+
+
+# ── get_payment_intent ──────────────────────────────────────────────
+
+
+class TestGetPaymentIntent:
+    def test_requires_id(self):
+        tool = StripeTool(api_key="sk_test_x")
+        assert tool.get_payment_intent(payment_intent_id="") == {
+            "error": "payment_intent_id is required"
+        }
+
+    def test_returns_normalised_intent(self):
+        tool = StripeTool(api_key="sk_test_x")
+        payload = {
+            "id": "pi_1",
+            "amount": 500,
+            "currency": "usd",
+            "status": "succeeded",
+            "customer": "cus_1",
+            "created": 1,
+        }
+        with patch("requests.request", return_value=_mock_response(payload)):
+            result = tool.get_payment_intent(payment_intent_id="pi_1")
+        assert result["status"] == "succeeded"
+        assert result["amount"] == 500
+        assert result["mode"] == "test"
+
+
+# ── list_payment_links ──────────────────────────────────────────────
+
+
+class TestListPaymentLinks:
+    def test_returns_links(self):
+        tool = StripeTool(api_key="sk_test_x")
+        payload = {
+            "data": [
+                {"id": "plink_1", "url": "https://pay/1", "active": True, "metadata": {}},
+                {"id": "plink_2", "url": "https://pay/2", "active": False, "metadata": {}},
+            ]
+        }
+        with patch("requests.request", return_value=_mock_response(payload)):
+            result = tool.list_payment_links(limit=2)
+        assert result["mode"] == "test"
+        assert len(result["payment_links"]) == 2
+        assert result["payment_links"][0]["id"] == "plink_1"
+
+    def test_clamps_limit(self):
+        tool = StripeTool(api_key="sk_test_x")
+        with patch("requests.request", return_value=_mock_response({"data": []})) as req:
+            tool.list_payment_links(limit=999)
+        assert "limit=100" in req.call_args[0][1]
+
+    def test_propagates_error(self):
+        tool = StripeTool(api_key="sk_test_x")
+        payload = {"error": {"message": "bad"}}
+        with patch("requests.request", return_value=_mock_response(payload)):
+            assert tool.list_payment_links() == {"error": "bad"}
+
+
+# ── create_payment_link (guardrails) ────────────────────────────────
+
+
+class TestCreatePaymentLink:
+    def test_requires_price_id(self):
+        tool = StripeTool(api_key="sk_test_x")
+        assert tool.create_payment_link(
+            price_id="", correlation_id="c1"
+        ) == {"error": "price_id is required"}
+
+    def test_requires_correlation_id(self):
+        tool = StripeTool(api_key="sk_test_x")
+        assert tool.create_payment_link(price_id="price_1") == {
+            "error": "correlation_id is required (stored as metadata)"
+        }
+
+    def test_rejects_bad_quantity(self):
+        tool = StripeTool(api_key="sk_test_x")
+        assert tool.create_payment_link(
+            price_id="price_1", correlation_id="c1", quantity=0
+        ) == {"error": "quantity must be >= 1"}
+
+    def test_success_sends_metadata_and_idempotency_key(self):
+        tool = StripeTool(api_key="sk_test_x")
+        payload = {
+            "id": "plink_1",
+            "url": "https://pay/1",
+            "active": True,
+            "metadata": {"correlation_id": "order-42"},
+        }
+        with patch("requests.request", return_value=_mock_response(payload)) as req:
+            result = tool.create_payment_link(
+                price_id="price_1",
+                quantity=2,
+                correlation_id="order-42",
+                idempotency_key="idem-1",
+            )
+
+        args, kwargs = req.call_args
+        assert args[0] == "post"
+        assert args[1].endswith("/payment_links")
+        assert kwargs["data"]["line_items[0][price]"] == "price_1"
+        assert kwargs["data"]["line_items[0][quantity]"] == 2
+        assert kwargs["data"]["metadata[correlation_id]"] == "order-42"
+        assert kwargs["headers"]["Idempotency-Key"] == "idem-1"
+        assert result["id"] == "plink_1"
+        assert result["mode"] == "test"
+
+
+# ── run() dispatcher ────────────────────────────────────────────────
+
+
+class TestRunDispatcher:
+    def test_unknown_action(self):
+        tool = StripeTool(api_key="sk_test_x")
+        assert tool.run(action="bogus") == {"error": "Unknown action: bogus"}
+
+    def test_routes_get_customer(self):
+        tool = StripeTool(api_key="sk_test_x")
+        with patch.object(tool, "get_customer", return_value={"ok": True}) as m:
+            tool.run(action="get_customer", customer_id="cus_1")
+        m.assert_called_once_with(customer_id="cus_1")
+
+    def test_routes_create_payment_link(self):
+        tool = StripeTool(api_key="sk_test_x")
+        with patch.object(tool, "create_payment_link", return_value={"ok": True}) as m:
+            tool.run(
+                action="create-payment-link",
+                price_id="price_1",
+                quantity=1,
+                correlation_id="c1",
+                idempotency_key="i1",
+            )
+        m.assert_called_once_with(
+            price_id="price_1", quantity=1, correlation_id="c1", idempotency_key="i1"
+        )
+
+
+# ── Module-level helper ─────────────────────────────────────────────
+
+
+class TestHelper:
+    def test_delegates_to_tool(self):
+        with patch.object(StripeTool, "get_customer", return_value={"id": "cus_1"}) as m:
+            assert get_stripe_customer("cus_1", api_key="sk_test_x") == {"id": "cus_1"}
+        m.assert_called_once_with(customer_id="cus_1")

tests/unit/tools/test_youdotcom_tool.py

@@ -0,0 +1,81 @@
+"""Unit tests for You.com tools (no network, no API key required)."""
+
+from unittest.mock import MagicMock, patch
+
+from praisonai_tools.tools.youdotcom_tool import (
+    YouTools,
+    ydc,
+    ydc_search,
+    ydc_contents,
+    ydc_news,
+    ydc_images,
+)
+
+
+_FIND_SPEC = "praisonai_tools.tools.youdotcom_tool.util.find_spec"
+
+
+def test_missing_api_key_returns_error(monkeypatch):
+    monkeypatch.delenv("YDC_API_KEY", raising=False)
+    with patch(_FIND_SPEC, return_value=object()):
+        for fn, arg in (
+            (ydc_search, "q"),
+            (ydc_contents, "https://example.com"),
+            (ydc_news, "q"),
+            (ydc_images, "q"),
+        ):
+            result = fn(arg)
+            assert result == {
+                "error": "YDC_API_KEY environment variable is not set. Please set it to use You.com tools."
+            }
+
+
+def test_explicit_api_key_used_without_env(monkeypatch):
+    monkeypatch.delenv("YDC_API_KEY", raising=False)
+    fake_module = MagicMock()
+    with patch(_FIND_SPEC, return_value=object()), patch.dict(
+        "sys.modules", {"youdotcom": fake_module}
+    ):
+        client, error = YouTools(api_key="k")._get_client()
+
+    assert error is None
+    assert client is fake_module.You.return_value
+    fake_module.You.assert_called_once_with(api_key_auth="k")
+
+
+def test_ydc_alias():
+    assert ydc is ydc_search
+
+
+def test_search_uses_client_and_returns_dict():
+    response = MagicMock()
+    response.model_dump.return_value = {"results": {"web": [], "news": []}}
+    client = MagicMock()
+    client.search.unified.return_value = response
+
+    with patch.object(YouTools, "_get_client", return_value=(client, None)):
+        result = YouTools().search("ai", count=3, freshness="week")
+
+    assert result == {"results": {"web": [], "news": []}}
+    kwargs = client.search.unified.call_args.kwargs
+    assert kwargs["query"] == "ai"
+    assert kwargs["count"] == 3
+    assert kwargs["freshness"] == "week"
+
+
+def test_search_exception_is_reported_as_error():
+    client = MagicMock()
+    client.search.unified.side_effect = RuntimeError("boom")
+
+    with patch.object(YouTools, "_get_client", return_value=(client, None)):
+        result = YouTools().search("ai")
+
+    assert "boom" in result["error"]
+
+
+def test_discovered_via_package():
+    import praisonai_tools.tools as tools_pkg
+
+    assert tools_pkg.YouTools is YouTools
+    assert tools_pkg.ydc_search is ydc_search
+    assert tools_pkg.ydc is ydc_search