chroma_get_collection_count
Retrieve the total number of documents in a specified Chroma collection using this tool, enabling efficient data management and analysis for AI-driven applications.
Instructions
Get the number of documents in a Chroma collection.
Args:
collection_name: Name of the collection to count
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| collection_name | Yes |
Implementation Reference
- src/chroma_mcp/server.py:255-268 (handler)The handler function decorated with @mcp.tool(), which registers the tool and defines its input schema via type hints (collection_name: str) and output (int). It fetches the Chroma client, retrieves the collection, and returns the document count.@mcp.tool() async def chroma_get_collection_count(collection_name: str) -> int: """Get the number of documents in a Chroma collection. Args: collection_name: Name of the collection to count """ client = get_chroma_client() try: collection = client.get_collection(collection_name) return collection.count() except Exception as e: raise Exception(f"Failed to get collection count for '{collection_name}': {str(e)}") from e
- src/chroma_mcp/server.py:255-255 (registration)The @mcp.tool() decorator registers the chroma_get_collection_count function as an MCP tool.@mcp.tool()
- src/chroma_mcp/server.py:256-256 (schema)The function signature provides the input schema (collection_name: str) and output type (int) for the tool, used by FastMCP for validation.async def chroma_get_collection_count(collection_name: str) -> int:
- src/chroma_mcp/server.py:72-141 (helper)Helper function to initialize and return the Chroma client instance, called within the tool handler.def get_chroma_client(args=None): """Get or create the global Chroma client instance.""" global _chroma_client if _chroma_client is None: if args is None: # Create parser and parse args if not provided parser = create_parser() args = parser.parse_args() # Load environment variables from .env file if it exists load_dotenv(dotenv_path=args.dotenv_path) if args.client_type == 'http': if not args.host: raise ValueError("Host must be provided via --host flag or CHROMA_HOST environment variable when using HTTP client") settings = Settings() if args.custom_auth_credentials: settings = Settings( chroma_client_auth_provider="chromadb.auth.basic_authn.BasicAuthClientProvider", chroma_client_auth_credentials=args.custom_auth_credentials ) # Handle SSL configuration try: _chroma_client = chromadb.HttpClient( host=args.host, port=args.port if args.port else None, ssl=args.ssl, settings=settings ) except ssl.SSLError as e: print(f"SSL connection failed: {str(e)}") raise except Exception as e: print(f"Error connecting to HTTP client: {str(e)}") raise elif args.client_type == 'cloud': if not args.tenant: raise ValueError("Tenant must be provided via --tenant flag or CHROMA_TENANT environment variable when using cloud client") if not args.database: raise ValueError("Database must be provided via --database flag or CHROMA_DATABASE environment variable when using cloud client") if not args.api_key: raise ValueError("API key must be provided via --api-key flag or CHROMA_API_KEY environment variable when using cloud client") try: _chroma_client = chromadb.HttpClient( host="api.trychroma.com", ssl=True, # Always use SSL for cloud tenant=args.tenant, database=args.database, headers={ 'x-chroma-token': args.api_key } ) except ssl.SSLError as e: print(f"SSL connection failed: {str(e)}") raise except Exception as e: print(f"Error connecting to cloud client: {str(e)}") raise elif args.client_type == 'persistent': if not args.data_dir: raise ValueError("Data directory must be provided via --data-dir flag when using persistent client") _chroma_client = chromadb.PersistentClient(path=args.data_dir) else: # ephemeral _chroma_client = chromadb.EphemeralClient() return _chroma_client