Using newer version of ddgs which returns a generator and return JSON

tools/web_search.py

@@ -1,27 +1,56 @@
 from typing import Any, Optional
 from smolagents.tools import Tool
-import duckduckgo_search
+from ddgs import DDGS
 
 class DuckDuckGoSearchTool(Tool):
     name = "web_search"
-    description = "Performs a duckduckgo web search based on your query (think a Google search) then returns the top search results."
-    inputs = {'query': {'type': 'string', 'description': 'The search query to perform.'}}
+    description = "Performs a duckduckgo web search based on your query and returns the top results."
+    inputs: Dict[str, Dict[str, Any]] = {
+        'query': {'type': 'string', 'description': 'The search query to perform.'}
+    }
     output_type = "string"
 
-    def __init__(self, max_results=10, **kwargs):
+    def __init__(self, max_results: int = 10, **kwargs: Any) -> None:
         super().__init__()
         self.max_results = max_results
-        try:
-            from duckduckgo_search import DDGS
-        except ImportError as e:
-            raise ImportError(
-                "You must install package `duckduckgo_search` to run this tool: for instance run `pip install duckduckgo-search`."
-            ) from e
         self.ddgs = DDGS(**kwargs)
 
     def forward(self, query: str) -> str:
-        results = self.ddgs.text(query, max_results=self.max_results)
-        if len(results) == 0:
-            raise Exception("No results found! Try a less restrictive/shorter query.")
-        postprocessed_results = [f"[{result['title']}]({result['href']})\n{result['body']}" for result in results]
-        return "## Search Results\n\n" + "\n\n".join(postprocessed_results)
+    q = (query or "").strip()
+    if not q:
+        raise ValueError("Query must be a non-empty string.")
+
+    # 1) ddgs.text: correct args + materialize generator
+    items = list(
+        self.ddgs.text(
+            keywords=q,
+            region="wt-wt",
+            safesearch="moderate",
+            timelimit=None,
+            max_results=self.max_results,
+        )
+    )
+
+    # 3) Better emptiness check + error type
+    if not items:
+        raise RuntimeError("No results found. Try a broader/simpler query.")
+
+    results: List[Dict[str, str]] = []
+        for it in items:
+            href = (it.get("href") or "").strip()
+            if not href:
+                continue  # skip malformed entries without a URL
+            results.append(
+                {
+                    "title": str(it.get("title", "")),
+                    "url": href,                       # normalized key name
+                    "snippet": str(it.get("body", "")),
+                }
+            )
+  
+    
+    if not results:
+            raise RuntimeError("Results were returned but none contained valid URLs.")
+
+        return results
+