Screenshot API with FastAPI: Async Endpoints, Background Tasks, and httpx

FastAPI is the async-first Python web framework — built on ASGI, using async/await natively, and designed for high-throughput API services. This guide covers screenshot API integration patterns that fit FastAPI's model: async httpx clients, background tasks, Pydantic validation, and dependency injection.

Prerequisites

A free API key from the screenshot API. Set SCREENSHOT_API_KEY in your environment.

# .env
SCREENSHOT_API_KEY=your_key_here
SCREENSHOT_API_BASE=https://hermesforge.dev

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    screenshot_api_key: str
    screenshot_api_base: str = 'https://hermesforge.dev'

    class Config:
        env_file = '.env'

settings = Settings()

Basic Async Endpoint

FastAPI with httpx — fully async, no thread-pool overhead:

# main.py
import httpx
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import Response
from config import settings

app = FastAPI()

@app.get('/api/screenshot')
async def screenshot(
    url: str = Query(..., description='Target URL to screenshot'),
    format: str = Query('webp', pattern='^(png|webp|jpeg)$'),
    width: int = Query(1280, ge=320, le=3840),
    height: int = Query(800, ge=240, le=2160),
):
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(
            f'{settings.screenshot_api_base}/api/screenshot',
            params={'url': url, 'format': format, 'width': width, 'height': height},
            headers={'X-API-Key': settings.screenshot_api_key},
        )

    if response.status_code != 200:
        raise HTTPException(
            status_code=response.status_code,
            detail=f'Screenshot API error: {response.status_code}',
        )

    return Response(
        content=response.content,
        media_type=response.headers.get('content-type', 'image/webp'),
        headers={'Cache-Control': 'public, max-age=3600'},
    )

Shared httpx Client (Recommended for Production)

Creating a new AsyncClient per request is wasteful. Use a shared client with connection pooling:

# main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

http_client: httpx.AsyncClient = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    http_client = httpx.AsyncClient(
        timeout=httpx.Timeout(30.0),
        limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
    )
    yield
    await http_client.aclose()

app = FastAPI(lifespan=lifespan)

Inject it as a dependency:

from fastapi import Depends

def get_http_client() -> httpx.AsyncClient:
    return http_client

@app.get('/api/screenshot')
async def screenshot(
    url: str = Query(...),
    client: httpx.AsyncClient = Depends(get_http_client),
):
    response = await client.get(
        f'{settings.screenshot_api_base}/api/screenshot',
        params={'url': url, 'format': 'webp'},
        headers={'X-API-Key': settings.screenshot_api_key},
    )
    if not response.is_success:
        raise HTTPException(status_code=502, detail='Screenshot failed')

    return Response(content=response.content, media_type='image/webp')

Pydantic Request Model (POST endpoint)

For richer request validation, use a Pydantic model with POST:

from pydantic import BaseModel, HttpUrl, Field

class ScreenshotRequest(BaseModel):
    url: HttpUrl
    format: str = Field('webp', pattern='^(png|webp|jpeg)$')
    width: int = Field(1280, ge=320, le=3840)
    height: int = Field(800, ge=240, le=2160)
    full_page: bool = False
    delay: int = Field(0, ge=0, le=10000)

class ScreenshotResponse(BaseModel):
    success: bool
    content_type: str
    size_bytes: int

@app.post('/api/screenshots', response_model=ScreenshotResponse)
async def create_screenshot(
    req: ScreenshotRequest,
    client: httpx.AsyncClient = Depends(get_http_client),
):
    params = {
        'url': str(req.url),
        'format': req.format,
        'width': str(req.width),
        'height': str(req.height),
    }
    if req.full_page:
        params['full_page'] = 'true'
    if req.delay:
        params['delay'] = str(req.delay)

    response = await client.get(
        f'{settings.screenshot_api_base}/api/screenshot',
        params=params,
        headers={'X-API-Key': settings.screenshot_api_key},
    )
    if not response.is_success:
        raise HTTPException(status_code=502, detail='Screenshot service error')

    return ScreenshotResponse(
        success=True,
        content_type=response.headers.get('content-type', 'image/webp'),
        size_bytes=len(response.content),
    )

Background Tasks for Async Screenshot Generation

FastAPI's BackgroundTasks is ideal for fire-and-forget screenshot jobs:

import asyncio
from fastapi import BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession

async def generate_screenshot_background(
    page_id: int,
    url: str,
    db: AsyncSession,
):
    """Run after the response is sent — doesn't block the HTTP response."""
    async with httpx.AsyncClient(timeout=30) as client:
        try:
            response = await client.get(
                f'{settings.screenshot_api_base}/api/screenshot',
                params={'url': url, 'format': 'png', 'width': '1200', 'height': '630'},
                headers={'X-API-Key': settings.screenshot_api_key},
            )
            response.raise_for_status()

            await db.execute(
                'UPDATE pages SET screenshot_data = :data, status = :status WHERE id = :id',
                {'data': response.content, 'status': 'ready', 'id': page_id},
            )
            await db.commit()
        except Exception:
            await db.execute(
                'UPDATE pages SET status = :status WHERE id = :id',
                {'status': 'failed', 'id': page_id},
            )
            await db.commit()

@app.post('/pages')
async def create_page(
    url: str,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        'INSERT INTO pages (url, status) VALUES (:url, :status) RETURNING id',
        {'url': url, 'status': 'pending'},
    )
    page_id = result.scalar()
    await db.commit()

    # Queue screenshot generation — response returns immediately
    background_tasks.add_task(generate_screenshot_background, page_id, url, db)

    return {'id': page_id, 'status': 'pending'}

Redis Caching with aioredis

Cache screenshot responses to avoid redundant API calls under load:

import hashlib
import json
import redis.asyncio as aioredis
from fastapi import Depends

redis_client: aioredis.Redis = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client, redis_client
    http_client = httpx.AsyncClient(timeout=30)
    redis_client = await aioredis.from_url('redis://localhost:6379', decode_responses=False)
    yield
    await http_client.aclose()
    await redis_client.aclose()

def get_redis() -> aioredis.Redis:
    return redis_client

CACHE_TTL = 3600

@app.get('/api/screenshot/cached')
async def screenshot_cached(
    url: str = Query(...),
    format: str = Query('webp'),
    width: int = Query(1280),
    height: int = Query(800),
    redis: aioredis.Redis = Depends(get_redis),
    client: httpx.AsyncClient = Depends(get_http_client),
):
    cache_key = 'screenshot:' + hashlib.sha256(
        f'{url}:{format}:{width}:{height}'.encode()
    ).hexdigest()[:20]

    cached = await redis.get(cache_key)
    if cached:
        meta_key = cache_key + ':meta'
        content_type = (await redis.get(meta_key) or b'image/webp').decode()
        return Response(
            content=cached,
            media_type=content_type,
            headers={'X-Cache': 'HIT'},
        )

    response = await client.get(
        f'{settings.screenshot_api_base}/api/screenshot',
        params={'url': url, 'format': format, 'width': width, 'height': height},
        headers={'X-API-Key': settings.screenshot_api_key},
    )
    if not response.is_success:
        raise HTTPException(status_code=502, detail='Screenshot failed')

    content_type = response.headers.get('content-type', 'image/webp')
    await redis.setex(cache_key, CACHE_TTL, response.content)
    await redis.setex(cache_key + ':meta', CACHE_TTL, content_type.encode())

    return Response(content=response.content, media_type=content_type)

Streaming Response

For large screenshots, stream the response instead of buffering in memory:

from fastapi.responses import StreamingResponse

@app.get('/api/screenshot/stream')
async def screenshot_stream(
    url: str = Query(...),
    client: httpx.AsyncClient = Depends(get_http_client),
):
    async def generate():
        async with client.stream(
            'GET',
            f'{settings.screenshot_api_base}/api/screenshot',
            params={'url': url, 'format': 'png'},
            headers={'X-API-Key': settings.screenshot_api_key},
        ) as response:
            if not response.is_success:
                raise HTTPException(status_code=502, detail='Screenshot failed')
            async for chunk in response.aiter_bytes(chunk_size=8192):
                yield chunk

    return StreamingResponse(generate(), media_type='image/png')

Rate Limiting with slowapi

Protect your endpoint from abuse using slowapi (Starlette-compatible):

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get('/api/screenshot')
@limiter.limit('10/minute')
async def screenshot(request: Request, url: str = Query(...)):
    # request parameter required for slowapi to extract IP
    ...

Running with uvicorn

# Development
uvicorn main:app --reload --port 8000

# Production (with multiple workers)
uvicorn main:app --workers 4 --port 8000 --host 0.0.0.0

Or with gunicorn for process management:

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Dependencies

fastapi>=0.110.0
uvicorn[standard]>=0.29.0
httpx>=0.27.0
pydantic-settings>=2.2.0
redis>=5.0.0        # for aioredis caching
slowapi>=0.1.9      # for rate limiting

Full API reference: /api. Get a free key (50/day): /api/keys.