slack-indexed

Instructions

Implementation Reference

• server.py:247-269 (registration)

  The @mcp.tool() decorator registers the list_channels function as an MCP tool. This is the registration point where the tool is exposed through the MCP server.

  @mcp.tool()
  def list_channels() -> dict:
      """List all indexed channels and their document counts."""
      store = _get_store()
      all_docs = store.get(include=["metadatas"])
  
      channels: Dict[str, Dict] = {}
      for meta in all_docs["metadatas"]:
          name = meta.get("channel_name", "unknown")
          if name not in channels:
              channels[name] = {
                  "channel_id": meta.get("channel_id", ""),
                  "total_docs": 0,
                  "threads": 0,
                  "links": 0,
              }
          channels[name]["total_docs"] += 1
          if meta.get("source") == "slack_thread":
              channels[name]["threads"] += 1
          else:
              channels[name]["links"] += 1
  
      return {"channels": channels, "total_documents": store.count()}

• server.py:248-269 (handler)

  The list_channels function is the main handler that executes the tool logic. It retrieves all documents from the vector store, aggregates them by channel name, and returns channel statistics including document counts, thread counts, and link counts.

  def list_channels() -> dict:
      """List all indexed channels and their document counts."""
      store = _get_store()
      all_docs = store.get(include=["metadatas"])
  
      channels: Dict[str, Dict] = {}
      for meta in all_docs["metadatas"]:
          name = meta.get("channel_name", "unknown")
          if name not in channels:
              channels[name] = {
                  "channel_id": meta.get("channel_id", ""),
                  "total_docs": 0,
                  "threads": 0,
                  "links": 0,
              }
          channels[name]["total_docs"] += 1
          if meta.get("source") == "slack_thread":
              channels[name]["threads"] += 1
          else:
              channels[name]["links"] += 1
  
      return {"channels": channels, "total_documents": store.count()}

• server.py:80-90 (helper)

  The _get_store() helper function lazily initializes and returns a QdrantVectorStore instance. This is used by the list_channels handler to access the indexed Slack data.

  def _get_store() -> QdrantVectorStore:
      global _store
      if _store is None:
          _store = QdrantVectorStore(
              url=_qdrant_url,
              collection_name=_qdrant_collection,
              embedder=_get_embedder(),
              timeout=_qdrant_timeout,
              api_key=_qdrant_api_key,
          )
      return _store