search
Search DuckDuckGo to find web information using queries and return formatted results for integration with language models.
Instructions
Search DuckDuckGo and return formatted results.
Args:
query: The search query string
max_results: Maximum number of results to return (default: 10)
ctx: MCP context for logging
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| max_results | No |
Implementation Reference
- The main handler function for the 'search' tool, registered via the @mcp.tool() decorator. It coordinates the search execution and result formatting.@mcp.tool() async def search(query: str, ctx: Context, max_results: int = 10) -> str: """ Search DuckDuckGo and return formatted results. Args: query: The search query string max_results: Maximum number of results to return (default: 10) ctx: MCP context for logging """ try: results = await searcher.search(query, ctx, max_results) return searcher.format_results_for_llm(results) except Exception as e: traceback.print_exc(file=sys.stderr) return f"An error occurred while searching: {str(e)}"
- Dataclass defining the schema/structure for individual search results returned by the internal search logic.@dataclass class SearchResult: title: str link: str snippet: str position: int
- Core helper method in DuckDuckGoSearcher class that performs the actual web scraping of DuckDuckGo, including rate limiting, parsing, and result extraction.async def search( self, query: str, ctx: Context, max_results: int = 10 ) -> List[SearchResult]: try: # Apply rate limiting await self.rate_limiter.acquire() # Create form data for POST request data = { "q": query, "b": "", "kl": "", } await ctx.info(f"Searching DuckDuckGo for: {query}") async with httpx.AsyncClient() as client: response = await client.post( self.BASE_URL, data=data, headers=self.HEADERS, timeout=30.0 ) response.raise_for_status() # Parse HTML response soup = BeautifulSoup(response.text, "html.parser") if not soup: await ctx.error("Failed to parse HTML response") return [] results = [] for result in soup.select(".result"): title_elem = result.select_one(".result__title") if not title_elem: continue link_elem = title_elem.find("a") if not link_elem: continue title = link_elem.get_text(strip=True) link = link_elem.get("href", "") # Skip ad results if "y.js" in link: continue # Clean up DuckDuckGo redirect URLs if link.startswith("//duckduckgo.com/l/?uddg="): link = urllib.parse.unquote(link.split("uddg=")[1].split("&")[0]) snippet_elem = result.select_one(".result__snippet") snippet = snippet_elem.get_text(strip=True) if snippet_elem else "" results.append( SearchResult( title=title, link=link, snippet=snippet, position=len(results) + 1, ) ) if len(results) >= max_results: break await ctx.info(f"Successfully found {len(results)} results") return results except httpx.TimeoutException: await ctx.error("Search request timed out") return [] except httpx.HTTPError as e: await ctx.error(f"HTTP error occurred: {str(e)}") return [] except Exception as e: await ctx.error(f"Unexpected error during search: {str(e)}") traceback.print_exc(file=sys.stderr) return []
- Helper method that formats the list of SearchResult objects into a human-readable string optimized for LLM consumption.def format_results_for_llm(self, results: List[SearchResult]) -> str: """Format results in a natural language style that's easier for LLMs to process""" if not results: return "No results were found for your search query. This could be due to DuckDuckGo's bot detection or the query returned no matches. Please try rephrasing your search or try again in a few minutes." output = [] output.append(f"Found {len(results)} search results:\n") for result in results: output.append(f"{result.position}. {result.title}") output.append(f" URL: {result.link}") output.append(f" Summary: {result.snippet}") output.append("") # Empty line between results return "\n".join(output)