Chroma MCP Server

Instructions

Implementation Reference

• src/chroma/server.py:362-410 (handler)

  The core handler function for the 'create_document' tool. It validates inputs, checks for existing documents, processes metadata, and adds the document to the Chroma collection using collection.add().
  async def handle_create_document(arguments: dict) -> list[types.TextContent]: """Handle document creation with retry logic""" doc_id = arguments.get("document_id") content = arguments.get("content") metadata = arguments.get("metadata") if not doc_id or not content: raise DocumentOperationError("Missing document_id or content") try: # Check if document exists using get() instead of collection.get() try: existing = collection.get( ids=[doc_id], include=['metadatas'] ) if existing and existing['ids']: raise DocumentOperationError(f"Document already exists [id={doc_id}]") except Exception as e: if "not found" not in str(e).lower(): raise # Process metadata if metadata: processed_metadata = { k: str(v) if isinstance(v, (int, float)) else v for k, v in metadata.items() } else: processed_metadata = {} # Add document collection.add( documents=[content], ids=[doc_id], metadatas=[processed_metadata] ) return [ types.TextContent( type="text", text=f"Created document '{doc_id}' successfully" ) ] except DocumentOperationError: raise except Exception as e: raise DocumentOperationError(str(e))
• src/chroma/server.py:331-332 (registration)

  Tool dispatch logic in the main call_tool handler that routes 'create_document' calls to the specific handle_create_document function.
  if name == "create_document": return await handle_create_document(arguments)
• src/chroma/server.py:140-148 (schema)

  JSON schema definition for the 'create_document' tool inputs, used in server.command_options.
  "create_document": { "type": "object", "properties": { "document_id": {"type": "string"}, "content": {"type": "string"}, "metadata": {"type": "object", "additionalProperties": True} }, "required": ["document_id", "content"] },
• src/chroma/server.py:242-256 (registration)

  Tool registration in the list_tools handler, defining the 'create_document' tool with name, description, and input schema.
  name="create_document", description="Create a new document in the Chroma vector database", inputSchema={ "type": "object", "properties": { "document_id": {"type": "string"}, "content": {"type": "string"}, "metadata": { "type": "object", "additionalProperties": True } }, "required": ["document_id", "content"] } ),
• src/chroma/server.py:41-84 (helper)

  Retry decorator applied to create_document handler (@retry_operation('create_document') at line 361), providing exponential backoff and error handling for Chroma operations.
  def retry_operation(operation_name: str): """Decorator to retry document operations with exponential backoff""" def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): max_retries = 3 for attempt in range(max_retries): try: return await func(*args, **kwargs) except DocumentOperationError as e: if attempt == max_retries - 1: raise e await asyncio.sleep(2 ** attempt) except Exception as e: if attempt == max_retries - 1: # Clean up error message msg = str(e) if msg.lower().startswith(operation_name.lower()): msg = msg[len(operation_name):].lstrip(': ') if msg.lower().startswith('failed'): msg = msg[7:].lstrip(': ') if msg.lower().startswith('search failed'): msg = msg[13:].lstrip(': ') # Map error patterns to friendly messages error_msg = msg.lower() doc_id = kwargs.get('arguments', {}).get('document_id') if "not found" in error_msg: error = f"Document not found{f' [id={doc_id}]' if doc_id else ''}" elif "already exists" in error_msg: error = f"Document already exists{f' [id={doc_id}]' if doc_id else ''}" elif "invalid" in error_msg: error = "Invalid input" elif "filter" in error_msg: error = "Invalid filter" else: error = "Operation failed" raise DocumentOperationError(error) await asyncio.sleep(2 ** attempt) return None return wrapper return decorator