Building ASGI Microservices with Cloudflare Python Workers in Production

Python microservices have historically been tethered to centralized infrastructure, whether that means containers running behind a load balancer in a single region or Lambda functions cold-starting in one AWS availability zone. Cloudflare Python Workers flip that model by running ASGI frameworks like FastAPI directly at the edge, and this guide walks through building a production-ready aggregation microservice that handles authenticated fan-out, caches results in Workers KV, and deploys globally with optimized cold starts.

Table of Contents

• What ASGI at the Edge Actually Gets You
• How Cloudflare Python Workers Actually Work
• Project Setup and Wrangler Configuration
• Building the ASGI Microservice
• Conquering Pyodide Cold Starts in Production
• Production Hardening
• Deployment and CI/CD
• Performance Benchmarks and Limitations
• Next Steps

What ASGI at the Edge Actually Gets You

Python microservices have historically been tethered to centralized infrastructure, whether that means containers running behind a load balancer in a single region or Lambda functions cold-starting in one AWS availability zone. Cloudflare Python Workers flip that model by running ASGI frameworks like FastAPI directly at the edge, across Cloudflare's global network (see cloudflare.com/network for current data center count). For users near edge nodes, this can cut p50 latency from roughly 200ms to under 50ms compared to single-region deployments, though the serverless edge environment introduces constraints that traditional deployment guides never address.

This guide targets Python developers already comfortable with FastAPI or Starlette and familiar with serverless concepts who want to deploy production ASGI microservices on Cloudflare Workers. By the end, readers will have built a fully functional API aggregation microservice that handles authenticated fan-out to multiple external APIs, caches results in Workers KV, and deploys globally with optimized cold starts.

Cloudflare’s official examples repository includes a runnable FastAPI Worker built on the same ASGI patterns used in this guide. Readers can clone it and follow along: github.com/cloudflare/python-workers-examples/tree/main/fastapi.

How Cloudflare Python Workers Actually Work

The Pyodide Runtime Model

Cloudflare Python Workers run on Pyodide, which is CPython compiled to WebAssembly, executing inside V8 isolates rather than traditional containers or virtual machines. This is fundamentally different from running Python on a server. There is no filesystem access, no subprocess spawning, and no support for native C extensions unless the Pyodide team has pre-compiled and included them in Pyodide's package set. The runtime provides a CPython-compatible environment, but the underlying execution model is a V8 isolate with WebAssembly, not a Linux process.

Packages that rely on compiled C code (such as ujson, native pydantic v2 validators via pydantic-core, or psycopg2) will fail at import time unless Pyodide ships a pre-built version. Verify availability against the Pyodide packages list. Pure-Python packages and those already ported to Pyodide's ecosystem work without issue. Understanding this boundary is essential before selecting any dependency.

There is no filesystem access, no subprocess spawning, and no support for native C extensions unless the Pyodide team has pre-compiled and included them in Pyodide's package set.

ASGI Protocol Bridging

Cloudflare bridges the Workers fetch handler to the ASGI interface natively. When a request arrives at the edge, the Workers runtime invokes the fetch event, and the runtime then translates it into ASGI scope, receive, and send callables that a standard ASGI application can consume. This differs from the AWS Lambda approach, where an adapter like Mangum (for API Gateway v1/v2 events) sits between the Lambda invocation event and the ASGI application, performing explicit translation of API Gateway events. On Cloudflare, the bridging is built into the platform via the workers module.

# src/worker.py — Minimal ASGI entry point for Cloudflare Python Workers
# Requires the cloudflare-workers Python package.
# Verify current API at https://developers.cloudflare.com/workers/languages/python/
# As of authoring, the module is imported as shown; confirm against your installed SDK version.
from workers import ASGIWorker
from app import create_app

app = create_app()

async def on_fetch(request, env, ctx):
    asgi_worker = ASGIWorker(app)
    return await asgi_worker.fetch(request, env, ctx)

This skeleton wires any ASGI-compatible application (FastAPI, Starlette, or bare ASGI) to the Cloudflare Workers fetch event. The env object carries bindings like KV namespaces and secrets. The ctx object provides the waitUntil method for background tasks.

Project Setup and Wrangler Configuration

Prerequisites

You will need Node.js 20 or later for the Wrangler CLI toolchain. Install the Wrangler CLI pinned to a known-good version (e.g., wrangler@3). Python 3.12 is required and must match the Pyodide runtime's supported CPython version. A Cloudflare account with a Workers Paid plan is necessary because the free plan's 10ms CPU limit per invocation makes the patterns in this guide impractical. You will also need a CLOUDFLARE_API_TOKEN with Workers:Edit and KV:Edit permissions, and a real KV namespace ID obtained via wrangler kv namespace create RESPONSE_CACHE.

Scaffolding with Wrangler CLI

Project scaffolding begins with Cloudflare's CLI tool:

npm create cloudflare@latest my-asgi-service -- --template python

This generates a directory structure with src/ containing the worker entry point, a requirements.txt for Python dependencies, and a wrangler.jsonc configuration file. The src/ directory is where all Python source lives. There is no node_modules involved in the runtime; Node.js is used only for the Wrangler toolchain.

Configuring wrangler.jsonc for Python

// wrangler.jsonc — Production configuration for Python ASGI Worker
{
  "name": "asgi-aggregator",
  "main": "src/entry.py",
  "compatibility_date": "2024-12-01",
  "compatibility_flags": ["python_workers"],

  // KV namespace bindings
  "kv_namespaces": [
    {
      "binding": "RESPONSE_CACHE",
      "id": "YOUR_KV_NAMESPACE_ID_HERE" // obtain via: wrangler kv namespace create RESPONSE_CACHE
    }
  ],

  // Environment variables (non-secret)
  "vars": {
    "CACHE_TTL_SECONDS": "300",
    "MAX_SUBREQUEST_TIMEOUT_MS": "5000"
  },

  // Build configuration for dependency installation
  "build": {
    "command": "PYDANTIC_PURE_PYTHON=1 pip install -r requirements.txt --target ./src/lib"
  },

  // Staging environment override
  "env": {
    "staging": {
      "name": "asgi-aggregator-staging",
      "vars": {
        "CACHE_TTL_SECONDS": "60"
      }
    }
  }
}

The compatibility_flags array must include "python_workers" to enable the Pyodide runtime. KV namespace bindings expose the RESPONSE_CACHE binding on the env object inside the worker. Secrets like API keys are never placed here; they are managed separately via wrangler secret put. Each upstream service should have its own secret (e.g., wrangler secret put WEATHER_API_KEY, wrangler secret put NEWS_API_KEY, wrangler secret put STOCK_API_KEY) to follow the principle of least privilege.

Dependency Management and Pruning

Pyodide ships with a substantial set of built-in packages, including json, asyncio, re, and many standard library modules. You can install packages available through Pyodide's ecosystem via micropip-compatible pure-Python wheels. The critical constraint: any package with native C extensions that Pyodide has not pre-compiled will fail.

For HTTP clients, prefer httpx over requests because requests is synchronous only and incompatible with async ASGI handlers; httpx provides native async support. For data validation with FastAPI, pydantic v2's default installation includes compiled Rust validators via pydantic-core, which will not run in Pyodide. Work around this by forcing pure-Python mode.

# requirements.txt — Curated for Pyodide compatibility
fastapi==0.115.0          # ASGI framework; pure-Python core
httpx==0.27.0             # Async HTTP client; no C extensions
pydantic==2.10.0          # Validation; must use pure-Python mode (see below)
starlette==0.40.0         # Pulled by FastAPI; listed for version pinning

# EXCLUDED:
# requests — synchronous only; incompatible with async ASGI handlers. Use httpx.
# uvicorn — not needed; Cloudflare provides the server runtime
# orjson — C extension; use stdlib json
# pydantic-settings — unnecessary in Workers; use env object directly

# Set PYDANTIC_PURE_PYTHON=1 at build time (requires pydantic ≥ 2.10).
# See build.command in wrangler.jsonc.

The PYDANTIC_PURE_PYTHON=1 environment variable is set in the build.command of wrangler.jsonc to prevent Pydantic from attempting to load its compiled core. This falls back to the pure-Python validation path, which is slower but fully compatible with Pyodide.

Building the ASGI Microservice

Application Architecture

The target service implements the Backend-for-Frontend (BFF) pattern: it accepts a single request from a client, fans out concurrently to two or three external APIs, merges their responses into a unified JSON payload, and caches the result in Workers KV. This pattern is representative of real production edge workloads where an API gateway aggregates upstream data to reduce client-side round trips and keep API keys off the client.

Defining Routes and Request Handling

# src/app.py — FastAPI application with routes and models
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional
from cache import get_or_fetch
from subrequests import fan_out_requests
import json
import re

class AggregatedResponse(BaseModel):
    resource_id: str
    weather: Optional[dict] = None
    news: Optional[list] = None
    stock: Optional[dict] = None
    cached: bool = False
    errors: Optional[list] = None

class HealthResponse(BaseModel):
    status: str
    runtime: str

def create_app() -> FastAPI:
    app = FastAPI(title="Edge Aggregator", version="1.0.0")

    @app.get("/health", response_model=HealthResponse)
    async def health_check():
        return HealthResponse(status="ok", runtime="pyodide")

    @app.get("/aggregate/{resource_id}", response_model=AggregatedResponse)
    async def aggregate(resource_id: str, request: Request):
        # Validate resource_id to prevent path traversal and injection
        if not resource_id or not re.match(r'^[a-zA-Z0-9_-]+$', resource_id):
            return JSONResponse(
                status_code=400,
                content={"error": "Invalid resource_id. Only alphanumeric characters, hyphens, and underscores are allowed."},
            )

        env = request.state.env
        cached = await get_or_fetch(
            key=f"agg:{resource_id}",
            env=env,
            # fetcher must return a coroutine; cache.py awaits fetcher()
            fetcher=lambda: fan_out_requests(resource_id, env),
        )
        return JSONResponse(content=cached)

    return app

The env object must be explicitly attached to request.state in the worker entry point; the ASGI bridge does not do this automatically. The entry.py pattern below shows how to perform this injection. Pydantic models define the response contract, and FastAPI handles serialization and OpenAPI documentation generation automatically. The resource_id is validated against a strict allowlist pattern to prevent path traversal and URL injection attacks.

Asynchronous Subrequest Connection Handling

Workers cannot maintain persistent TCP connections across invocations. Each invocation is isolated, so connection pooling in the traditional sense is not viable. The practical pattern is to instantiate a single httpx.AsyncClient per fan-out invocation with short timeouts and no expectation of connection reuse across Worker invocations.

Fan-out uses asyncio.gather() with return_exceptions=True to ensure that a failure in one upstream API does not collapse the entire response. Partial results are returned with error metadata. Each upstream service uses its own dedicated API key secret, configured via wrangler secret put, to follow the principle of least privilege and limit the blast radius if any single key is compromised.

# src/subrequests.py — Concurrent fan-out with graceful partial failure
import asyncio
import httpx
from urllib.parse import quote

async def fan_out_requests(resource_id: str, env) -> dict:
    # Per-service API keys — set via wrangler secret put WEATHER_API_KEY, etc.
    api_keys = {
        "weather": env.WEATHER_API_KEY,
        "news":    env.NEWS_API_KEY,
        "stock":   env.STOCK_API_KEY,
    }
    timeout = float(env.MAX_SUBREQUEST_TIMEOUT_MS) / 1000

    safe_id = quote(resource_id, safe="")

    endpoints = {
        "weather": f"https://api.weather.example.com/v1/{safe_id}",
        "news": f"https://api.news.example.com/v1/search?q={safe_id}",
        "stock": f"https://api.stock.example.com/v1/quote/{safe_id}",
    }

    async with httpx.AsyncClient(timeout=timeout) as client:
        async def fetch_one(name: str, url: str) -> tuple:
            try:
                resp = await client.get(
                    url, headers={"Authorization": f"Bearer {api_keys[name]}"}
                )
                resp.raise_for_status()
                return (name, resp.json(), None)
            except Exception as e:
                # Avoid leaking URLs or secrets in error strings
                return (name, None, type(e).__name__)

        tasks = [fetch_one(name, url) for name, url in endpoints.items()]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    payload = {"resource_id": resource_id, "cached": False, "errors": []}
    for result in results:
        if isinstance(result, Exception):
            payload["errors"].append(type(result).__name__)
        else:
            name, data, error = result
            payload[name] = data
            if error:
                payload["errors"].append(f"{name}: {error}")

    if not payload["errors"]:
        payload["errors"] = None
    return payload

A single httpx.AsyncClient is created per fan-out invocation and shared across all subrequests, avoiding redundant connection setup overhead. The timeout parameter is drawn from the environment variable configured in wrangler.jsonc. Workers enforce subrequest limits per invocation; verify current subrequest limits at developers.cloudflare.com/workers/platform/limits, as limits may differ by plan tier.

KV-Backed State Caching

Workers KV is an eventually consistent, read-optimized key-value store well suited for caching aggregated API responses where staleness of a few seconds is acceptable. The cache-aside pattern checks KV first, performs the fan-out on a miss, and writes the result back with a TTL. An in-process lock prevents thundering herd / cache stampede within a single isolate, ensuring that concurrent cold-miss requests for the same key only trigger one upstream fan-out.

# src/cache.py — Cache-aside pattern with Workers KV and stampede protection
import json
import asyncio

_cache_locks: dict = {}

async def get_or_fetch(key: str, env, fetcher) -> dict:
    kv = env.RESPONSE_CACHE
    cached_value = await kv.get(key)

    if cached_value is not None:
        return {**json.loads(cached_value), "cached": True}

    # One lock per key within this isolate to prevent thundering herd
    if key not in _cache_locks:
        _cache_locks[key] = asyncio.Lock()

    async with _cache_locks[key]:
        # Re-check after acquiring lock (another coroutine may have populated the cache)
        cached_value = await kv.get(key)
        if cached_value is not None:
            return {**json.loads(cached_value), "cached": True}

        result = await fetcher()

        try:
            ttl = int(env.CACHE_TTL_SECONDS)
        except (ValueError, AttributeError):
            ttl = 300  # safe default if env var is missing or non-numeric

        if ttl < 60:
            ttl = 60  # KV minimum TTL

        await kv.put(
            key,
            json.dumps(result),
            expirationTtl=ttl,
        )
        return result

The kv.get() and kv.put() methods are accessed through the binding name defined in wrangler.jsonc. The expirationTtl parameter accepts seconds (minimum 60) and instructs KV to automatically evict the entry after the specified duration; values below 60 will return an API error, so the code enforces that minimum. Deserialization uses the standard library json module since orjson is unavailable in Pyodide.

Conquering Pyodide Cold Starts in Production

Understanding Cold-Start Anatomy

A cold start passes through four phases: V8 isolate creation (minimal, typically under 10ms), Pyodide runtime initialization (significant, as it loads the WebAssembly CPython binary), package loading (variable, depending on total dependency size), and application initialization (controllable through code structure). Unoptimized deployments with large dependency trees can see cold starts reaching 2 to 5 seconds. With the techniques in this guide, cold starts drop to roughly 650ms at the median (see the benchmark table below for methodology and conditions).

Unoptimized deployments with large dependency trees can see cold starts reaching 2 to 5 seconds. With the techniques in this guide, cold starts drop to roughly 650ms at the median.

Warm-Up Routines and Initialization Caching

Four techniques reduce cold-start impact in production:

1. Place all imports and heavy initialization at module-level scope. Pyodide caches module state within a V8 isolate's lifetime, so subsequent invocations reusing the same isolate skip re-initialization entirely.
2. Lazy-load non-critical dependencies inside the route handlers that actually need them, rather than importing at the top of the module.
3. Use Cloudflare's snapshot mechanism to pre-initialize the Pyodide environment during the build step, baking the initialized runtime state into the deployed artifact. Snapshot support for Python Workers is in preview as of early 2025; check the Cloudflare Workers changelog for availability and configuration syntax before relying on this.
4. Configure Cloudflare Cron Triggers to issue scheduled warm-up pings, keeping isolates alive in high-traffic regions.

# src/entry.py — Module-level initialization with warm-up pattern
from workers import ASGIWorker
from starlette.datastructures import State
import httpx

def _create_app():
    from app import create_app
    # Pre-import heavy modules so they're cached in the isolate
    import fastapi
    import pydantic
    return create_app()

# Execute at import time — runs during isolate initialization
_app = _create_app()

async def on_fetch(request, env, ctx):
    if not hasattr(request, "state") or not isinstance(request.state, State):
        request.state = State()
    request.state.env = env
    worker = ASGIWorker(_app)
    return await worker.fetch(request, env, ctx)

# Cron trigger handler — keeps isolates warm
async def on_scheduled(event, env, ctx):
    # Self-ping to prevent isolate eviction
    # Replace with your actual Worker subdomain from: wrangler whoami
    async with httpx.AsyncClient(timeout=5.0) as client:
        await client.get("https://asgi-aggregator.your-domain.workers.dev/health")

The _create_app() function runs at import time, meaning Pyodide executes it during isolate creation. This front-loads the cost into the cold start rather than the first request. The cron handler issues a lightweight health check (with an explicit timeout to prevent indefinite hangs) to prevent idle isolate eviction. Note that on_fetch attaches the env object to request.state using Starlette's State class; this is required because the ASGI bridge does not do this automatically. Using the proper State class ensures compatibility with any middleware or dependencies that read request.state.

Dependency Size Budget

Total dependency payload should stay under 5 to 10 MB to keep cold-start p50 under 1 second (see the benchmark table below). Every additional megabyte adds measurable latency to the package-loading phase. To audit package sizes, use pip show locally and check the .dist-info directory size, or use pip download --no-deps -d /tmp/pkgs && du -sh /tmp/pkgs/*. For Pyodide-specific availability, check pyodide.org/en/stable/usage/packages-in-pyodide.html.

A practical decision framework: if a dependency adds more than 1 MB and is used in fewer than 20% of requests, lazy-load it inside the specific handler that needs it rather than importing at module level.

Production Hardening

Error Handling and Observability

Structured logging in Workers relies on print() statements, which the Workers runtime captures and forwards to configured log destinations. Cloudflare Logpush can stream these logs to third-party services. Workers Analytics Engine provides built-in request metrics without additional instrumentation.

Every error response should include a correlation ID, generated per request, to enable tracing across distributed systems. Return proper HTTP status codes: 502 for upstream failures, 504 for subrequest timeouts, 422 for validation errors.

Secrets and Environment Variables

Set API keys and tokens via wrangler secret put SECRET_NAME, which encrypts and stores them in Cloudflare's infrastructure. Inside the worker, access secrets via env.SECRET_NAME. Never place secrets in wrangler.jsonc or commit them to version control. Use environment-specific overrides for staging versus production credentials. Each upstream service should have its own secret (e.g., WEATHER_API_KEY, NEWS_API_KEY, STOCK_API_KEY) to limit the blast radius if any single key is compromised.

Rate Limiting and Request Validation

Apply rate limiting at the Cloudflare zone level using built-in rate limiting rules, which operate before the Worker is invoked and consume no Worker CPU time. Application-level input validation is handled by Pydantic models in FastAPI routes and explicit resource_id validation in the route handler, providing defense-in-depth against malformed requests and injection attacks.

Testing and Local Development

wrangler dev starts a local development server running the actual Pyodide runtime, providing high-fidelity iteration. For automated testing, test the FastAPI app with httpx.ASGITransport and pytest, mocking the env object and KV bindings.

Because the env object is injected per-request via request.state (not at the application level), tests must replicate this injection using middleware. The following pattern adds a test-specific middleware that attaches the mock env to each request. The fan_out_requests function is also patched to prevent real HTTP calls from being made during tests:

# tests/test_aggregate.py — Pytest with mocked KV bindings and patched subrequests
import pytest
import httpx
from unittest.mock import AsyncMock, MagicMock, patch
from app import create_app

@pytest.fixture
def mock_env():
    env = MagicMock()
    mock_kv = AsyncMock()
    mock_kv.get = AsyncMock(return_value=None)
    mock_kv.put = AsyncMock()
    env.RESPONSE_CACHE = mock_kv
    env.WEATHER_API_KEY = "test-weather-key"
    env.NEWS_API_KEY = "test-news-key"
    env.STOCK_API_KEY = "test-stock-key"
    env.CACHE_TTL_SECONDS = "60"
    env.MAX_SUBREQUEST_TIMEOUT_MS = "3000"
    return env

@pytest.mark.asyncio
async def test_aggregate_cache_miss(mock_env):
    app = create_app()

    @app.middleware("http")
    async def inject_test_env(request, call_next):
        request.state.env = mock_env
        return await call_next(request)

    fake_payload = {
        "resource_id": "AAPL",
        "weather": {"temp": 72},
        "news": [],
        "stock": {"price": 189.0},
        "cached": False,
        "errors": None,
    }

    with patch("subrequests.fan_out_requests", new=AsyncMock(return_value=fake_payload)):
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.get("/aggregate/AAPL")

    assert response.status_code == 200
    data = response.json()
    assert data["resource_id"] == "AAPL"
    # Verify that the mock KV was actually called, confirming env injection works
    assert mock_env.RESPONSE_CACHE.get.called, "KV.get was never called — env injection failed"
    assert mock_env.RESPONSE_CACHE.put.called, "KV.put was never called — cache write failed"

@pytest.mark.asyncio
async def test_aggregate_cache_hit(mock_env):
    import json
    cached = {"resource_id": "AAPL", "cached": False, "errors": None}
    mock_env.RESPONSE_CACHE.get = AsyncMock(return_value=json.dumps(cached))

    app = create_app()

    @app.middleware("http")
    async def inject_test_env(request, call_next):
        request.state.env = mock_env
        return await call_next(request)

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/aggregate/AAPL")

    assert response.status_code == 200
    assert response.json()["cached"] is True
    mock_env.RESPONSE_CACHE.put.assert_not_called()

@pytest.mark.asyncio
async def test_invalid_resource_id_rejected(mock_env):
    app = create_app()

    @app.middleware("http")
    async def inject_test_env(request, call_next):
        request.state.env = mock_env
        return await call_next(request)

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/aggregate/../../etc/passwd")

    assert response.status_code in (400, 422)

This test verifies the aggregation endpoint with a fully mocked environment, avoiding any network calls or KV dependencies. The ASGITransport from httpx sends requests directly to the ASGI app in-process. The middleware-based injection mirrors how entry.py attaches env to request.state at runtime. The test suite covers cache miss, cache hit, and invalid input scenarios.

Deployment and CI/CD

Deploying with Wrangler

Deploy using a standard promotion workflow. wrangler deploy --env staging pushes to the staging environment defined in wrangler.jsonc. After smoke testing, wrangler deploy (without the --env flag) promotes to production. Rollbacks may be available via wrangler rollback; verify availability for Python Workers in the Cloudflare changelog. Alternatively, redeploy a previous git tag.

GitHub Actions Pipeline

Create requirements-dev.txt with at minimum the test dependencies:

# requirements-dev.txt
pytest==8.3.0
pytest-asyncio==0.24.0
httpx==0.27.0

# .github/workflows/deploy.yml
name: Deploy ASGI Worker
on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-node@v4
        with:
          node-version: "20"

      - name: Install Wrangler
        run: npm install -g wrangler@3  # Pin major version; check Cloudflare changelog before upgrading

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Validate no placeholder KV IDs
        run: |
          if grep -r "YOUR_KV" wrangler.jsonc; then
            echo "FAIL: placeholder KV namespace ID not replaced"
            exit 1
          fi

      - name: Run tests
        run: |
          pip install -r requirements-dev.txt
          pytest tests/ -v -m "not integration" --timeout=10

      - name: Deploy to staging
        run: wrangler deploy --env staging
        env:
          CLOUDFLARE_API_TOKEN: ${{ secrets.CF_API_TOKEN }}

      - name: Smoke test staging
        run: |
          # Replace with your actual staging subdomain
          HEALTH_RESPONSE=$(curl -sf https://asgi-aggregator-staging.your-domain.workers.dev/health)
          echo "$HEALTH_RESPONSE" | python -c "import sys,json; d=json.load(sys.stdin); sys.exit(0 if d.get('status')=='ok' else 1)"

      - name: Deploy to production
        run: wrangler deploy
        env:
          CLOUDFLARE_API_TOKEN: ${{ secrets.CF_API_TOKEN }}

The pipeline validates configuration, runs tests in CI (excluding integration tests that require a running worker), deploys to staging, runs a smoke test against the health endpoint (validating the JSON response body), and only then promotes to production. The CLOUDFLARE_API_TOKEN is stored as a GitHub Actions secret. If the smoke test fails, including cases where the endpoint returns an error payload with a 200 status, the pipeline stops before deploying to production.

Performance Benchmarks and Limitations

The following table reflects indicative results from applying the optimization techniques in this guide, comparing a naive deployment against the fully optimized template. These numbers were collected using hey with 50 concurrent connections and 1,000 total requests against a single Cloudflare region (US-East). They are approximate and will vary by region, dependency set, traffic pattern, and Wrangler version. Reproduce against your own deployed Worker using the same hey parameters above.

Several hard constraints apply. Pyodide running in V8 isolates is not suited for CPU-heavy workloads like ML inference or image processing; Workers enforce a 128 MB memory limit, up to 30 seconds of CPU time per invocation on the paid plan, and only 10ms CPU time per invocation on the free plan. Verify current limits at developers.cloudflare.com/workers/platform/limits. Packages with native C extensions remain unsupported unless Pyodide includes a pre-compiled version.

For CPU-intensive Python workloads, AWS Lambda with Mangum or containerized deployments remain better options. For teams already invested in JavaScript or TypeScript, native Workers provide faster cold starts and a larger ecosystem. This Python Workers approach is strongest for I/O-bound microservices where the team's expertise and existing codebase are Python.

This Python Workers approach is strongest for I/O-bound microservices where the team's expertise and existing codebase are Python.

Next Steps

Cloudflare Python Workers with ASGI frameworks let you deploy global Python microservices with sub-50ms warm-request latency (see the benchmark table above), no containers, no orchestration platforms, and no region selection. The three techniques that matter most for production readiness are aggressive dependency pruning to stay under the size budget, module-level initialization with warm-up routines to minimize cold-start impact, and KV-backed cache-aside patterns to reduce upstream API load.

Start from Cloudflare’s official FastAPI Worker example at github.com/cloudflare/python-workers-examples/tree/main/fastapi, replace the example API endpoints with real upstream services, configure secrets, and deploy. Looking ahead, Durable Objects may eventually support Python Workers for stateful edge applications (no public roadmap has been confirmed as of this writing), and Workers AI integration enables lightweight inference at the edge without the CPU constraints of general-purpose Pyodide execution.