collection_stats
Retrieve summary statistics about indexed Slack message collections to analyze data volume, distribution, and search readiness.
Instructions
Get summary statistics about the indexed collection.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- server.py:272-296 (handler)The collection_stats tool handler function that retrieves and aggregates statistics about the indexed Slack collection. It queries all documents, counts by source type and channel, tracks unique threads, and returns summary statistics including total documents, unique threads, and breakdowns by source and channel.
@mcp.tool() def collection_stats() -> dict: """Get summary statistics about the indexed collection.""" store = _get_store() all_docs = store.get(include=["metadatas"]) source_counts: Dict[str, int] = {} channel_counts: Dict[str, int] = {} unique_threads: set[str] = set() for meta in all_docs["metadatas"]: source = meta.get("source", "unknown") source_counts[source] = source_counts.get(source, 0) + 1 channel = meta.get("channel_name", "unknown") channel_counts[channel] = channel_counts.get(channel, 0) + 1 ts = meta.get("thread_ts") if ts: unique_threads.add(ts) return { "total_documents": store.count(), "unique_threads": len(unique_threads), "by_source": source_counts, "by_channel": channel_counts, } - server.py:272-272 (registration)The @mcp.tool() decorator registers the collection_stats function as an MCP tool, making it available for invocation by MCP clients.
@mcp.tool() - server.py:50-53 (registration)FastMCP server initialization that creates the MCP server instance where all tools (including collection_stats) are registered.
mcp = FastMCP( "slack-indexed", instructions="Search indexed Slack channel messages and linked resources (GitHub, Linear, Notion).", ) - server.py:80-90 (helper)The _get_store() helper function provides lazy initialization of the QdrantVectorStore instance used by collection_stats to query the indexed collection.
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