search
Search DuckDuckGo to find web information and return formatted results for queries, supporting content fetching and parsing through the MCP server.
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 primary handler function for the 'search' MCP tool. It is registered via the @mcp.tool() decorator and orchestrates the search by calling the DuckDuckGoSearcher and formatting results.@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)}"
- Core helper method implementing the DuckDuckGo search logic: rate limiting, HTTP POST to DuckDuckGo HTML endpoint, HTML parsing with BeautifulSoup, result extraction and cleaning.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 []
- Dataclass schema defining the structure of individual search results returned by the search implementation.@dataclass class SearchResult: title: str link: str snippet: str position: int
- Helper function to format the list of SearchResult objects into a human-readable string optimized for LLM processing.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)
- src/duckduckgo_mcp_server/server.py:209-212 (registration)Initialization of the FastMCP server instance named 'ddg-search' and the searcher/fetcher instances used by the tools.mcp = FastMCP("ddg-search") searcher = DuckDuckGoSearcher() fetcher = WebContentFetcher()