surreal-mcp

Instructions

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)}")

• surreal_mcp/database/__init__.py:252-279 (helper)

  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")

• surreal_mcp/database/__init__.py:18-27 (helper)

  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