Chroma MCP Server

Instructions

Args:
    collection_name: Name of the collection to fork
    new_collection_name: Name of the new collection to create
    metadata: Optional metadata dict to add to the new collection

Implementation Reference

• src/chroma_mcp/server.py:297-315 (handler)

  The main handler function for the 'chroma_fork_collection' tool. It forks an existing Chroma collection into a new one using the Chroma client's fork method. Registered via @mcp.tool() decorator.

  @mcp.tool()
  async def chroma_fork_collection(
      collection_name: str,
      new_collection_name: str,
  ) -> str:
      """Fork a Chroma collection.
      
      Args:
          collection_name: Name of the collection to fork
          new_collection_name: Name of the new collection to create
          metadata: Optional metadata dict to add to the new collection
      """
      client = get_chroma_client()
      try:
          collection = client.get_collection(collection_name)
          collection.fork(new_collection_name)
          return f"Successfully forked collection {collection_name} to {new_collection_name}"
      except Exception as e:
          raise Exception(f"Failed to fork collection '{collection_name}': {str(e)}") from e

• src/chroma_mcp/server.py:297-297 (registration)

  The @mcp.tool() decorator registers the chroma_fork_collection function as an MCP tool.

  @mcp.tool()

• src/chroma_mcp/server.py:298-301 (schema)

  Function signature defining input parameters (collection_name, new_collection_name) and return type (str), serving as the tool schema.

  async def chroma_fork_collection(
      collection_name: str,
      new_collection_name: str,
  ) -> str:

• src/chroma_mcp/server.py:72-141 (helper)

  Helper function to get or initialize the Chroma client instance, used by the tool.

  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