insert
Insert multiple records into a SurrealDB table in a single bulk operation. Optimized for efficient data insertion with auto-generated IDs, timestamps, and schema validation.
Instructions
Insert multiple records into a table in a single operation.
This tool is optimized for bulk inserts when you need to create many records at once. It's more efficient than calling 'create' multiple times. Each record will get:
An auto-generated unique ID
Automatic created/updated timestamps
Schema validation (if defined)
Args: table: The name of the table to insert records into (e.g., "user", "product") data: Array of dictionaries, each representing a record to insert. Example: [ {"name": "Alice", "email": "alice@example.com"}, {"name": "Bob", "email": "bob@example.com"}, {"name": "Charlie", "email": "charlie@example.com"} ] namespace: Optional SurrealDB namespace override. If not provided, uses SURREAL_NAMESPACE env var. database: Optional SurrealDB database override. If not provided, uses SURREAL_DATABASE env var.
Returns: A dictionary containing: - success: Boolean indicating if insertion was successful - data: Array of all inserted records with their generated IDs - count: Number of records successfully inserted - error: Error message if insertion failed (only present on failure)
Examples: >>> await insert("user", [ ... {"name": "Alice", "role": "admin"}, ... {"name": "Bob", "role": "user"} ... ]) { "success": true, "data": [ {"id": "user:ulid1", "name": "Alice", "role": "admin", "created": "..."}, {"id": "user:ulid2", "name": "Bob", "role": "user", "created": "..."} ], "count": 2 }
Note: For single record creation, use the 'create' tool instead.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table | Yes | ||
| data | Yes | ||
| namespace | No | ||
| database | No |
Implementation Reference
- surreal_mcp/server.py:709-788 (handler)The primary handler for the 'insert' tool. This FastMCP-decorated async function handles input validation, namespace resolution, timestamp addition, delegates to repo_insert helper, and formats the response.@mcp.tool() async def insert( table: str, data: List[Dict[str, Any]], namespace: Optional[str] = None, database: Optional[str] = None, ) -> Dict[str, Any]: """ Insert multiple records into a table in a single operation. This tool is optimized for bulk inserts when you need to create many records at once. It's more efficient than calling 'create' multiple times. Each record will get: - An auto-generated unique ID - Automatic created/updated timestamps - Schema validation (if defined) Args: table: The name of the table to insert records into (e.g., "user", "product") data: Array of dictionaries, each representing a record to insert. Example: [ {"name": "Alice", "email": "alice@example.com"}, {"name": "Bob", "email": "bob@example.com"}, {"name": "Charlie", "email": "charlie@example.com"} ] namespace: Optional SurrealDB namespace override. If not provided, uses SURREAL_NAMESPACE env var. database: Optional SurrealDB database override. If not provided, uses SURREAL_DATABASE env var. Returns: A dictionary containing: - success: Boolean indicating if insertion was successful - data: Array of all inserted records with their generated IDs - count: Number of records successfully inserted - error: Error message if insertion failed (only present on failure) Examples: >>> await insert("user", [ ... {"name": "Alice", "role": "admin"}, ... {"name": "Bob", "role": "user"} ... ]) { "success": true, "data": [ {"id": "user:ulid1", "name": "Alice", "role": "admin", "created": "..."}, {"id": "user:ulid2", "name": "Bob", "role": "user", "created": "..."} ], "count": 2 } Note: For single record creation, use the 'create' tool instead. """ try: ns, db = resolve_namespace_database(namespace, database) if not data or not isinstance(data, list): raise ValueError("Data must be a non-empty array of records") logger.info(f"Inserting {len(data)} records into table {table}") # Add timestamps to each record from datetime import datetime, timezone now = datetime.now(timezone.utc) for record in data: record["created"] = record.get("created", now) record["updated"] = record.get("updated", now) result = await repo_insert(table, data, namespace=ns, database=db) # Ensure result is a list if not isinstance(result, list): result = [result] if result else [] return { "success": True, "data": result, "count": len(result) } except Exception as e: logger.error(f"Insert failed for table {table}: {str(e)}") raise Exception(f"Failed to insert records into {table}: {str(e)}")
- Supporting function repo_insert that executes the bulk INSERT operation on the SurrealDB connection, handles RecordID parsing, and manages duplicate errors.async def repo_insert( table: str, data: List[Dict[str, Any]], ignore_duplicates: bool = False, namespace: Optional[str] = None, database: Optional[str] = None, ) -> List[Dict[str, Any]]: """Insert multiple records into a table. Args: table: The table to insert into data: List of records to insert ignore_duplicates: Whether to ignore duplicate key errors namespace: Optional namespace override (uses env var if not provided) database: Optional database override (uses env var if not provided) Returns: The inserted records """ try: async with db_connection(namespace, database) as connection: return parse_record_ids(await connection.insert(table, data)) except Exception as e: if ignore_duplicates and "already contains" in str(e): return [] logger.exception(e) raise RuntimeError("Failed to create record")
- Utility helper to recursively convert SurrealDB RecordID objects to strings in query results, used by repo_insert.def parse_record_ids(obj: Any) -> Any: """Recursively parse and convert RecordIDs into strings.""" if isinstance(obj, dict): return {k: parse_record_ids(v) for k, v in obj.items()} elif isinstance(obj, list): return [parse_record_ids(item) for item in obj] elif isinstance(obj, RecordID): return str(obj) return obj