slack-indexed

Instructions

Implementation Reference

• server.py:107-200 (handler)

  The main handler function for the 'search' MCP tool. Implements semantic search over indexed Slack data using vector similarity with Cohere reranking. Accepts parameters for query, result count, and filters (source, channel, user). Returns ranked results with metadata.

  @mcp.tool()
  def search(
      query: str,
      n_results: int = 10,
      source_filter: str | None = None,
      channel_filter: str | None = None,
      user_filter: str | None = None,
      include_links: bool = True,
  ) -> dict:
      """Search the Slack index using semantic similarity with Cohere reranking.
  
      Args:
          query: Natural language search query.
          n_results: Number of results to return (default 10, max 50).
          source_filter: Filter by source type. One of: slack_thread, github_issue,
                         github_pr, github_file, linear_issue, notion_page.
          channel_filter: Filter by Slack channel name (partial match).
          user_filter: Filter by participant name (partial match on thread_users).
          include_links: If False, exclude linked resources and return only Slack threads.
      """
      store = _get_store()
      reranker = _get_reranker()
      n_results = min(max(1, n_results), 50)
  
      where_clauses: List[Dict] = []
      if source_filter:
          if source_filter not in SOURCE_TYPES:
              return {
                  "error": f"Invalid source_filter. Must be one of: {sorted(SOURCE_TYPES)}"
              }
          where_clauses.append({"source": source_filter})
      elif not include_links:
          where_clauses.append({"source": "slack_thread"})
  
      if channel_filter:
          where_clauses.append({"channel_name": {"$contains": channel_filter}})
      if user_filter:
          where_clauses.append({"thread_users": {"$contains": user_filter}})
  
      where = None
      if len(where_clauses) == 1:
          where = where_clauses[0]
      elif len(where_clauses) > 1:
          where = {"$and": where_clauses}
  
      fetch_n = n_results * 3
      kwargs: Dict[str, Any] = {"query_texts": [query], "n_results": fetch_n}
      if where:
          kwargs["where"] = where
  
      try:
          results = store.query(**kwargs)
          candidates_ids = results["ids"][0]
          candidates_docs = results["documents"][0]
          candidates_distances = results["distances"][0]
          candidates_metas = results["metadatas"][0]
  
          if not candidates_docs:
              return {"query": query, "count": 0, "results": []}
  
          try:
              rerank_results = reranker.rerank(
                  query=query, documents=candidates_docs, top_n=n_results
              )
              formatted = []
              for r in rerank_results:
                  idx = r["index"]
                  formatted.append(
                      {
                          "id": candidates_ids[idx],
                          "text": candidates_docs[idx],
                          "rerank_score": round(r["score"], 4),
                          "vector_distance": round(candidates_distances[idx], 4),
                          "metadata": candidates_metas[idx],
                      }
                  )
          except Exception as e:
              logger.warning(f"Reranker failed, falling back to vector results: {e}")
              formatted = []
              for i in range(min(n_results, len(candidates_docs))):
                  formatted.append(
                      {
                          "id": candidates_ids[i],
                          "text": candidates_docs[i],
                          "rerank_score": None,
                          "vector_distance": round(candidates_distances[i], 4),
                          "metadata": candidates_metas[i],
                      }
                  )
  
          return {"query": query, "count": len(formatted), "results": formatted}
      except Exception as e:
          logger.error(f"Error in search: {e}")
          return {"error": str(e)}

• server.py:109-115 (schema)

  Input schema definition for the search tool. Defines typed parameters: query (string), n_results (int, default 10), source_filter (optional string), channel_filter (optional string), user_filter (optional string), and include_links (bool, default True).

      query: str,
      n_results: int = 10,
      source_filter: str | None = None,
      channel_filter: str | None = None,
      user_filter: str | None = None,
      include_links: bool = True,
  ) -> dict:

• server.py:107-107 (registration)

  Registration of the search tool with the MCP server using the @mcp.tool() decorator, which exposes the function as an MCP tool named 'search'.

  @mcp.tool()

• vector_store.py:48-95 (helper)

  QdrantVectorStore.query() method - helper function that performs the actual vector similarity search against the Qdrant database. Generates embeddings for queries, applies filters, and returns matching documents with distances and metadata.

  def query(
      self, query_texts: List[str], n_results: int = 10, **kwargs: Any
  ) -> Dict[str, Any]:
      query_embeddings = self.embedder.embed_queries(query_texts)
  
      all_ids: List[List[str]] = []
      all_docs: List[List[str]] = []
      all_distances: List[List[float]] = []
      all_metadatas: List[List[Dict]] = []
  
      qdrant_filter = None
      if "where" in kwargs:
          qdrant_filter = _translate_filter(kwargs["where"])
  
      for query_embedding in query_embeddings:
          results = self.client.query_points(
              collection_name=self.collection_name,
              query=query_embedding,
              limit=n_results,
              query_filter=qdrant_filter,
              with_payload=True,
          ).points
  
          ids, docs, distances, metadatas = [], [], [], []
          for point in results:
              ids.append(point.payload.get("_string_id", str(point.id)))
              docs.append(point.payload.get("document", ""))
              distances.append(
                  1.0 - point.score if point.score is not None else 1.0
              )
              meta = {
                  k: v
                  for k, v in point.payload.items()
                  if k not in ("document", "_string_id")
              }
              metadatas.append(meta)
  
          all_ids.append(ids)
          all_docs.append(docs)
          all_distances.append(distances)
          all_metadatas.append(metadatas)
  
      return {
          "ids": all_ids,
          "documents": all_docs,
          "distances": all_distances,
          "metadatas": all_metadatas,
      }

• reranker.py:23-31 (helper)

  Reranker.rerank() method - helper function that reranks search results using Cohere's reranking model on AWS Bedrock. Takes query and candidate documents, returns top_n results with relevance scores.

  def rerank(
      self, query: str, documents: List[str], top_n: int = 10
  ) -> List[Dict[str, Any]]:
      response = self.client.rerank(
          model=self.model, query=query, documents=documents, top_n=top_n
      )
      return [
          {"index": r.index, "score": r.relevance_score} for r in response.results
      ]