mcp-local-rag

Instructions

Implementation Reference

• src/mcp_local_rag/main.py:7-43 (handler)

  Core implementation of the rag_search MCP tool. Performs web search with duckduckgo-search (DDGS), embeds query and results using MediaPipe TextEmbedder for similarity scoring, sorts top results, fetches content asynchronously from URLs, returns dict with list of result dicts containing text and optional URLs. Registered via FastMCP @tool decorator.
  @mcp.tool() def rag_search(query: str, num_results:int=10, top_k:int=5, include_urls:bool=True) -> Dict: """ Search the web for a given query. Give back context to the LLM with a RAG-like similarity sort. Args: query (str): The query to search for. num_results (int): Number of results to return. top_k (int): Use top "k" results for content. include_urls (bool): Whether to include URLs in the results. If True, the results will be a list of dictionaries with the following keys: - type: "text" - text: The content of the result - url: The URL of the result Returns: Dict of strings containing best search based on input query. Formatted in markdown. """ # 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:44-60 (helper)

  Helper function called by rag_search to compute cosine similarity scores between query embedding and search result body embeddings using a local MediaPipe TFLite embedder model.
  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)

  Helper utility used by rag_search to sort scored search results by similarity score 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)

  Helper function used by rag_search to concurrently fetch text content from top search result URLs using ThreadPoolExecutor (max 5 workers), parses with BeautifulSoup, limits to 10k chars, formats as list of dicts.
  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
• src/mcp_local_rag/main.py:7-7 (registration)

  FastMCP decorator that registers the rag_search function as a tool with name derived from function name, auto-generates schema from signature and docstring.
  @mcp.tool()