Handling Customer File Uploads in Web Messaging: MIME Validation and Size Limits

What You Will Build

• You will build a server-side validation layer that intercepts file uploads from the Genesys Cloud Web Messaging SDK, verifies MIME types against an allowlist, and enforces strict file size limits before forwarding the media to the platform.
• This tutorial uses the Genesys Cloud V2 API (/api/v2/externalcontacts) and the Python genesyscloud SDK to manage external contact media attachments.
• The implementation covers Python for the backend validation service and JavaScript for the client-side Web Messaging configuration.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant) for server-side API calls.
• Required Scopes:
  • externalcontacts:externalcontact:write (to create/update external contacts with media)
  • externalcontacts:externalcontact:read (to verify existing contacts if needed)
• SDK Version: genesyscloud Python SDK v12.0.0 or higher.
• Runtime Requirements: Python 3.9+ with pip installed.
• External Dependencies:
  • requests (for HTTP handling)
  • python-magic (for robust MIME type detection)
  • flask (for the example validation endpoint)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-side file handling, you must use the Client Credentials grant flow. This requires storing your Client ID and Client Secret securely.

Python Token Acquisition

The following code demonstrates how to acquire and cache an access token. In production, you should implement token expiration checks and automatic refresh logic.

import os
import time
import requests
from typing import Optional, Dict

# Configuration
GENESYS_CLOUD_REGION = "mypurecloud.com"  # Change to your region (e.g., usw2.pure.cloud)
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")

def get_access_token() -> str:
    """
    Retrieves an OAuth access token from Genesys Cloud using Client Credentials.
    """
    url = f"https://{GENESYS_CLOUD_REGION}/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }

    try:
        response = requests.post(url, data=payload)
        response.raise_for_status()
        token_data = response.json()
        return token_data["access_token"]
    except requests.exceptions.HTTPError as e:
        if response.status_code == 401:
            raise ValueError("Invalid Client ID or Secret.") from e
        raise Exception(f"Failed to obtain token: {e}") from e

# Cache token to avoid excessive calls
_token_cache: Dict[str, float] = {}

def get_cached_token() -> str:
    if "token" in _token_cache and time.time() < _token_cache["expiry"]:
        return _token_cache["token"]
    
    token = get_access_token()
    _token_cache["token"] = token
    _token_cache["expiry"] = time.time() + 5400  # Cache for 90 minutes (token lasts 1 hour)
    return token

Implementation

Step 1: Configure Web Messaging File Upload Limits

Before writing backend code, you must understand the constraints imposed by the Genesys Cloud Web Messaging SDK. The SDK allows you to configure allowed file types and maximum file sizes on the client side. However, client-side validation is not security; it is only UX. You must enforce these rules on the server.

Client-Side Configuration (JavaScript)

In your Web Messaging initialization, you define the fileAttachment options.

// web-messaging-config.js
const config = {
    // ... other config options
    fileAttachment: {
        allowedFileTypes: [
            "image/png",
            "image/jpeg",
            "application/pdf",
            "text/plain"
        ],
        maxSize: 5242880 // 5 MB in bytes
    }
};

// Initialize Web Messaging
genesysWebMessaging.init(config);

Critical Note: The maxSize parameter in the SDK prevents the UI from sending files larger than the limit. It does not prevent a malicious actor from bypassing the SDK and sending a large file directly to your backend or Genesys API. Your backend must re-validate this.

Step 2: Backend Validation and MIME Detection

You will create a Flask endpoint that receives the file from the client. This endpoint performs two critical checks:

1. File Size: Ensures the file does not exceed the platform limit (default 10 MB for Web Messaging media, but you may set a stricter policy).
2. MIME Type: Uses python-magic to inspect the file header (magic bytes) rather than trusting the Content-Type header sent by the browser.

The Validation Endpoint

import magic
import os
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename

app = Flask(__name__)

# Configuration
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5 MB
ALLOWED_MIMES = {
    "image/png",
    "image/jpeg",
    "application/pdf",
    "text/plain"
}

@app.route("/upload", methods=["POST"])
def handle_upload():
    """
    Receives a file from the client, validates size and MIME type,
    and prepares it for Genesys Cloud integration.
    """
    # Check if the post request has the file part
    if "file" not in request.files:
        return jsonify({"error": "No file part in the request"}), 400

    file = request.files["file"]

    if file.filename == "":
        return jsonify({"error": "No selected file"}), 400

    # 1. Validate File Size
    # Read the file into memory to check size and MIME type
    # Note: For very large files, consider streaming, but 5MB is small enough for memory.
    file_content = file.read()
    
    if len(file_content) > MAX_FILE_SIZE:
        return jsonify({
            "error": "File size exceeds the limit",
            "limit_bytes": MAX_FILE_SIZE,
            "file_size_bytes": len(file_content)
        }), 413  # 413 Payload Too Large

    # 2. Validate MIME Type using python-magic
    # python-magic inspects the binary content, not the header
    detected_mime = magic.from_buffer(file_content, mime=True)

    if detected_mime not in ALLOWED_MIMES:
        return jsonify({
            "error": "File type not allowed",
            "detected_mime": detected_mime,
            "allowed_mimes": list(ALLOWED_MIMES)
        }), 415  # 415 Unsupported Media Type

    # 3. Secure the filename
    safe_filename = secure_filename(file.filename)
    
    # Proceed to upload to Genesys Cloud
    try:
        upload_to_genesys(file_content, detected_mime, safe_filename)
        return jsonify({"message": "File uploaded successfully"}), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

def upload_to_genesys(content: bytes, mime_type: str, filename: str):
    """
    Placeholder for the actual upload logic.
    In Step 3, we will implement the Genesys Cloud API call.
    """
    pass

Step 3: Uploading Media to Genesys Cloud

Genesys Cloud does not have a single “upload file” endpoint that returns a generic URL. Instead, media files in Web Messaging are typically attached to External Contacts or Conversations. For Web Messaging, the standard pattern is to associate the file with an External Contact, which represents the customer session.

If you are building a custom integration that mimics the Web Messaging SDK behavior, you must:

1. Create or retrieve the External Contact.
2. Upload the media as an attachment to that External Contact.

Using the Genesys Cloud Python SDK

The genesyscloud SDK simplifies the multipart form-data upload required by the API.

from genesyscloud.rest import Configuration
from genesyscloud.external_contacts.rest import ExternalContactsApi
from genesyscloud.external_contacts.model import ExternalContact
import io

def upload_to_genesys(content: bytes, mime_type: str, filename: str, contact_id: str):
    """
    Uploads a file as an attachment to an existing External Contact.
    """
    # Initialize the API client
    configuration = Configuration()
    configuration.host = f"https://{GENESYS_CLOUD_REGION}"
    configuration.access_token = get_cached_token()

    external_contacts_api = ExternalContactsApi(configuration)

    # Prepare the file for upload
    file_stream = io.BytesIO(content)
    file_stream.name = filename  # SDK requires the file object to have a 'name' attribute

    # The API endpoint is: POST /api/v2/externalcontacts/{externalContactId}/attachments
    # This endpoint supports multipart/form-data

    try:
        # Upload the attachment
        # Note: The SDK method may vary slightly by version. 
        # In v12+, it is often done via a specific upload method or generic post.
        # Here we use the generic approach for clarity if specific method is missing in SDK version.
        
        # However, the standard SDK method for uploading an attachment to a contact is:
        # external_contacts_api.post_external_contacts_external_contact_attachments(
        #     external_contact_id=contact_id,
        #     file=file_stream,
        #     file_name=filename
        # )
        
        # Since the specific method signature can be complex, let's use the requests library 
        # for the final upload to ensure exact control over multipart fields, 
        # as the SDK sometimes abstracts this differently.
        
        upload_file_with_requests(content, filename, contact_id)
        
    except Exception as e:
        raise Exception(f"Failed to upload to Genesys Cloud: {str(e)}")

def upload_file_with_requests(content: bytes, filename: str, contact_id: str):
    """
    Direct API call to upload media to an External Contact.
    Endpoint: POST /api/v2/externalcontacts/{externalContactId}/attachments
    """
    url = f"https://{GENESYS_CLOUD_REGION}/api/v2/externalcontacts/{contact_id}/attachments"
    headers = {
        "Authorization": f"Bearer {get_cached_token()}",
        "Accept": "application/json"
        # Do not set Content-Type; requests will set it to multipart/form-data with boundary
    }

    files = {
        "file": (filename, content, "application/octet-stream")
    }

    response = requests.post(url, headers=headers, files=files)
    
    if response.status_code == 401:
        raise Exception("Authentication failed. Token may be expired.")
    elif response.status_code == 403:
        raise Exception("Insufficient permissions. Check OAuth scopes.")
    elif response.status_code == 404:
        raise Exception(f"External Contact {contact_id} not found.")
    elif response.status_code == 413:
        raise Exception("File too large for Genesys Cloud limits.")
    elif response.status_code == 429:
        raise Exception("Rate limited. Implement retry logic.")
    else:
        response.raise_for_status()
        
    return response.json()

Step 4: Handling Large Files and Pagination

Genesys Cloud has a default media attachment size limit of 10 MB per file. If you need to handle larger files, you cannot simply upload them to the External Contact attachment endpoint. Instead, you must use a different strategy:

1. Upload to External Storage: Upload the file to Amazon S3, Azure Blob Storage, or Google Cloud Storage.
2. Generate a Signed URL: Create a time-limited, pre-signed URL for the file.
3. Send the URL in a Message: Send the URL as a text message or a rich card in the Web Messaging conversation.

This approach bypasses the 10 MB attachment limit. The following code shows how to send a message with a file link instead of an attachment.

from genesyscloud.conversations_messages.rest import ConversationsMessagesApi
from genesyscloud.conversations_messages.model import MessageSendRequest

def send_file_link(contact_id: str, conversation_id: str, file_url: str, filename: str):
    """
    Sends a message containing a link to the file instead of uploading the file itself.
    """
    configuration = Configuration()
    configuration.host = f"https://{GENESYS_CLOUD_REGION}"
    configuration.access_token = get_cached_token()

    messages_api = ConversationsMessagesApi(configuration)

    # Construct a rich text message or plain text
    message_body = f"Please download the attached file: <a href='{file_url}'>{filename}</a>"

    send_request = MessageSendRequest(
        to=[contact_id],
        type="text",
        body=message_body
    )

    # POST /api/v2/conversations/messages
    # Note: You must specify the conversation ID in the header or use the conversation-specific endpoint
    # For Web Messaging, the conversation ID is usually known from the session.
    
    headers = {
        "X-Genesys-Application-Id": "web-messaging", # Optional but recommended
        "Conversation-Id": conversation_id # Required for context
    }

    response = messages_api.post_conversations_messages(send_request, headers=headers)
    
    if response.status_code != 200:
        raise Exception(f"Failed to send message: {response.text}")
        
    return response

Complete Working Example

Below is the complete, runnable Flask application that integrates validation, Genesys Cloud authentication, and file upload.

import os
import time
import io
import requests
import magic
from flask import Flask, request, jsonify
from werkzeug.utils import secure_filename
from typing import Dict

# --- Configuration ---
GENESYS_CLOUD_REGION = "mypurecloud.com"
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID", "your-client-id")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET", "your-client-secret")

MAX_FILE_SIZE = 5 * 1024 * 1024  # 5 MB
ALLOWED_MIMES = {
    "image/png",
    "image/jpeg",
    "application/pdf",
    "text/plain"
}

_token_cache: Dict[str, float] = {}

app = Flask(__name__)

# --- Authentication ---
def get_access_token() -> str:
    url = f"https://{GENESYS_CLOUD_REGION}/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    try:
        response = requests.post(url, data=payload)
        response.raise_for_status()
        return response.json()["access_token"]
    except Exception as e:
        raise Exception(f"Token acquisition failed: {e}")

def get_cached_token() -> str:
    if "token" in _token_cache and time.time() < _token_cache["expiry"]:
        return _token_cache["token"]
    token = get_access_token()
    _token_cache["token"] = token
    _token_cache["expiry"] = time.time() + 5400
    return token

# --- Genesys Cloud Upload ---
def upload_to_genesys(content: bytes, filename: str, contact_id: str):
    url = f"https://{GENESYS_CLOUD_REGION}/api/v2/externalcontacts/{contact_id}/attachments"
    headers = {
        "Authorization": f"Bearer {get_cached_token()}",
        "Accept": "application/json"
    }
    files = {
        "file": (filename, content, "application/octet-stream")
    }
    
    response = requests.post(url, headers=headers, files=files)
    
    if response.status_code == 401:
        raise Exception("Auth failed")
    elif response.status_code == 403:
        raise Exception("Permission denied")
    elif response.status_code == 404:
        raise Exception("Contact not found")
    elif response.status_code == 429:
        raise Exception("Rate limited")
    
    response.raise_for_status()
    return response.json()

# --- Main Endpoint ---
@app.route("/upload", methods=["POST"])
def handle_upload():
    if "file" not in request.files:
        return jsonify({"error": "No file part"}), 400

    file = request.files["file"]
    if file.filename == "":
        return jsonify({"error": "No selected file"}), 400

    file_content = file.read()
    
    # 1. Size Check
    if len(file_content) > MAX_FILE_SIZE:
        return jsonify({"error": "File too large"}), 413

    # 2. MIME Check
    detected_mime = magic.from_buffer(file_content, mime=True)
    if detected_mime not in ALLOWED_MIMES:
        return jsonify({"error": "Invalid file type", "detected": detected_mime}), 415

    # 3. Secure Filename
    safe_filename = secure_filename(file.filename)

    # 4. Upload to Genesys
    # In a real scenario, you would pass the actual External Contact ID
    # For this example, we assume the contact ID is passed in the request body or header
    contact_id = request.form.get("contact_id")
    if not contact_id:
        return jsonify({"error": "Contact ID required"}), 400

    try:
        result = upload_to_genesys(file_content, safe_filename, contact_id)
        return jsonify({"message": "Success", "result": result}), 200
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == "__main__":
    app.run(debug=True)

Common Errors & Debugging

Error: 413 Payload Too Large

Cause: The file size exceeds the MAX_FILE_SIZE defined in your backend or the Genesys Cloud platform limit (10 MB for attachments).

Fix:

1. Check your backend MAX_FILE_SIZE constant.
2. If you need larger files, switch to the “Upload to External Storage” pattern described in Step 4.
3. Ensure your web server (Nginx, Apache) allows the request body size. For Nginx, add client_max_body_size 5M; to your server block.

Error: 415 Unsupported Media Type

Cause: The python-magic library detected a MIME type not in your ALLOWED_MIMES set.

Fix:

1. Log the detected_mime value to identify what the user is trying to upload.
2. Add the MIME type to ALLOWED_MIMES if it is safe.
3. Ensure python-magic is installed correctly. On macOS, you may need brew install libmagic. On Ubuntu, apt-get install libmagic1.

Error: 401 Unauthorized

Cause: The OAuth token is invalid, expired, or the Client ID/Secret is incorrect.

Fix:

1. Verify GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET in your environment variables.
2. Check the token expiration logic in get_cached_token().
3. Ensure the OAuth client has the externalcontacts:externalcontact:write scope.

Error: 404 Not Found

Cause: The contact_id provided does not exist in Genesys Cloud.

Fix:

1. Ensure the Web Messaging session has successfully created an External Contact before attempting to upload.
2. Log the contact_id and verify it in the Genesys Cloud Admin Console under Contacts > External Contacts.

Official References

• Genesys Cloud Web Messaging SDK Documentation
• External Contacts API Reference
• Python SDK Documentation
• OAuth 2.0 Client Credentials Grant