Using a Screenshot API in AWS Lambda: Serverless Visual Capture Patterns

Most screenshot API guides assume a long-running server process: a Flask app, a Django view, a Node.js Express route that accepts a request, calls the API, and returns a response. That model works, but it is not the only model. Serverless functions — AWS Lambda in particular — offer a compelling alternative: pay per invocation, no idle capacity, scales to zero between requests, and no server maintenance.

The challenge with Lambda is that it introduces constraints that a long-running server does not have: execution timeouts (default 3 seconds, configurable to 15 minutes), response payload size limits (6MB synchronous, 256KB for Lambda Function URLs with response streaming), and cold start latency. This guide covers how to work within those constraints to build reliable screenshot pipelines on Lambda.

Why Lambda for Screenshot Workflows

Screenshot API calls are ideal Lambda workloads:

• Irregular demand: Screenshot generation is often burst-driven — a batch of reports at month-end, certificate generation at course completion, audit snapshots on regulatory trigger. Lambda scales to zero between bursts and costs nothing at rest.
• No browser cold start: Since you are calling an external API rather than running a headless browser inside Lambda, the cold start cost is just Python runtime initialization — typically under 300ms.
• Natural event wiring: Lambda integrates natively with S3 events, SQS, EventBridge, API Gateway, and Step Functions — all natural triggers for screenshot workflows.
• Cost alignment: Per-invocation pricing aligns with per-screenshot billing from the API. Both costs scale linearly with usage.

Basic Lambda Function

import json
import os
import urllib.request
import urllib.parse
import urllib.error

def lambda_handler(event, context):
    """
    Minimal screenshot Lambda. Accepts:
    - event["url"]: target URL
    - event["format"]: "png" | "webp" | "jpeg" (optional, default "webp")
    Returns: {"screenshot_url": "...", "bytes": N}
    """
    target_url = event.get("url")
    if not target_url:
        return {"statusCode": 400, "body": json.dumps({"error": "url is required"})}

    api_key = os.environ["SCREENSHOT_API_KEY"]
    fmt = event.get("format", "webp")

    params = urllib.parse.urlencode({
        "url": target_url,
        "format": fmt,
        "full_page": "true",
        "block_ads": "true",
    })

    req = urllib.request.Request(
        f"https://hermesforge.dev/api/screenshot?{params}",
        headers={"X-API-Key": api_key},
    )

    try:
        with urllib.request.urlopen(req, timeout=25) as resp:
            image_bytes = resp.read()
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        return {"statusCode": e.code, "body": body}
    except urllib.error.URLError as e:
        raise RuntimeError(f"Screenshot API unreachable: {e.reason}") from e

    return {
        "statusCode": 200,
        "bytes": len(image_bytes),
        # Returning raw bytes inline only works for small images.
        # For production, store in S3 — see below.
        "image_base64": __import__("base64").b64encode(image_bytes).decode(),
    }

Timeout configuration: Set your Lambda timeout to at least 30 seconds. Screenshot APIs can take 10–20 seconds for JavaScript-heavy pages. The function itself uses a 25-second urllib timeout, which surfaces as a URLError rather than a silent Lambda timeout.

aws lambda update-function-configuration \
  --function-name screenshot-capture \
  --timeout 30 \
  --memory-size 256

S3 Storage Pattern

Returning image bytes inline hits Lambda's 6MB synchronous response limit quickly. Store screenshots in S3 instead, return a presigned URL:

import json
import os
import urllib.request
import urllib.parse
import urllib.error
import boto3
from datetime import datetime, UTC

s3 = boto3.client("s3")
BUCKET = os.environ["SCREENSHOT_BUCKET"]
API_KEY = os.environ["SCREENSHOT_API_KEY"]


def capture_screenshot(url: str, fmt: str = "webp") -> bytes:
    params = urllib.parse.urlencode({
        "url": url,
        "format": fmt,
        "full_page": "true",
        "block_ads": "true",
        "viewport_width": "1280",
        "viewport_height": "800",
    })
    req = urllib.request.Request(
        f"https://hermesforge.dev/api/screenshot?{params}",
        headers={"X-API-Key": API_KEY},
    )
    with urllib.request.urlopen(req, timeout=25) as resp:
        return resp.read()


def store_in_s3(image_bytes: bytes, key: str, fmt: str) -> str:
    """Store screenshot in S3, return presigned URL valid for 1 hour."""
    content_types = {"webp": "image/webp", "png": "image/png", "jpeg": "image/jpeg"}
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=image_bytes,
        ContentType=content_types.get(fmt, "image/webp"),
        ServerSideEncryption="AES256",
    )
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": BUCKET, "Key": key},
        ExpiresIn=3600,
    )


def lambda_handler(event, context):
    target_url = event.get("url")
    if not target_url:
        return {"statusCode": 400, "body": json.dumps({"error": "url is required"})}

    fmt = event.get("format", "webp")
    ts = datetime.now(UTC).strftime("%Y/%m/%d/%H%M%S")
    safe_host = urllib.parse.urlparse(target_url).netloc.replace(".", "-")
    key = f"screenshots/{ts}/{safe_host}.{fmt}"

    try:
        image_bytes = capture_screenshot(target_url, fmt)
    except urllib.error.HTTPError as e:
        return {"statusCode": e.code, "body": e.read().decode()}

    presigned_url = store_in_s3(image_bytes, key, fmt)

    return {
        "statusCode": 200,
        "s3_key": key,
        "presigned_url": presigned_url,
        "bytes": len(image_bytes),
        "expires_in": 3600,
    }

Retry with Exponential Backoff

Screenshot APIs are external services — transient failures happen. Lambda does not retry synchronous invocations automatically. Add your own retry loop for resilience:

import time
import urllib.error


def capture_with_retry(
    url: str,
    fmt: str = "webp",
    max_attempts: int = 3,
    base_delay: float = 1.0,
) -> bytes:
    """Capture screenshot with exponential backoff on transient errors."""
    last_error = None

    for attempt in range(max_attempts):
        try:
            return capture_screenshot(url, fmt)
        except urllib.error.HTTPError as e:
            # 429 rate limit and 5xx server errors are retryable
            if e.code == 429 or e.code >= 500:
                last_error = e
                delay = base_delay * (2 ** attempt)
                print(f"Attempt {attempt + 1}/{max_attempts} failed ({e.code}), "
                      f"retrying in {delay:.1f}s")
                time.sleep(delay)
            else:
                # 4xx client errors are not retryable
                raise
        except urllib.error.URLError as e:
            # Network errors are retryable
            last_error = e
            delay = base_delay * (2 ** attempt)
            print(f"Network error on attempt {attempt + 1}: {e.reason}, "
                  f"retrying in {delay:.1f}s")
            time.sleep(delay)

    raise RuntimeError(f"All {max_attempts} attempts failed") from last_error

For 429 responses specifically, respect the Retry-After header if present:

except urllib.error.HTTPError as e:
    if e.code == 429:
        retry_after = float(e.headers.get("Retry-After", base_delay * (2 ** attempt)))
        time.sleep(retry_after)

SQS-Triggered Batch Processing

For batch screenshot workflows — generating 500 report PDFs at month-end, capturing 200 regulatory disclosure pages before a filing deadline — SQS provides natural batching with concurrency control:

import json
import os
import urllib.parse
import urllib.request
import boto3

s3 = boto3.client("s3")
BUCKET = os.environ["SCREENSHOT_BUCKET"]
API_KEY = os.environ["SCREENSHOT_API_KEY"]


def lambda_handler(event, context):
    """Process SQS batch of screenshot requests."""
    results = []
    failed_message_ids = []

    for record in event["Records"]:
        message_id = record["messageId"]
        try:
            body = json.loads(record["body"])
            url = body["url"]
            fmt = body.get("format", "webp")
            reference_id = body.get("reference_id", message_id)

            image_bytes = capture_with_retry(url, fmt)

            key = f"batch/{reference_id}.{fmt}"
            s3.put_object(
                Bucket=BUCKET,
                Key=key,
                Body=image_bytes,
                ContentType=f"image/{fmt}",
                ServerSideEncryption="AES256",
            )

            results.append({
                "messageId": message_id,
                "reference_id": reference_id,
                "s3_key": key,
                "bytes": len(image_bytes),
                "status": "ok",
            })
            print(f"[OK] {reference_id}: {len(image_bytes)} bytes → s3://{BUCKET}/{key}")

        except Exception as exc:
            print(f"[FAIL] {message_id}: {exc}")
            failed_message_ids.append(message_id)
            results.append({
                "messageId": message_id,
                "status": "error",
                "error": str(exc),
            })

    # Return failed message IDs so SQS requeues them
    return {
        "batchItemFailures": [
            {"itemIdentifier": mid} for mid in failed_message_ids
        ]
    }

Configure the SQS event source with a batch size of 5–10 and a visibility timeout at least 2x your Lambda timeout (60s for a 30s function). Set ReportBatchItemFailures on the event source mapping so only genuinely failed messages are requeued.

EventBridge Scheduled Captures

For daily regulatory snapshots, monitoring dashboards, or scheduled report generation:

# EventBridge rule (CDK):
# schedule: events.Schedule.cron(hour="13", minute="30", weekDay="MON-FRI")
# Captures at 13:30Z (09:30 ET) each trading day

def lambda_handler(event, context):
    """
    EventBridge-triggered daily capture.
    event["source"] == "aws.events" for scheduled invocations.
    """
    capture_date = datetime.now(UTC).strftime("%Y-%m-%d")

    pages = [
        {"url": "https://example.com/dashboard/risk", "label": "risk"},
        {"url": "https://example.com/dashboard/positions", "label": "positions"},
        {"url": "https://example.com/dashboard/pnl", "label": "pnl"},
    ]

    captured = []
    for page in pages:
        image_bytes = capture_with_retry(page["url"])
        key = f"daily/{capture_date}/{page['label']}.webp"
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=image_bytes,
            ContentType="image/webp",
            ServerSideEncryption="AES256",
        )
        captured.append({"label": page["label"], "key": key, "bytes": len(image_bytes)})
        print(f"Captured {page['label']}: {len(image_bytes)} bytes")

    return {"date": capture_date, "captured": captured}

CDK Deployment

from aws_cdk import (
    Stack, Duration, aws_lambda as lambda_, aws_s3 as s3,
    aws_iam as iam, aws_sqs as sqs, aws_lambda_event_sources as sources,
    aws_events as events, aws_events_targets as targets,
)
from constructs import Construct


class ScreenshotStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs):
        super().__init__(scope, construct_id, **kwargs)

        # S3 bucket for screenshot storage
        bucket = s3.Bucket(
            self, "ScreenshotBucket",
            encryption=s3.BucketEncryption.S3_MANAGED,
            lifecycle_rules=[
                s3.LifecycleRule(
                    id="expire-presigned-originals",
                    prefix="screenshots/",
                    expiration=Duration.days(90),
                )
            ],
        )

        # Lambda function
        fn = lambda_.Function(
            self, "ScreenshotFunction",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="handler.lambda_handler",
            code=lambda_.Code.from_asset("lambda_src"),
            timeout=Duration.seconds(30),
            memory_size=256,
            environment={
                "SCREENSHOT_BUCKET": bucket.bucket_name,
                "SCREENSHOT_API_KEY": "{{resolve:ssm:/hermes/screenshot-api-key}}",
            },
        )

        bucket.grant_put(fn)
        bucket.grant_read(fn)

        # SQS queue for batch processing
        queue = sqs.Queue(
            self, "ScreenshotQueue",
            visibility_timeout=Duration.seconds(60),  # 2x Lambda timeout
        )

        fn.add_event_source(
            sources.SqsEventSource(
                queue,
                batch_size=5,
                report_batch_item_failures=True,
            )
        )

        # Daily scheduled capture (weekdays at 13:30Z)
        rule = events.Rule(
            self, "DailyCapture",
            schedule=events.Schedule.cron(
                hour="13", minute="30", week_day="MON-FRI"
            ),
        )
        rule.add_target(targets.LambdaFunction(fn))

Store the API key in SSM Parameter Store as a SecureString:

aws ssm put-parameter \
  --name /hermes/screenshot-api-key \
  --value "your-api-key-here" \
  --type SecureString

Lambda Layers for Dependencies

If you need requests or boto3 (boto3 is available in the Lambda runtime by default, but pinning is safer):

# requirements.txt
requests==2.32.3

# Build and publish a Lambda Layer
mkdir -p layer/python
pip install -r requirements.txt -t layer/python
cd layer && zip -r ../screenshot-layer.zip python/
aws lambda publish-layer-version \
  --layer-name screenshot-deps \
  --zip-file fileb://../screenshot-layer.zip \
  --compatible-runtimes python3.12

The examples above use only the standard library (urllib, json, os, base64, time) to avoid the layer dependency entirely for the core screenshot call.

Response Size and Streaming

Lambda's synchronous response limit is 6MB (compressed). A full-page WebP screenshot of a content-heavy page can exceed this. Three approaches:

1. Always store in S3, return presigned URL — recommended for production. No size limit, URL is small.
2. Lambda response streaming — available with Function URLs using RESPONSE_STREAM invoke mode. Streams the image bytes as they arrive. Requires the Lambda Web Adapter or custom streaming handler.
3. Reduce image size — use viewport_height to capture above-the-fold only, or use JPEG at 80% quality for non-archival captures.

For most workflows, option 1 (S3 + presigned URL) is the right default. Option 3 is appropriate for lightweight previews where full-page capture is not required.

Cost Model

For 1000 screenshots/month: Lambda compute ≈ $0.10, S3 ≈ $0.075, API ≈ $9. The API cost dominates at moderate volumes. At very high volumes (>30k/month), the API pricing tier matters more than Lambda optimization.

Summary

Lambda and screenshot APIs are a natural fit: event-driven, pay-per-call, no idle capacity. The key configuration decisions are timeout (30s minimum), memory (256MB is sufficient — no browser), storage (S3 + presigned URL, not inline bytes), and retry logic (exponential backoff with 429 handling). The CDK pattern above gives you a production-ready deployment with SQS batching and EventBridge scheduling in under 100 lines of infrastructure code.

Get a free API key at hermesforge.dev to start building.