DevLens MCP

Instructions

Implementation Reference

• src/devlens/tools/research.py:11-111 (handler)

  The actual implementation of the tool_deep_dive functionality, which uses DDG search and Scraper tools to aggregate a research report.

  async def deep_dive(topic: str, depth: int = 3, *, parallel: bool = True) -> str:
      """Research a topic by searching and scraping multiple sources.
  
      Args:
          topic: Topic to research.
          depth: Number of sources to scrape (1-10).
          parallel: Whether to scrape sources in parallel (faster).
  
      Returns:
          Aggregated research report with sources.
  
      Example:
          >>> report = await deep_dive("Python async/await tutorial", depth=5)
      """
      import asyncio
  
      depth = min(max(depth, 1), 10)
  
      # Search for sources
      results = await _ddg.search(topic, limit=depth * 2)  # Get more to filter
  
      # Filter to unique domains for diversity
      from urllib.parse import urlparse
  
      seen_domains = set()
      filtered_results = []
      for r in results:
          domain = urlparse(r.url).netloc
          if domain not in seen_domains and len(filtered_results) < depth:
              seen_domains.add(domain)
              filtered_results.append(r)
  
      results = filtered_results
  
      if not results:
          return f"# Research: {topic}\n\nNo results found for this topic."
  
      # Build report header
      report_lines = [
          f"# Research: {topic}\n",
          f"*Analyzed {len(results)} sources*\n",
          "## Sources\n",
      ]
  
      # Add sources list
      for i, r in enumerate(results, 1):
          report_lines.append(f"{i}. [{r.title}]({r.url})")
  
      report_lines.append("\n## Content\n")
  
      # Scrape sources (parallel or sequential)
      async def fetch_source(i: int, r) -> tuple[int, str, str, str | None]:
          """Fetch a single source."""
          try:
              doc = await _scraper.fetch(r.url)
              # Extract content without header
              lines = doc.content.split("\n")
              start = 0
              for j, line in enumerate(lines):
                  if line.startswith("> Source:"):
                      start = j + 1
                      break
              content = "\n".join(lines[start:]).strip()
              # Truncate if too long
              if len(content) > 3000:
                  content = content[:3000] + "\n\n*[Content truncated...]*"
              return (i, r.title, r.url, content)
          except Exception:
              return (i, r.title, r.url, None)
  
      if parallel:
          # Fetch all sources concurrently
          tasks = [fetch_source(i, r) for i, r in enumerate(results, 1)]
          fetched = await asyncio.gather(*tasks, return_exceptions=False)
      else:
          # Fetch sequentially
          fetched = []
          for i, r in enumerate(results, 1):
              result = await fetch_source(i, r)
              fetched.append(result)
  
      # Build content sections
      successful = 0
      for i, title, url, content in fetched:
          report_lines.append(f"### Source {i}: {title}\n")
          report_lines.append(f"> {url}\n")
  
          if content:
              report_lines.append(content)
              successful += 1
          else:
              report_lines.append("*Failed to fetch content*")
  
          report_lines.append("\n")
  
      # Add summary footer
      report_lines.append(
          f"---\n\n*Successfully retrieved {successful}/{len(results)} sources*"
      )
  
      return "\n".join(report_lines)

• src/devlens/server.py:83-96 (registration)

  Registration of the tool_deep_dive tool using the @mcp.tool() decorator.

  @mcp.tool()
  async def tool_deep_dive(topic: str, depth: int = 3) -> str:
      """Research a topic from multiple sources.
  
      Searches and scrapes multiple pages to build a report.
  
      Args:
          topic: Topic to research.
          depth: Number of sources (1-10, default 3).
  
      Returns:
          Aggregated research report.
      """
      return await deep_dive(topic, depth)