Chroma MCP Server

Instructions

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