Uploading Rich Media Attachments to Genesys Cloud Web Messaging via Presigned S3 URLs in Python
What You Will Build
A Python script that requests a presigned S3 upload URL from Genesys Cloud, uploads a local file directly to AWS S3, and sends a Web Messaging guest message containing the attachment metadata. This tutorial uses the Genesys Cloud Messaging API and the purecloud-platform-client-v2 Python SDK. The code covers Python 3.9 and newer.
Prerequisites
- OAuth2 client credentials with scopes:
message:send,webchat:attachment:write,webchat:attachment:read purecloud-platform-client-v2>=130.0.0requests>=2.31.0- Python 3.9+
- A valid Genesys Cloud organization with Web Messaging enabled
- A target Web Messaging channel ID and guest ID for message routing
Authentication Setup
Genesys Cloud uses OAuth2 client credentials flow for server-to-server integrations. The Python SDK handles token caching internally, but you must configure the Configuration object with your environment host, client ID, and client secret. The SDK automatically appends the Bearer token to subsequent API calls.
from purecloud_platform_client_v2 import Configuration, ApiClient
from purecloud_platform_client_v2.api import AuthenticationApi
def get_authenticated_api_client(
environment: str,
client_id: str,
client_secret: str
) -> ApiClient:
"""Initializes and authenticates the Genesys Cloud Python SDK client."""
config = Configuration(
environment=environment,
client_id=client_id,
client_secret=client_secret
)
api_client = ApiClient(configuration=config)
# Verify authentication by fetching a token
auth_api = AuthenticationApi(api_client=api_client)
try:
auth_api.post_oauth2_token(
grant_type="client_credentials",
scope="message:send webchat:attachment:write webchat:attachment:read"
)
except Exception as e:
raise RuntimeError(f"Authentication failed: {e}") from e
return api_client
The post_oauth2_token call validates credentials and stores the access token in the ApiClient instance. Subsequent calls to MessagingApi will automatically attach the Authorization: Bearer <token> header. Token expiration is handled automatically by the SDK, which refreshes credentials when a 401 response is detected.
Implementation
Step 1: Request Presigned Upload URL
Genesys Cloud does not accept binary file uploads directly in the message payload. Instead, you must request a presigned S3 URL. The endpoint /api/v2/messaging/attachments/guest/upload-url returns a temporary PUT URL and an attachment identifier. You must provide the file name, MIME type, and byte size in the request body.
from purecloud_platform_client_v2.api import MessagingApi
from purecloud_platform_client_v2.rest import ApiException
from typing import Dict, Any
def request_presigned_url(
messaging_api: MessagingApi,
file_name: str,
content_type: str,
file_size: int
) -> Dict[str, Any]:
"""Requests a presigned S3 upload URL from Genesys Cloud."""
request_body = {
"fileName": file_name,
"contentType": content_type,
"size": file_size
}
try:
response = messaging_api.post_messaging_attachments_guest_upload_url(
body=request_body
)
return {
"url": response.url,
"method": response.method,
"attachment_id": response.attachment_id,
"headers": response.headers or {}
}
except ApiException as e:
if e.status == 400:
raise ValueError(f"Invalid attachment parameters: {e.body}") from e
if e.status == 403:
raise PermissionError("Missing webchat:attachment:write scope") from e
raise
The response contains the url for the S3 PUT request, the HTTP method (always PUT), the attachment_id required for the message payload, and optional S3 headers (such as x-amz-acl). You must pass these exact headers during the upload step.
Step 2: Upload File to S3
Once you have the presigned URL, you upload the file directly to AWS S3 using a standard HTTP PUT request. The SDK does not manage S3 transfers, so you will use the requests library. You must implement retry logic for 429 Too Many Requests responses, as S3 and Genesys Cloud proxy layers may throttle concurrent uploads.
import requests
import time
from typing import Optional
def upload_to_s3_presigned_url(
presigned_url: str,
file_path: str,
extra_headers: Optional[Dict[str, str]] = None,
max_retries: int = 3
) -> bool:
"""Uploads a local file to S3 using a presigned URL with exponential backoff."""
headers = {"Content-Type": ""}
if extra_headers:
headers.update(extra_headers)
with open(file_path, "rb") as f:
file_data = f.read()
for attempt in range(1, max_retries + 1):
try:
response = requests.put(
presigned_url,
data=file_data,
headers=headers,
timeout=30
)
response.raise_for_status()
return True
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
wait_time = min(2 ** attempt, 30)
print(f"Rate limited (429). Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
raise RuntimeError(f"S3 upload failed with status {response.status_code}: {response.text}") from e
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Network error during S3 upload: {e}") from e
raise RuntimeError("Max retries exceeded for S3 upload")
The function reads the entire file into memory for simplicity. For files larger than 100 MB, you should stream the data using requests.put(url, data=open(file_path, "rb"), headers=headers). The retry loop catches 429 responses and applies exponential backoff up to 30 seconds.
Step 3: Inject Metadata and Send Guest Message
After the S3 upload succeeds, Genesys Cloud indexes the file. You now construct the guest message payload. The attachments array must contain the attachment_id, name, contentType, and size returned or calculated earlier. The endpoint /api/v2/messaging/messages accepts the payload via POST.
from purecloud_platform_client_v2.api import MessagingApi
from purecloud_platform_client_v2.rest import ApiException
def send_guest_message_with_attachment(
messaging_api: MessagingApi,
channel_id: str,
guest_id: str,
text_body: str,
attachment_id: str,
file_name: str,
content_type: str,
file_size: int
) -> Dict[str, Any]:
"""Sends a Web Messaging guest message with attachment metadata."""
message_payload = {
"to": {"id": channel_id},
"from": {"id": guest_id},
"text": text_body,
"attachments": [
{
"id": attachment_id,
"name": file_name,
"contentType": content_type,
"size": file_size
}
]
}
try:
response = messaging_api.post_messaging_messages(body=message_payload)
return {
"message_id": response.id,
"status": response.status,
"created_time": response.created_time
}
except ApiException as e:
if e.status == 400:
raise ValueError(f"Invalid message payload: {e.body}") from e
if e.status == 403:
raise PermissionError("Missing message:send scope") from e
raise
The post_messaging_messages method validates the payload structure. The attachment_id must match the identifier returned in Step 1. If the ID is incorrect or the S3 upload failed, Genesys Cloud returns a 400 error with a validation message.
Complete Working Example
The following script combines authentication, presigned URL generation, S3 upload, and message delivery into a single executable module. Replace the credential placeholders and file paths before execution.
#!/usr/bin/env python3
import sys
import os
import requests
import time
from typing import Dict, Any, Optional
from purecloud_platform_client_v2 import Configuration, ApiClient
from purecloud_platform_client_v2.api import AuthenticationApi, MessagingApi
from purecloud_platform_client_v2.rest import ApiException
def get_authenticated_api_client(
environment: str,
client_id: str,
client_secret: str
) -> ApiClient:
config = Configuration(
environment=environment,
client_id=client_id,
client_secret=client_secret
)
api_client = ApiClient(configuration=config)
auth_api = AuthenticationApi(api_client=api_client)
try:
auth_api.post_oauth2_token(
grant_type="client_credentials",
scope="message:send webchat:attachment:write webchat:attachment:read"
)
except Exception as e:
raise RuntimeError(f"Authentication failed: {e}") from e
return api_client
def request_presigned_url(
messaging_api: MessagingApi,
file_name: str,
content_type: str,
file_size: int
) -> Dict[str, Any]:
request_body = {
"fileName": file_name,
"contentType": content_type,
"size": file_size
}
try:
response = messaging_api.post_messaging_attachments_guest_upload_url(body=request_body)
return {
"url": response.url,
"method": response.method,
"attachment_id": response.attachment_id,
"headers": response.headers or {}
}
except ApiException as e:
if e.status == 400:
raise ValueError(f"Invalid attachment parameters: {e.body}") from e
if e.status == 403:
raise PermissionError("Missing webchat:attachment:write scope") from e
raise
def upload_to_s3_presigned_url(
presigned_url: str,
file_path: str,
extra_headers: Optional[Dict[str, str]] = None,
max_retries: int = 3
) -> bool:
headers = {"Content-Type": ""}
if extra_headers:
headers.update(extra_headers)
with open(file_path, "rb") as f:
file_data = f.read()
for attempt in range(1, max_retries + 1):
try:
response = requests.put(
presigned_url,
data=file_data,
headers=headers,
timeout=30
)
response.raise_for_status()
return True
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
wait_time = min(2 ** attempt, 30)
print(f"Rate limited (429). Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
raise RuntimeError(f"S3 upload failed with status {response.status_code}: {response.text}") from e
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Network error during S3 upload: {e}") from e
raise RuntimeError("Max retries exceeded for S3 upload")
def send_guest_message_with_attachment(
messaging_api: MessagingApi,
channel_id: str,
guest_id: str,
text_body: str,
attachment_id: str,
file_name: str,
content_type: str,
file_size: int
) -> Dict[str, Any]:
message_payload = {
"to": {"id": channel_id},
"from": {"id": guest_id},
"text": text_body,
"attachments": [
{
"id": attachment_id,
"name": file_name,
"contentType": content_type,
"size": file_size
}
]
}
try:
response = messaging_api.post_messaging_messages(body=message_payload)
return {
"message_id": response.id,
"status": response.status,
"created_time": response.created_time
}
except ApiException as e:
if e.status == 400:
raise ValueError(f"Invalid message payload: {e.body}") from e
if e.status == 403:
raise PermissionError("Missing message:send scope") from e
raise
def main():
# Configuration
ENVIRONMENT = "mypurecloud.com"
CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"
FILE_PATH = "/path/to/report.pdf"
CHANNEL_ID = "webchat-channel-uuid"
GUEST_ID = "guest-uuid-or-external-id"
if not os.path.exists(FILE_PATH):
print(f"File not found: {FILE_PATH}")
sys.exit(1)
file_name = os.path.basename(FILE_PATH)
file_size = os.path.getsize(FILE_PATH)
content_type = "application/pdf"
print("Initializing SDK and authenticating...")
api_client = get_authenticated_api_client(ENVIRONMENT, CLIENT_ID, CLIENT_SECRET)
messaging_api = MessagingApi(api_client=api_client)
print("Requesting presigned upload URL...")
url_data = request_presigned_url(messaging_api, file_name, content_type, file_size)
print(f"Presigned URL obtained. Attachment ID: {url_data['attachment_id']}")
print(f"Uploading {file_name} to S3...")
upload_to_s3_presigned_url(
presigned_url=url_data["url"],
file_path=FILE_PATH,
extra_headers=url_data["headers"]
)
print("S3 upload completed successfully.")
print("Sending guest message with attachment metadata...")
result = send_guest_message_with_attachment(
messaging_api=messaging_api,
channel_id=CHANNEL_ID,
guest_id=GUEST_ID,
text_body="Please review the attached document.",
attachment_id=url_data["attachment_id"],
file_name=file_name,
content_type=content_type,
file_size=file_size
)
print(f"Message sent. ID: {result['message_id']}, Status: {result['status']}")
if __name__ == "__main__":
main()
Common Errors and Debugging
Error: 401 Unauthorized
The OAuth token expired or was never generated. The purecloud-platform-client-v2 SDK automatically refreshes tokens, but if you manually construct requests outside the SDK, you must handle expiration. Verify that your client credentials possess the message:send and webchat:attachment:write scopes. Run post_oauth2_token directly to confirm token issuance.
Error: 403 Forbidden
The OAuth client lacks the required scopes. Web Messaging attachments require webchat:attachment:write for URL generation and uploads. Message delivery requires message:send. Add these scopes in the Genesys Cloud Admin console under Platform > OAuth Clients > Scopes.
Error: 400 Bad Request
The attachment payload contains mismatched metadata. The size field in the upload URL request must exactly match the actual file size in bytes. The contentType must be a valid MIME type. If the attachment_id in the message payload does not match the ID returned from the upload URL endpoint, Genesys Cloud rejects the message. Verify the JSON structure matches the schema exactly.
Error: 429 Too Many Requests
S3 or Genesys Cloud rate-limited the presigned URL generation or PUT request. The retry logic in the upload function handles this by waiting 2, 4, 8 seconds up to a maximum of 30 seconds. If you encounter persistent 429 errors, reduce concurrent upload threads or implement a queue with a token bucket algorithm.
Error: 502 Bad Gateway or 504 Gateway Timeout
The Genesys Cloud messaging service is experiencing high latency or the S3 presigned URL expired. Presigned URLs typically expire within 5 to 10 minutes. If your application caches URLs or delays the PUT request, the S3 server returns a signature failure. Always generate the URL immediately before the upload step.