rag_search_ddgs

Instructions

Implementation Reference

• src/mcp_local_rag/main.py:26-73 (handler)

  The main handler function for the 'rag_search_ddgs' MCP tool. Uses DuckDuckGo (DDGS) to perform a web search, scores results by semantic similarity using text embeddings, fetches content from the top-ranked URLs via a thread pool, and returns markdown-formatted content as a Dict.

  def rag_search_ddgs(query: str, num_results:int=10, top_k:int=5, include_urls:bool=True) -> Dict:
      """
      Search the web for a given query using DuckDuckGo. Returns context to the LLM
      with RAG-like similarity scoring to prioritize the most relevant results.
      
      This tool fetches web search results, scores them by semantic similarity to the query
      using text embeddings, and returns the top-ranked content as markdown text.
      
      Args:
          query (str): The search query. Use natural language questions or keywords.
                      Example: "latest developments in quantum computing"
          num_results (int): Number of initial search results to fetch from DuckDuckGo.
                            More results provide better coverage but increase processing time.
                            Default: 10
          top_k (int): Number of top-scored results to include in the final output.
                      These are the most semantically relevant results after scoring.
                      Default: 5
          include_urls (bool): Whether to include source URLs in the results.
                              If True, each result includes its URL for citation.
                              Default: True
      
      Returns:
          Dict: A dictionary with a single key "content" containing the search results.
                The content is formatted as markdown text with the most relevant information
                from the top_k web pages. If include_urls is True, each section includes
                its source URL.
                
      Example:
          {"content": "# Result 1\\n\\nContent here...\\n\\nSource: https://example.com"}
      """
      
      # Import heavy dependencies only when tool is invoked
      from ddgs import DDGS
      from .utils.fetch import fetch_all_content
      from .utils.tools import sort_by_score
      
      ddgs = DDGS()
      results = ddgs.text(query, max_results=num_results) 
      scored_results = sort_by_score(add_score_to_dict(query, results))
      top_results = scored_results[0:top_k]
  
      # fetch content using thread pool
      md_content = fetch_all_content(top_results, include_urls)
  
      # formatted as dict
      return {
          "content": md_content
              }

• src/mcp_local_rag/main.py:25-25 (registration)

  The tool is registered with FastMCP using the @mcp.tool() decorator on line 25.

  @mcp.tool()

• src/mcp_local_rag/main.py:7-23 (helper)

  add_score_to_dict computes cosine similarity scores between the query and each search result using a MediaPipe text embedder.

  def add_score_to_dict(query: str, results: List[Dict]) -> List[Dict]:
      """Add similarity scores to search results."""
      # Import heavy dependencies only when needed (slow import!)
      from importlib.resources import files
      from mediapipe.tasks.python import text
      from .utils.fetch import fetch_embedder, get_path_str
      
      path = get_path_str(files('mcp_local_rag.embedder').joinpath('embedder.tflite'))
      embedder = fetch_embedder(path)
      query_embedding = embedder.embed(query)
  
      for i in results:
          i['score'] = text.TextEmbedder.cosine_similarity(
                          embedder.embed(i['body']).embeddings[0],
                          query_embedding.embeddings[0])
  
      return results

• src/mcp_local_rag/utils/tools.py:11-13 (helper)

  sort_by_score sorts the list of result dicts by their 'score' field in descending order.

  def sort_by_score(results: List[Dict]) -> List[Dict]:
      """Sort results by similarity score."""
      return sorted(results, key=lambda x: x['score'], reverse=True)

• src/mcp_local_rag/utils/fetch.py:46-70 (helper)

  fetch_all_content fetches content from all result URLs in parallel using a ThreadPoolExecutor, optionally including URLs in the output.

  def fetch_all_content(results: List[Dict], include_urls:bool=True) -> List[str]:
      """Fetch content from all URLs using a thread pool."""
      urls = [site['href'] for site in results if site.get('href')]
      
      # parallelize requests
      with ThreadPoolExecutor(max_workers=5) as executor:
          # submit fetch tasks to executor
          future_to_url = {executor.submit(fetch_content, url): url for url in urls}
          
          content_list = []
          for future, url in future_to_url.items():
              try:
                  content = future.result()
                  if content:
                      result = {
                          "type": "text",
                          "text": content,
                      }
                      if include_urls:
                          result["url"] = url
                      content_list.append(result)
              except Exception as e:
                  print(f"Request failed with exception: {e}")
          
      return content_list