Chroma MCP Server

Instructions

Args:
    collection_name: Name of the collection to count

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