DuckDuckGo MCP Server

Instructions

Implementation Reference

• src/duckduckgo_mcp_server/server.py:214-229 (handler)

  The primary handler function for the 'search' MCP tool. It orchestrates the search using DuckDuckGoSearcher, formats results for LLM consumption, and handles errors.
  @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)}"
• src/duckduckgo_mcp_server/server.py:44-146 (helper)

  Core helper class implementing DuckDuckGo search functionality, including rate limiting, HTTP requests, HTML parsing with BeautifulSoup, result extraction, and result formatting.
  class DuckDuckGoSearcher: BASE_URL = "https://html.duckduckgo.com/html" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } def __init__(self): self.rate_limiter = RateLimiter() 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) 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 []
• src/duckduckgo_mcp_server/server.py:15-21 (schema)

  Dataclass schema defining the structure of a single search result used internally by the search tool.
  @dataclass class SearchResult: title: str link: str snippet: str position: int
• src/duckduckgo_mcp_server/server.py:209-212 (registration)

  Initialization of the FastMCP server instance named 'ddg-search' and creation of helper instances (searcher and fetcher) used by the registered tools.
  mcp = FastMCP("ddg-search") searcher = DuckDuckGoSearcher() fetcher = WebContentFetcher()
• src/duckduckgo_mcp_server/server.py:23-42 (helper)

  Rate limiting utility class used by DuckDuckGoSearcher to prevent excessive requests to DuckDuckGo.
  class RateLimiter: def __init__(self, requests_per_minute: int = 30): self.requests_per_minute = requests_per_minute self.requests = [] async def acquire(self): now = datetime.now() # Remove requests older than 1 minute self.requests = [ req for req in self.requests if now - req < timedelta(minutes=1) ] if len(self.requests) >= self.requests_per_minute: # Wait until we can make another request wait_time = 60 - (now - self.requests[0]).total_seconds() if wait_time > 0: await asyncio.sleep(wait_time) self.requests.append(now)