MCP MixSearch

Instructions

{
  "query": "sustainable energy solutions 2024"
}

{
  "query": "latest AI developments",
  "limit": 3
}

{
  "query": "AI technology breakthroughs",
  "source": "news",
  "recency_days": 7,
  "language": "en"
}

{
  "query": "climate change policy",
  "country": "FR",
  "language": "fr",
  "limit": 5
}

{
  "query": "machine learning frameworks comparison",
  "limit": 8,
  "include_content": true,
  "max_content_length": 25000,
  "recency_days": 30,
  "language": "en"
}

{
  "query": "python tutorial",
  "source": "videos",
  "limit": 5,
  "include_content": false
}

{
  "query": "quantum computing breakthroughs",
  "limit": 5,
  "include_content": false
}

Implementation Reference

• features/web_search/tool.py:26-98 (handler)

  The primary handler function for the MCP tool 'full_web_search'. Registered via @mcp.tool() decorator within the register_tool function. Handles parameters, invokes WebSearchService, formats search results into MCP-compatible text content.

  @mcp.tool()
  @inject_docstring(lambda: load_instruction("instructions.md", __file__))
  async def full_web_search(
      query: str,
      limit: int = 5,
      include_content: bool = True,
      max_content_length: int = None,
      top_n: int = None,  # Alternative name for limit (for compatibility)
      recency_days: int = None,  # Days to look back (now supported!)
      source: str = None,  # Source type filter (partially supported)
      language: str = None,  # Language filter (now supported!)
      country: str = None  # Country filter (now supported!)
  ) -> Dict[str, Any]:
      """Comprehensive web search with full content extraction"""
      try:
          # Handle alternative parameter names
          actual_limit = top_n if top_n is not None else limit
          
          # Prepare supported parameters
          search_params = {}
          if recency_days is not None:
              search_params['recency_days'] = recency_days
              logger.info(f"Using recency_days filter: {recency_days} days")
          if source is not None:
              search_params['source'] = source
              logger.info(f"Using source filter: '{source}'")
          if language is not None:
              search_params['language'] = language
              logger.info(f"Using language filter: '{language}'")
          if country is not None:
              search_params['country'] = country
              logger.info(f"Using country filter: '{country}'")
              
          logger.info(f"MCP tool full_web_search: query='{query}', limit={actual_limit}")
  
          result = await web_search_service.search_and_extract(
              query=query,
              limit=min(actual_limit, 10),  # Cap at 10
              max_content_length=max_content_length,
              include_content=include_content,
              **search_params  # Pass all supported parameters to service
          )
  
          # Format for MCP response
          response_text = f"Search completed for '{result['query']}' with {result['total_results']} results:\n\n"
  
          for i, search_result in enumerate(result['results'], 1):
              response_text += f"**{i}. {search_result['title']}**\n"
              response_text += f"URL: {search_result['url']}\n"
              response_text += f"Description: {search_result['description']}\n"
  
              if search_result.get('full_content'):
                  content = search_result['full_content']
                  if max_content_length and len(content) > max_content_length:
                      content = content[:max_content_length] + f"\n\n[Content truncated at {max_content_length} characters]"
                  response_text += f"\n**Full Content:**\n{content}\n"
              elif search_result.get('content_preview'):
                  response_text += f"\n**Content Preview:**\n{search_result['content_preview']}\n"
              elif search_result.get('fetch_status') == 'error':
                  response_text += f"\n**Content Extraction Failed:** {search_result.get('error', 'Unknown error')}\n"
  
              response_text += "\n---\n\n"
  
          logger.info(f"MCP tool full_web_search completed: {result['total_results']} results")
  
          return {
              "content": [{"type": "text", "text": response_text}]
          }
  
      except Exception as e:
          logger.error(f"MCP tool full_web_search error: {e}")
          raise

• features/web_search/service.py:64-81 (helper)

  Core helper method in WebSearchService that performs multi-engine web search and optional content extraction, called directly by the tool handler.

  async def search_and_extract(self, query: str, limit: int = 5,
                             max_content_length: Optional[int] = None,
                             include_content: bool = True, **kwargs) -> Dict[str, Any]:
      """Full web search with content extraction"""
      logger.info(f"Starting comprehensive web search for: '{query}' with parameters: {kwargs}")
  
      results = await self._multi_engine_search(query, limit, **kwargs)
  
      # Extract full content concurrently if requested
      if include_content:
          await self._extract_content_concurrent(results, max_content_length)
  
      return {
          "query": query,
          "results": [self._result_to_dict(r) for r in results],
          "total_results": len(results),
          "status": "success"
      }

• features/web_search/tool.py:17-25 (registration)

  The register_tool function that defines and registers the full_web_search tool (and others) with the FastMCP server instance.

  def register_tool(mcp: FastMCP, web_search_service: WebSearchService) -> None:
      """
      Register web search tools with the MCP server
  
      Args:
          mcp: FastMCP server instance
          web_search_service: WebSearchService instance
      """