Using a Screenshot API in Azure Functions: Serverless Visual Capture on Azure

Azure Functions rounds out the serverless platform arc. Where AWS Lambda integrates with SQS and EventBridge, and GCF integrates with Pub/Sub and Cloud Scheduler, Azure Functions integrates with Service Bus, Event Grid, and Azure Monitor — with managed identity as the idiomatic credential pattern rather than an explicit secrets service. This guide covers Azure-specific patterns for screenshot API integrations, with Bicep deployment throughout.

Azure Functions vs Lambda vs GCF

Azure's managed identity pattern is the most elegant of the three: the function's system-assigned identity is granted RBAC roles directly on resources (Key Vault, Blob Storage, Service Bus), with no credentials to manage, rotate, or store anywhere.

Project Structure

screenshot-function/
├── host.json
├── requirements.txt
├── local.settings.json
├── capture_http/
│   ├── __init__.py
│   └── function.json
├── capture_timer/
│   ├── __init__.py
│   └── function.json
└── capture_servicebus/
    ├── __init__.py
    └── function.json

// host.json
{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": { "isEnabled": true }
    }
  },
  "functionTimeout": "00:10:00"
}

# requirements.txt
azure-functions
azure-storage-blob>=12.0.0
azure-keyvault-secrets>=4.0.0
azure-identity>=1.0.0

HTTP-Triggered Function

# capture_http/__init__.py
import azure.functions as func
import json
import os
import urllib.request
import urllib.parse
import urllib.error
import base64

def main(req: func.HttpRequest) -> func.HttpResponse:
    """HTTP-triggered screenshot capture."""
    try:
        body = req.get_json()
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "Invalid JSON"}),
            status_code=400,
            mimetype="application/json",
        )

    target_url = body.get("url")
    if not target_url:
        return func.HttpResponse(
            json.dumps({"error": "url is required"}),
            status_code=400,
            mimetype="application/json",
        )

    api_key = _get_api_key()
    fmt = body.get("format", "webp")

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

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

    try:
        with urllib.request.urlopen(req_obj, timeout=25) as resp:
            image_bytes = resp.read()
    except urllib.error.HTTPError as e:
        return func.HttpResponse(
            e.read().decode(),
            status_code=e.code,
            mimetype="application/json",
        )
    except urllib.error.URLError as e:
        return func.HttpResponse(
            json.dumps({"error": str(e.reason)}),
            status_code=502,
            mimetype="application/json",
        )

    return func.HttpResponse(
        json.dumps({
            "bytes": len(image_bytes),
            "image_base64": base64.b64encode(image_bytes).decode(),
        }),
        status_code=200,
        mimetype="application/json",
    )


# Module-level cache — survives warm invocations
_API_KEY: str | None = None

def _get_api_key() -> str:
    global _API_KEY
    if _API_KEY is not None:
        return _API_KEY

    # Try environment variable first (local dev / explicit config)
    key = os.environ.get("SCREENSHOT_API_KEY")
    if key:
        _API_KEY = key
        return _API_KEY

    # Production: fetch from Key Vault via managed identity
    from azure.identity import DefaultAzureCredential
    from azure.keyvault.secrets import SecretClient

    vault_url = os.environ["KEY_VAULT_URL"]
    credential = DefaultAzureCredential()
    client = SecretClient(vault_url=vault_url, credential=credential)
    _API_KEY = client.get_secret("screenshot-api-key").value
    return _API_KEY

// capture_http/function.json
{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": ["post"]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

Blob Storage Integration

# Shared storage helper — import in each function
import os
import urllib.parse
from datetime import datetime, UTC, timedelta
from azure.storage.blob import BlobServiceClient, generate_blob_sas, BlobSasPermissions
from azure.identity import DefaultAzureCredential

def store_screenshot(image_bytes: bytes, target_url: str, fmt: str) -> dict:
    """Store screenshot in Azure Blob Storage, return SAS URL."""
    account_name = os.environ["STORAGE_ACCOUNT_NAME"]
    container_name = os.environ["STORAGE_CONTAINER_NAME"]

    # Use managed identity for auth — no connection string or key needed
    credential = DefaultAzureCredential()
    account_url = f"https://{account_name}.blob.core.windows.net"
    service_client = BlobServiceClient(account_url=account_url, credential=credential)

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

    content_types = {"webp": "image/webp", "png": "image/png", "jpeg": "image/jpeg"}
    blob_client = service_client.get_blob_client(container=container_name, blob=blob_name)
    blob_client.upload_blob(
        image_bytes,
        overwrite=False,
        content_settings={"content_type": content_types.get(fmt, "image/webp")},
    )

    # Generate SAS URL valid for 1 hour
    # Requires Storage Blob Data Contributor role on the container
    sas_token = generate_blob_sas(
        account_name=account_name,
        container_name=container_name,
        blob_name=blob_name,
        account_key=None,  # Not needed with managed identity
        user_delegation_key=service_client.get_user_delegation_key(
            key_start_time=datetime.now(UTC),
            key_expiry_time=datetime.now(UTC) + timedelta(hours=1),
        ),
        permission=BlobSasPermissions(read=True),
        expiry=datetime.now(UTC) + timedelta(hours=1),
    )

    return {
        "blob_name": blob_name,
        "sas_url": f"{account_url}/{container_name}/{blob_name}?{sas_token}",
        "bytes": len(image_bytes),
    }

Timer-Triggered Daily Captures

# capture_timer/__init__.py
import azure.functions as func
import logging
import os
import urllib.request
import urllib.parse
import urllib.error
from datetime import datetime, UTC

logger = logging.getLogger(__name__)


def main(mytimer: func.TimerRequest) -> None:
    """
    Timer-triggered daily regulatory capture.
    Runs at 13:30 UTC Monday-Friday (cron: 0 30 13 * * 1-5)
    """
    if mytimer.past_due:
        logger.warning("Timer is running late — previous execution may have failed")

    capture_date = datetime.now(UTC).strftime("%Y-%m-%d")
    pages = [
        {"url": "https://example.com/reports/daily-risk", "label": "daily-risk"},
        {"url": "https://example.com/reports/positions", "label": "positions"},
        {"url": "https://example.com/reports/exposure", "label": "exposure"},
    ]

    api_key = _get_api_key()
    success = 0

    for page in pages:
        try:
            image_bytes = _capture_with_retry(page["url"], api_key)
            result = store_screenshot(image_bytes, page["url"], "webp")
            logger.info(f"[OK] {page['label']}: {result['bytes']} bytes → {result['blob_name']}")
            success += 1
        except Exception as exc:
            logger.error(f"[FAIL] {page['label']}: {exc}")

    logger.info(f"Daily capture complete: {success}/{len(pages)} succeeded for {capture_date}")

// capture_timer/function.json
{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "0 30 13 * * 1-5"
    }
  ]
}

Service Bus-Triggered Batch Processing

# capture_servicebus/__init__.py
import azure.functions as func
import json
import logging
import os
import urllib.request
import urllib.parse
import urllib.error

logger = logging.getLogger(__name__)


def main(msg: func.ServiceBusMessage) -> None:
    """
    Service Bus-triggered screenshot capture.
    Message body JSON: {"url": "...", "reference_id": "...", "format": "webp"}

    Dead letter: message is dead-lettered after maxDeliveryCount retries
    (configured on the queue, typically 10).
    Raise an exception to trigger retry. Return normally to complete (ack).
    """
    try:
        body = json.loads(msg.get_body().decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        # Bad message — complete (don't retry) to avoid dead letter queue pollution
        logger.error(f"Invalid message body: {e}")
        return

    target_url = body.get("url")
    reference_id = body.get("reference_id", msg.message_id)
    fmt = body.get("format", "webp")

    if not target_url:
        logger.warning(f"No URL in message {reference_id}, completing without capture")
        return

    api_key = _get_api_key()

    try:
        image_bytes = _capture_with_retry(target_url, api_key)
    except Exception as exc:
        logger.error(f"[FAIL] {reference_id}: {exc}")
        # Raise to trigger Service Bus retry (up to maxDeliveryCount)
        raise

    try:
        result = store_screenshot(image_bytes, target_url, fmt)
        logger.info(f"[OK] {reference_id}: {result['bytes']} bytes → {result['blob_name']}")
    except Exception as exc:
        logger.error(f"Storage failed for {reference_id}: {exc}")
        raise

// capture_servicebus/function.json
{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "msg",
      "type": "serviceBusTrigger",
      "direction": "in",
      "queueName": "screenshot-requests",
      "connection": "SERVICE_BUS_CONNECTION"
    }
  ]
}

For Service Bus with managed identity (no connection string), set connection to a named app setting prefix and configure the identity-based connection:

// local.settings.json (dev only — use app settings in production)
{
  "Values": {
    "SERVICE_BUS_CONNECTION__fullyQualifiedNamespace": "your-namespace.servicebus.windows.net"
  }
}

Retry Helper

import time
import urllib.error
import urllib.request
import urllib.parse


def _capture_with_retry(url: str, api_key: str, fmt: str = "webp", max_attempts: int = 3) -> bytes:
    last_error = None
    for attempt in range(max_attempts):
        try:
            params = urllib.parse.urlencode({
                "url": 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},
            )
            with urllib.request.urlopen(req, timeout=25) as resp:
                return resp.read()
        except urllib.error.HTTPError as e:
            if e.code == 429 or e.code >= 500:
                delay = float(e.headers.get("Retry-After", 2 ** attempt))
                time.sleep(delay)
                last_error = e
            else:
                raise
        except urllib.error.URLError as e:
            time.sleep(2 ** attempt)
            last_error = e
    raise RuntimeError(f"All {max_attempts} attempts failed") from last_error

Bicep Deployment

// main.bicep
param location string = resourceGroup().location
param functionAppName string
param storageAccountName string
param keyVaultName string
param serviceBusNamespaceName string
param screenshotApiKey string

// Storage Account for screenshots
resource storageAccount 'Microsoft.Storage/storageAccounts@2023-01-01' = {
  name: storageAccountName
  location: location
  sku: { name: 'Standard_LRS' }
  kind: 'StorageV2'
  properties: {
    minimumTlsVersion: 'TLS1_2'
    supportsHttpsTrafficOnly: true
  }
}

resource screenshotsContainer 'Microsoft.Storage/storageAccounts/blobServices/containers@2023-01-01' = {
  name: '${storageAccount.name}/default/screenshots'
  properties: { publicAccess: 'None' }
}

// Key Vault for API key
resource keyVault 'Microsoft.KeyVault/vaults@2023-07-01' = {
  name: keyVaultName
  location: location
  properties: {
    sku: { family: 'A', name: 'standard' }
    tenantId: subscription().tenantId
    enableRbacAuthorization: true  // Use RBAC instead of access policies
  }
}

resource apiKeySecret 'Microsoft.KeyVault/vaults/secrets@2023-07-01' = {
  parent: keyVault
  name: 'screenshot-api-key'
  properties: { value: screenshotApiKey }
}

// Service Bus for batch queuing
resource serviceBusNamespace 'Microsoft.ServiceBus/namespaces@2022-10-01-preview' = {
  name: serviceBusNamespaceName
  location: location
  sku: { name: 'Standard', tier: 'Standard' }
}

resource screenshotQueue 'Microsoft.ServiceBus/namespaces/queues@2022-10-01-preview' = {
  parent: serviceBusNamespace
  name: 'screenshot-requests'
  properties: {
    maxDeliveryCount: 10
    deadLetteringOnMessageExpiration: true
    defaultMessageTimeToLive: 'PT1H'
  }
}

// App Service Plan (Consumption)
resource appServicePlan 'Microsoft.Web/serverfarms@2023-01-01' = {
  name: '${functionAppName}-plan'
  location: location
  sku: { name: 'Y1', tier: 'Dynamic' }
  kind: 'functionapp'
}

// Function App storage (separate from screenshot storage)
resource functionStorage 'Microsoft.Storage/storageAccounts@2023-01-01' = {
  name: '${functionAppName}store'
  location: location
  sku: { name: 'Standard_LRS' }
  kind: 'StorageV2'
}

// Function App
resource functionApp 'Microsoft.Web/sites@2023-01-01' = {
  name: functionAppName
  location: location
  kind: 'functionapp'
  identity: { type: 'SystemAssigned' }  // Managed identity — no credentials to manage
  properties: {
    serverFarmId: appServicePlan.id
    siteConfig: {
      pythonVersion: '3.12'
      appSettings: [
        { name: 'AzureWebJobsStorage', value: 'DefaultEndpointsProtocol=https;AccountName=${functionStorage.name};AccountKey=${functionStorage.listKeys().keys[0].value}' }
        { name: 'FUNCTIONS_EXTENSION_VERSION', value: '~4' }
        { name: 'FUNCTIONS_WORKER_RUNTIME', value: 'python' }
        { name: 'STORAGE_ACCOUNT_NAME', value: storageAccountName }
        { name: 'STORAGE_CONTAINER_NAME', value: 'screenshots' }
        { name: 'KEY_VAULT_URL', value: keyVault.properties.vaultUri }
        { name: 'SERVICE_BUS_CONNECTION__fullyQualifiedNamespace', value: '${serviceBusNamespaceName}.servicebus.windows.net' }
      ]
    }
  }
}

// RBAC: Function App managed identity → Key Vault Secrets User
resource kvSecretsUserRole 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(keyVault.id, functionApp.id, 'Key Vault Secrets User')
  scope: keyVault
  properties: {
    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', '4633458b-17de-408a-b874-0445c86b69e6')
    principalId: functionApp.identity.principalId
    principalType: 'ServicePrincipal'
  }
}

// RBAC: Function App managed identity → Storage Blob Data Contributor
resource storageBlobContributorRole 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(storageAccount.id, functionApp.id, 'Storage Blob Data Contributor')
  scope: storageAccount
  properties: {
    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', 'ba92f5b4-2d11-453d-a403-e96b0029c9fe')
    principalId: functionApp.identity.principalId
    principalType: 'ServicePrincipal'
  }
}

// RBAC: Function App managed identity → Service Bus Data Receiver
resource sbDataReceiverRole 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(serviceBusNamespace.id, functionApp.id, 'Service Bus Data Receiver')
  scope: serviceBusNamespace
  properties: {
    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', '4f6d3b9b-027b-4f4c-9142-0e5a2a2247e0')
    principalId: functionApp.identity.principalId
    principalType: 'ServicePrincipal'
  }
}

Deploy with:

# Create resource group
az group create --name screenshot-rg --location eastus

# Deploy Bicep template
az deployment group create \
  --resource-group screenshot-rg \
  --template-file main.bicep \
  --parameters \
    functionAppName=screenshot-fn \
    storageAccountName=screenshotstorage \
    keyVaultName=screenshot-kv \
    serviceBusNamespaceName=screenshot-sb \
    screenshotApiKey="your-api-key-here"

# Deploy function code
func azure functionapp publish screenshot-fn

Managed Identity: The Key Advantage

The Bicep above grants three RBAC roles to the function's system-assigned identity: - Key Vault Secrets User — read the API key secret - Storage Blob Data Contributor — read and write screenshots - Service Bus Data Receiver — consume messages from the queue

No connection strings. No API keys in app settings. No credentials to rotate or leak. The managed identity is created automatically when the function app is deployed, and the RBAC assignments are idempotent — re-running the deployment is safe. This is the most operationally clean credential pattern of the three platforms.

Cost Comparison

Same pattern as Lambda and GCF: API cost dominates at moderate volumes. Azure's 1M free executions/month means the function compute cost is zero for most screenshot workloads until volume exceeds ~40k screenshots/month.

Summary

Azure Functions' managed identity pattern eliminates credential management entirely — the function identity is granted RBAC roles directly on Key Vault, Blob Storage, and Service Bus, with no secrets to store, rotate, or accidentally expose. Timer triggers handle scheduled captures with standard cron syntax, Service Bus provides queuing with configurable dead letter handling, and Bicep gives infrastructure-as-code that's idempotent and Azure-native. The Consumption plan's 1M free executions/month makes Azure Functions the lowest-cost option for low-to-moderate screenshot volumes.

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