search

Instructions

Implementation Reference

• src/mcp_server.py:218-261 (handler)

  The primary handler function for the 'search' MCP tool, decorated with @mcp.tool(name="search"). It limits results using settings.search_top_k, calls ChromaService.search, serializes PostSearchResult objects into a response dictionary with query, results list, and count.

  @mcp.tool(name="search")
  async def search(
      query: str,
      limit: int = 10,
  ) -> dict[str, Any]:
      """
      Search blog posts using semantic search.
  
      Args:
          query: Search query text
          limit: Maximum number of results to return (default: 10)
  
      Returns:
          List of search results with relevance scores
      """
      if limit > settings.search_top_k:
          limit = settings.search_top_k
  
      chroma_service = await get_chroma_service()
      results = await chroma_service.search(query, limit)
  
      serialized_results: list[dict[str, Any]] = []
      for result in results:
          if not result.post_url:
              logger.debug("Skipping search result without canonical URL: %s", result.post_slug)
              continue
  
          serialized_results.append(
              {
                  "id": result.post_url,
                  "title": result.post_title,
                  "url": result.post_url,
                  "excerpt": result.excerpt,
                  "relevance_score": result.relevance_score,
                  "published_at": result.published_at.isoformat() if result.published_at else None,
                  "tags": result.tags,
              }
          )
  
      return {
          "query": query,
          "results": serialized_results,
          "count": len(serialized_results),
      }

• src/models.py:51-59 (schema)

  Pydantic BaseModel defining the structure of individual search results (aliased as PostSearchResult), including post metadata, excerpt, relevance score, used directly in the tool's implementation.

  class SearchResult(BaseModel):
      post_slug: str
      post_title: str
      post_url: str
      excerpt: str
      relevance_score: float
      published_at: datetime | None
      tags: list[str] = Field(default_factory=list)

• src/chroma_service.py:258-284 (helper)

  ChromaService.search method implementing hybrid dense+sparse semantic search using RRF fusion, embedding the query, executing via _hybrid_rrf_search, parsing into PostSearchResult list; invoked by the MCP tool handler.

  async def search(self, query: str, limit: int = 10) -> list[PostSearchResult]:
      dense_embedding = self.embedding_service.embed_query(query)
      sparse_embedding = self.embedding_service.sparse_embed_query(query)
  
      results = self._hybrid_rrf_search(
          dense_embedding=dense_embedding,
          sparse_embedding=sparse_embedding,
          limit=limit,
      )
  
      # Console-friendly printout for quick debugging/inspection
      try:
          print(f"\n📊 Hybrid Search Results (RRF Combined) — Query: {query!r}")
          if results:
              for i, result in enumerate(results):
                  title = result.post_title or "Unknown"
                  score = result.relevance_score or 0.0
                  excerpt = (result.excerpt or "").strip()
                  print(f"   {i + 1}. [{title}] Score: {score:.4f}")
                  if excerpt:
                      print(f"      Text: {excerpt[:100]}...")
          else:
              print("   No results")
      except Exception:  # Best-effort printing; never block search on logging issues
          pass
  
      return results

• src/config.py:31-31 (helper)

  Configuration setting search_top_k used to cap the limit parameter in the search tool handler.

  search_top_k: int = Field(default=10, description="Number of search results to return")