Documentation Search MCP Server

Instructions

Implementation Reference

• src/documentation_search_enhanced/main.py:814-875 (handler)

  The main handler function for the 'preindex_docs' MCP tool. It processes a list of libraries, calls preindex_site for each, and saves the preindexed state.
  async def preindex_docs( libraries: LibrariesParam, include_sitemap: bool = False, persist_path: Optional[str] = None, max_concurrent_sites: int = 3, ): """ Pre-download and persist docs site indexes for Serper-free search. This caches MkDocs/Sphinx search indexes (and optionally sitemaps) to disk so the server can search supported documentation sites without requiring Serper. """ await enforce_rate_limit("preindex_docs") targets = libraries or sorted(docs_urls.keys()) if not targets: return { "status": "no_targets", "message": "No libraries configured to preindex", } global http_client if http_client is None: http_client = httpx.AsyncClient(timeout=httpx.Timeout(30.0, read=60.0)) concurrency = max(1, min(int(max_concurrent_sites), 10)) semaphore = asyncio.Semaphore(concurrency) async def _run_one(library: str) -> Dict[str, Any]: docs_root = docs_urls.get(library) if not docs_root: return {"library": library, "status": "unsupported"} async with semaphore: summary = await preindex_site( docs_root, http_client, user_agent=USER_AGENT, include_sitemap=include_sitemap, ) summary["library"] = library return summary results = await asyncio.gather(*[_run_one(lib) for lib in targets]) path = persist_path or site_index_path try: save_preindexed_state(path) persisted: Dict[str, Any] = {"status": "ok", "path": path} except Exception as e: persisted = {"status": "error", "path": path, "error": str(e)} return { "status": "ok", "persist": persisted, "real_time_search_enabled": real_time_search_enabled, "include_sitemap": include_sitemap, "max_concurrent_sites": concurrency, "total_libraries": len(targets), "results": results, }
• src/documentation_search_enhanced/site_search.py:237-316 (helper)

  Core helper function that performs the actual preindexing of a single documentation site by fetching MkDocs/Sphinx search indexes and optionally sitemaps.
  async def preindex_site( site_url: str, client: httpx.AsyncClient, *, user_agent: str, include_sitemap: bool = False, ) -> Dict[str, Any]: """Fetch and cache on-site search indexes for a docs site.""" parsed = urlparse(site_url) if not parsed.scheme or not parsed.netloc: return {"site_url": site_url, "status": "invalid_url"} origin = f"{parsed.scheme}://{parsed.netloc}" results: Dict[str, Any] = { "site_url": site_url, "origin": origin, "mkdocs_index": None, "sphinx_index": None, "sitemap": None, "errors": [], } for index_url in _mkdocs_index_candidates(site_url): try: docs = await _get_cached_index( client, index_url, user_agent=user_agent, kind="mkdocs", timeout_seconds=20.0, ) except Exception as e: results["errors"].append(f"mkdocs:{index_url}: {e}") continue if docs: results["mkdocs_index"] = {"index_url": index_url, "documents": len(docs)} break for index_url in _sphinx_index_candidates(site_url): try: index = await _get_cached_index( client, index_url, user_agent=user_agent, kind="sphinx", timeout_seconds=20.0, ) except Exception as e: results["errors"].append(f"sphinx:{index_url}: {e}") continue if isinstance(index, dict): filenames = index.get("filenames") results["sphinx_index"] = { "index_url": index_url, "documents": len(filenames) if isinstance(filenames, list) else None, } break if include_sitemap: try: urls = await _load_site_sitemap_urls( client, site_url, user_agent=user_agent ) if urls: _sitemap_cache[origin] = _SitemapCacheEntry( fetched_at=datetime.now(), urls=tuple(urls) ) results["sitemap"] = {"urls": len(urls)} except Exception as e: results["errors"].append(f"sitemap:{origin}: {e}") results["status"] = ( "ok" if results.get("mkdocs_index") or results.get("sphinx_index") or results.get("sitemap") else "no_index_found" ) return results
• src/documentation_search_enhanced/site_search.py:204-213 (helper)

  Helper function to save the preindexed state to disk, called by the tool handler.
  def save_preindexed_state(path: str) -> None: """Persist current in-memory sitemap/index caches to disk.""" if not path: raise ValueError("persist path must be non-empty") state = export_preindexed_state() tmp_path = f"{path}.tmp" with open(tmp_path, "w", encoding="utf-8") as fh: json.dump(state, fh) os.replace(tmp_path, path)