Web Research Assistant

Instructions

Implementation Reference

• src/searxng_mcp/server.py:102-135 (handler)

  Primary handler and registration for the 'crawl_url' MCP tool. Uses CrawlerClient to fetch and process URL content, with input schema via Annotated types, error handling, and usage tracking.
  @mcp.tool() async def crawl_url( url: Annotated[str, "HTTP(S) URL (ideally from web_search output)"], reasoning: Annotated[str, "Why you're crawling this URL (required for analytics)"], max_chars: Annotated[int, "Trim textual result to this many characters"] = CRAWL_MAX_CHARS, ) -> str: """Fetch a URL with crawl4ai when you need the actual page text for quoting or analysis.""" start_time = time.time() success = False error_msg = None result = "" try: text = await crawler_client.fetch(url, max_chars=max_chars) result = clamp_text(text, MAX_RESPONSE_CHARS) success = True except Exception as exc: # noqa: BLE001 error_msg = str(exc) result = f"Crawl failed for {url}: {exc}" finally: # Track usage response_time = (time.time() - start_time) * 1000 tracker.track_usage( tool_name="crawl_url", reasoning=reasoning, parameters={"url": url, "max_chars": max_chars}, response_time_ms=response_time, success=success, error_message=error_msg, response_size=len(result.encode("utf-8")), ) return result
• src/searxng_mcp/crawler.py:14-37 (helper)

  Core crawling logic in CrawlerClient.fetch(), called by the tool handler. Uses crawl4ai AsyncWebCrawler to fetch, extract markdown/content, clean, and trim text.
  async def fetch(self, url: str, *, max_chars: int | None = None) -> str: """Fetch *url* and return cleaned markdown, trimmed to *max_chars*.""" run_config = CrawlerRunConfig(cache_mode=self.cache_mode) async with AsyncWebCrawler() as crawler: result = await crawler.arun(url=url, config=run_config) if getattr(result, "error", None): raise RuntimeError(str(result.error)) # type: ignore text = ( getattr(result, "markdown", None) or getattr(result, "content", None) or getattr(result, "html", None) or "" ) text = text.strip() if not text: raise RuntimeError("Crawl completed but returned no readable content.") limit = max_chars or CRAWL_MAX_CHARS return clamp_text(text, limit)
• src/searxng_mcp/crawler.py:8-11 (helper)

  CrawlerClient class initialization and configuration, instantiated globally as crawler_client in server.py.
  class CrawlerClient: """Lightweight wrapper around crawl4ai's async crawler.""" def __init__(self, *, cache_mode: CacheMode = CacheMode.BYPASS) -> None:
• src/searxng_mcp/server.py:21-102 (registration)

  Import and global instantiation of CrawlerClient used by the crawl_url tool.
  from .crawler import CrawlerClient from .errors import ErrorParser from .extractor import DataExtractor from .github import GitHubClient, RepoInfo from .images import PixabayClient from .registry import PackageInfo, PackageRegistryClient from .search import SearxSearcher from .service_health import ServiceHealthChecker from .tracking import get_tracker mcp = FastMCP("web-research-assistant") searcher = SearxSearcher() crawler_client = CrawlerClient() registry_client = PackageRegistryClient() github_client = GitHubClient() pixabay_client = PixabayClient() error_parser = ErrorParser() api_docs_detector = APIDocsDetector() api_docs_extractor = APIDocsExtractor() data_extractor = DataExtractor() tech_comparator = TechComparator(searcher, github_client, registry_client) changelog_fetcher = ChangelogFetcher(github_client, registry_client) service_health_checker = ServiceHealthChecker(crawler_client) tracker = get_tracker() def _format_search_hits(hits): lines = [] for idx, hit in enumerate(hits, 1): snippet = f"\n{hit.snippet}" if hit.snippet else "" lines.append(f"{idx}. {hit.title} — {hit.url}{snippet}") body = "\n\n".join(lines) return clamp_text(body, MAX_RESPONSE_CHARS) @mcp.tool() async def web_search( query: Annotated[str, "Natural-language web query"], reasoning: Annotated[str, "Why you're using this tool (required for analytics)"], category: Annotated[ str, "Optional SearXNG category (general, images, news, it, science, etc.)" ] = DEFAULT_CATEGORY, max_results: Annotated[int, "How many ranked hits to return (1-10)"] = DEFAULT_MAX_RESULTS, ) -> str: """Use this first to gather fresh web search results via the local SearXNG instance.""" start_time = time.time() success = False error_msg = None result = "" try: hits = await searcher.search(query, category=category, max_results=max_results) if not hits: result = f"No results for '{query}' in category '{category}'." else: result = _format_search_hits(hits) success = True except Exception as exc: # noqa: BLE001 error_msg = str(exc) result = f"Search failed: {exc}" finally: # Track usage response_time = (time.time() - start_time) * 1000 # Convert to ms tracker.track_usage( tool_name="web_search", reasoning=reasoning, parameters={ "query": query, "category": category, "max_results": max_results, }, response_time_ms=response_time, success=success, error_message=error_msg, response_size=len(result.encode("utf-8")), ) return result @mcp.tool()