MCP DuckDuckGo Search Plugin

Instructions

Implementation Reference

• mcp_duckduckgo/tools.py:45-113 (handler)

  Main handler function for the 'web_search' tool. Includes input schema via Pydantic Field annotations and executes the search logic using DuckDuckGo via the search_web helper.
  async def web_search( query: str = Field(..., description="Search query", max_length=400), max_results: int = Field( 10, description="Maximum number of results to return (1-20)", ge=1, le=20 ), ctx: Context = Field(default_factory=Context), ) -> Dict[str, Any]: """ Search the web using DuckDuckGo. Returns a list of search results with titles, URLs, descriptions, and domains. """ logger.info("Searching for: '%s' (max %d results)", query, max_results) try: # Get HTTP client from context http_client = None close_client = False # Try to get HTTP client from lifespan context if ( hasattr(ctx, "lifespan_context") and ctx.lifespan_context and "http_client" in ctx.lifespan_context ): logger.info("Using HTTP client from lifespan context") http_client = ctx.lifespan_context["http_client"] else: # Create a new HTTP client logger.info("Creating new HTTP client") http_client = httpx.AsyncClient(timeout=10.0) close_client = True try: # Perform the search results = await search_web(query, http_client, max_results) # Convert to dict format search_results = [ { "title": result.title, "url": result.url, "description": result.description, "domain": result.domain, } for result in results ] return { "query": query, "results": search_results, "total_results": len(search_results), "status": "success", } finally: if close_client: await http_client.aclose() except (httpx.RequestError, httpx.HTTPError, ValueError) as e: logger.error("Search failed: %s", e) return { "query": query, "results": [], "total_results": 0, "status": "error", "error": str(e), }
• mcp_duckduckgo/server.py:42-49 (registration)

  Registers the search tools (including web_search) by calling register_search_tools on the FastMCP server instance during server creation.
  def create_mcp_server() -> FastMCP: """Create and return a FastMCP server instance with proper tool registration.""" server = FastMCP("DuckDuckGo Search", lifespan=app_lifespan) # Register tools directly with the server instance register_search_tools(server) return server
• mcp_duckduckgo/search.py:254-293 (helper)

  Core helper function that implements the DuckDuckGo web search logic, querying both the instant API and HTML search page, parsing results, and deduplicating them. Called by the web_search handler.
  async def search_web( query: str, http_client: httpx.AsyncClient, count: int = 10 ) -> List[SearchResult]: """ Main search function that tries multiple methods. Args: query: Search query string http_client: HTTP client to use for requests count: Maximum number of results to return Returns: List of unique SearchResult objects from both instant answers and HTML search """ logger.info("Searching for: '%s' (max %d results)", query, count) # Try instant answers first instant_results = await search_duckduckgo_instant(query, http_client) logger.info("Instant answers found %d results", len(instant_results)) # Always try HTML search for more comprehensive results html_results = await search_duckduckgo_html(query, http_client, count) logger.info("HTML search found %d results", len(html_results)) # Combine and deduplicate all_results = instant_results + html_results # Remove duplicates based on URL seen_urls = set() unique_results = [] for result in all_results: if result.url and result.url not in seen_urls and result.url.startswith("http"): seen_urls.add(result.url) unique_results.append(result) if len(unique_results) >= count: break logger.info("Returning %d unique valid results", len(unique_results)) return unique_results
• mcp_duckduckgo/search.py:53-59 (schema)

  Dataclass defining the structure of individual search results used in the output of web_search.
  @dataclass class SearchResult: title: str url: str description: str domain: str = ""
• mcp_duckduckgo/search.py:61-77 (helper)

  Utility function to extract the domain from a URL, used in search result processing.
  def extract_domain(url: str) -> str: """ Extract domain from URL. Args: url: URL string to extract domain from Returns: Lowercase domain name or empty string if parsing fails """ try: parsed = urllib.parse.urlparse(url) return parsed.netloc.lower() except Exception as e: logger.debug("Failed to extract domain from URL %s: %s", url, e) return ""