insert
Insert multiple records into a database table in a single operation. This bulk insert tool creates many records at once with auto-generated IDs, timestamps, and schema validation for efficient data management.
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"} ]
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 |
|---|---|---|---|
| data | Yes | ||
| table | Yes |
Implementation Reference
- surreal_mcp/server.py:672-751 (handler)The primary handler for the 'insert' MCP tool. It resolves namespace/database, validates input, adds timestamps to records, calls the repo_insert helper, and returns formatted results including success status, data, and count.@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 helper function repo_insert that executes the actual bulk insert using the SurrealDB client connection pool, parses RecordIDs, and handles 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")