---
title: Process background jobs in Python with Vercel Queues
description: Move slow and failure-prone work out of your Python request handlers, with typed topics, consumer groups, and built-in retries.
url: "https://vercel.com/kb/guide/python-and-vercel-queues"
published: 2026-10-26
last_updated: 2026-10-26
authors: Ben Sabic
related_resources:
  - title: "Backends on Vercel: Python, containers, and Daemons"
    url: "https://vercel.com/blog/backends-on-vercel-python-containers-and-daemons"
    description: "Explore how Python, container images, and background workers fit into the Vercel platform."
install_vercel_plugin: npx plugins add vercel/vercel-plugin
---

When a customer clicks **Place order**, a checkout endpoint may save the order, email a receipt, and call the warehouse before returning a response. Slow email delivery keeps the customer waiting; a warehouse outage can make checkout appear to fail even after the order is saved.

Move those follow-up tasks into [Vercel Queues](https://vercel.com/docs/queues) so the endpoint waits for the queue to accept the order message. Python functions decorated with `@subscribe` process the message independently and retry failed deliveries within the configured limits.

Vercel manages the queue and runs the subscribers. Your application still needs to handle duplicate deliveries and track work that cannot be completed.

## Overview

In this guide, you'll learn how to:

- Explain how Vercel Queues delivers messages, and what at-least-once delivery means for your code
  
- Declare a typed topic and publish messages to it from Python
  
- Consume messages with `@subscribe`, and fan one message out to several independent consumers
  
- Control retries, delays, and duplicate sends
  
- Test the whole path locally, then deploy it with a FastAPI app on Vercel
  
- Choose between Queues, Workflows, and Cron Jobs for a given job
  

> The Python Queues SDK is in beta. APIs may change before general availability.

## What is Vercel Queues?

[Vercel Queues](https://vercel.com/docs/queues) stores messages in topics for independent consumer groups to process. Topics retain messages until they expire, even after a group acknowledges them. Each group tracks its own progress.

The Python SDK provides asynchronous and synchronous clients, Pydantic-backed topics, and an embedded queue for local tests. Import the async API from `vercel.queue` or the sync API from `vercel.queue.sync`.

The checkout route waits for `send()` to accept the message. Consumer groups then process it independently, with failed deliveries retried until the applicable attempt limit or message expiry.

### The pieces you'll use

| Piece          | What it is                                                     | In the SDK                  |
| -------------- | -------------------------------------------------------------- | --------------------------- |
| Topic          | A named stream of messages, with a payload type                | `Topic[Order]("orders")`    |
| Message        | One unit of work: a payload plus delivery metadata             | `send(topic, payload)`      |
| Subscriber     | A function that handles each message on a topic                | `@subscribe(topic=orders)`  |
| Consumer group | A named set of subscribers that share one copy of each message | `consumer_group="receipts"` |

### How a message moves

On Vercel, subscribers run in push mode. List their modules in `pyproject.toml` so Vercel can build private functions that receive queue deliveries. One generated function can handle several subscribers from the same module.

The SDK renews the message lease while the handler runs. Returning successfully acknowledges the delivery for that consumer group. If the handler raises, the message remains eligible for another attempt within its retention period and configured delivery limit.

Each consumer group gets its own copy of every message. Two subscribers in different groups both see every order. Two subscribers in the _same_ group share the work, and each message goes to only one of them.

### At least once, so make handlers safe to repeat

Queues uses at-least-once delivery, so a handler can receive the same message again after a timeout or crash. Messages must become available before their retention period expires to be delivered.

Use stable idempotency keys enforced by the email provider for receipts. For database writes, use an upsert or unique constraint.

Checking whether work already happened is only sufficient when the check and effect are protected against concurrent attempts and failures.

## Prerequisites

Before you begin, make sure you have:

- A [Vercel account](https://vercel.com/signup).
  
- Python 3.12 or a newer version supported by the [Vercel Python runtime](https://vercel.com/docs/functions/runtimes/python).
  
- Working knowledge of `async`/`await`, [Pydantic](https://docs.pydantic.dev/), and [FastAPI](https://fastapi.tiangolo.com/).
  
- An existing application with database, email, and warehouse integrations.
  

For local development, you also need [uv](https://docs.astral.sh/uv/), Node.js 24+, and the [Vercel CLI](https://vercel.com/docs/cli).

## Step 1: Set up the project

Create a project, add FastAPI and the Vercel SDK, and link it:

```bash
uv init --python 3.12 checkout
cd checkout
uv add fastapi vercel
vercel link
```

_The_ `vercel` _package includes the Queues SDK and Pydantic. For a smaller install, use_ `vercel-queue[typed]` _instead._

This example adds Queues to an existing checkout. Create `app_clients.py` to wrap your application's integrations before running the code. These adapters are application code, separate from the Vercel SDK:

| Adapter                                                                   | Required behavior                                                                   |
| ------------------------------------------------------------------------- | ----------------------------------------------------------------------------------- |
| `db.orders.save(order)`                                                   | Save the trusted order idempotently, keyed by `order_id`                            |
| `email.send(to=..., template=..., data=..., idempotency_key=...)`         | Send through a provider that enforces the supplied idempotency key                  |
| `warehouse.create_shipment(order_id=..., items=..., idempotency_key=...)` | Enforce duplicate protection and raise `WarehouseUnavailable` for retryable outages |
| `failed_jobs.record(key=..., order=..., error=...)`                       | Persist a failed order for operator review, deduplicated by `key`                   |

All four methods are asynchronous. Keep provider credentials in server environment variables, and configure an alert or review process for records written by `failed_jobs`.

Create the following files, including an empty `queues/__init__.py`:

```plaintext
checkout/
├── pyproject.toml
├── uv.lock
├── app_clients.py    # adapters for your existing integrations
├── main.py           # FastAPI app: publishes messages (Step 3)
└── queues/
    ├── __init__.py
    └── orders.py     # topic and subscribers (Step 2)
```

## Step 2: Declare the topic and its subscribers

Define the shared order model, topic, and subscribers in `queues/orders.py`:

```python
from pydantic import BaseModel
from vercel.queue import Topic, subscribe

from app_clients import email, warehouse


class Order(BaseModel):
    order_id: str
    email: str
    items: list[str]
    total_cents: int


orders = Topic[Order]("orders")


@subscribe(topic=orders, consumer_group="receipts")
async def send_receipt(order: Order) -> None:
    await email.send(
        to=order.email,
        template="receipt",
        data=order.model_dump(),
        idempotency_key=f"receipt-{order.order_id}",
    )


@subscribe(topic=orders, consumer_group="warehouse")
async def notify_warehouse(order: Order) -> None:
    await warehouse.create_shipment(
        order_id=order.order_id,
        items=order.items,
        idempotency_key=f"shipment-{order.order_id}",
    )
```

Both consumer groups receive each order. The SDK validates the payload as an `Order` before calling the handlers, and each adapter receives a stable idempotency key for its external operation.

These keys work only if the adapters enforce duplicate protection for the retry window you need. The Queues SDK does not deduplicate email or shipping calls. Explicit consumer group names also make these handlers easier to identify in the dashboard; omitted names are derived from the function's import path.

Now tell Vercel where the subscribers live. Add this to `pyproject.toml`:

```toml
[[tool.vercel.subscribers]]
entrypoint = "queues.orders"
```

_The_ `entrypoint` _is a module import path, not a file path:_ `queues.orders`_, not_ `queues/orders.py`_._

At build time, Vercel imports that module, finds every `@subscribe` in it, and turns them into a private function that the queue calls. You don't touch `vercel.json`.

## Step 3: Publish from your checkout route

Add the route below to `main.py`. It shows the queue handoff after saving an order. In a production checkout, authenticate the caller and construct the `Order` from records the caller is authorized to use. Derive prices, fulfillment items, and the recipient on the server.

```python
from fastapi import FastAPI
from vercel.queue import send

from app_clients import db
from queues.orders import Order, orders

app = FastAPI()


@app.post("/api/orders")
async def place_order(order: Order):
    await db.orders.save(order)
    await send(orders, order, idempotency_key=f"order-{order.order_id}")
    return {"status": "accepted", "order_id": order.order_id}
```

The response waits for the database save and queue publish, but no longer waits for email or the warehouse. Importing the shared `Order` and topic keeps the sender and subscribers on the same message schema.

Saving and publishing are separate operations. If `send()` fails after the database save, retry with the same order and idempotency key.

To recover without client retries, save the order and an outbox record in one database transaction. Publish pending records from a separate process.

The `idempotency_key` deduplicates repeated publishes, including retries after a double-click or network timeout. Reusing the key during its window is accepted without delivering another copy. The window lasts for the shorter of 24 hours and the message's retention period.

Use the same order ID when retrying, and make the database save idempotent too. This key does not cover the database write or the subscribers' external side effects.

## Step 4: Handle failures and retries

Failed deliveries retry within the message's retention period, which defaults to 24 hours. The default retry delay is 60 seconds. For a warehouse outage, increase the delay between attempts and record unresolved orders for an operator to review.

Replace `notify_warehouse` in `queues/orders.py` with this version, keeping the `Order`, `orders`, and receipt handler from Step 2. The full `Message[Order]` gives the handler access to its delivery count:

```python
from vercel.queue import Message, RetryAfter, subscribe

from app_clients import WarehouseUnavailable, failed_jobs, warehouse

MAX_SHIPMENT_ATTEMPTS = 8


@subscribe(topic=orders, consumer_group="warehouse")
async def notify_warehouse(message: Message[Order]) -> None:
    order = message.payload
    attempt = message.metadata.delivery_count

    try:
        await warehouse.create_shipment(
            order_id=order.order_id,
            items=order.items,
            idempotency_key=f"shipment-{order.order_id}",
        )
    except WarehouseUnavailable as exc:
        if attempt >= MAX_SHIPMENT_ATTEMPTS:
            await failed_jobs.record(
                key=f"warehouse-{order.order_id}",
                order=order.model_dump(),
                error=str(exc),
            )
            return
        raise RetryAfter(min(600, 30 * 2 ** (attempt - 1))) from exc
```

Requested delays double from 30 seconds to a maximum of 10 minutes. `RetryAfter` sets eligibility for redelivery; dispatch can happen later. Other exceptions use the subscriber's base `retry_after` setting.

After eight or more delivery attempts, a retryable warehouse error is recorded for operator review. The handler acknowledges only after that write succeeds. If the write fails, delivery can retry until the message expires. The stable failure-record key prevents duplicate review items.

Queues has no built-in dead-letter queue. `max_attempts` is a separate cap on push deliveries. It can stop a trigger from delivering a message even if your final error handler never completed, so this example leaves it unset. Monitor retry depth and the failed-job store independently; neither a delivery cap nor a notification call guarantees a successful handoff.

Handlers accept one argument, either a typed payload or a `Message` containing that payload and delivery metadata. Use the message form when you need its attempt count or message ID.

## Step 5: Run it locally

Start the app with the Vercel CLI:

```bash
vercel dev
```

With the application adapters in place, `vercel dev` runs FastAPI and an in-memory queue that dispatches to your subscribers. Use it to check the local integration; restarting it clears local queue state.

Place an order from a second terminal:

```bash
curl -X POST http://localhost:3000/api/orders \
  -H "Content-Type: application/json" \
  -d '{"order_id": "ord_1", "email": "ada@example.com", "items": ["mug"], "total_cents": 1800}'
```

The endpoint returns an accepted status after saving the order and queuing its message. Use local logs or test adapters to confirm both subscribers ran.

To exercise retries, configure the warehouse adapter to raise `WarehouseUnavailable` in your local test environment and send a new order ID. The receipt handler can still succeed while the warehouse handler retries with increasing delays. Use test credentials or fake adapters for this exercise.

Run the app with `vercel dev` rather than plain `uvicorn`. Outside `vercel dev`, `send()` has no local queue to talk to, and it needs a region and a deployment ID before it will reach the real service.

## Step 6: Test your subscribers

The decorator returns the original function, so tests can call a handler directly. Install pytest for the test runner; AnyIO is included through the Queues SDK:

```bash
uv add --dev pytest
```

Create `tests/conftest.py` with these fakes. They replace the adapters imported by `queues.orders` and use the asyncio backend:

```python
import asyncio
from types import SimpleNamespace

import pytest

import queues.orders as handlers


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
def fake_email(monkeypatch):
    sent = []
    lock = asyncio.Lock()

    async def send(*, to, template, data, idempotency_key):
        async with lock:
            if idempotency_key not in sent:
                sent.append(idempotency_key)

    fake = SimpleNamespace(send=send, sent=sent)
    monkeypatch.setattr(handlers, "email", fake)
    return fake


@pytest.fixture
def fake_warehouse(monkeypatch):
    shipments = []
    keys = set()

    async def create_shipment(*, order_id, items, idempotency_key):
        if idempotency_key not in keys:
            keys.add(idempotency_key)
            shipments.append(order_id)

    fake = SimpleNamespace(create_shipment=create_shipment, shipments=shipments)
    monkeypatch.setattr(handlers, "warehouse", fake)
    return fake
```

Create `tests/test_orders.py` with a repeated-delivery test:

```python
import asyncio

import pytest

from queues.orders import Order, send_receipt

ORDER = Order(order_id="ord_1", email="ada@example.com", items=["mug"], total_cents=1800)


@pytest.mark.anyio
async def test_receipt_uses_a_stable_key(fake_email):
    await send_receipt(ORDER)
    await asyncio.gather(send_receipt(ORDER), send_receipt(ORDER))
    assert fake_email.sent == ["receipt-ord_1"]
```

This test checks key reuse with an idempotent fake. Verify the real email adapter's timeout retries and the provider's key retention period.

Use the embedded queue service to test publishing and dispatch together. It delivers messages to registered subscribers within the test process:

```python
import anyio
from vercel.queue.embedded import embedded_queue_service

from queues.orders import orders  # importing the module registers its subscribers


@pytest.mark.anyio
async def test_order_reaches_both_consumers(fake_email, fake_warehouse):
    async with embedded_queue_service() as service:
        await service.get_async_client().send(orders, ORDER)
        with anyio.fail_after(5):
            while not (fake_email.sent and fake_warehouse.shipments):
                await anyio.sleep(0.01)

    assert fake_email.sent == ["receipt-ord_1"]
    assert fake_warehouse.shipments == ["ord_1"]
```

Append this test to `tests/test_orders.py` and run `uv run pytest`. It reuses `ORDER` and the fixtures above. Publishing through `service.get_async_client()` routes the message to the embedded service.

Wait for your handlers' side effects, not for the queue to report the message as handled. With two consumer groups on one topic, the embedded queue marks a message acknowledged as soon as _either_ group finishes, so a test that waits on that can pass before the second handler has even run.

For tests that only send or poll messages, the SDK also provides an `embedded_queue_server` fixture. Enable it by adding `pytest_plugins = ["vercel.queue.testing.pytest"]` to `tests/conftest.py`. This fixture exposes the queue API without dispatching subscribers.

## Step 7: Deploy

Add `.vercel` to `.gitignore` if it isn't there, commit `uv.lock`, and deploy:

```bash
vercel deploy        # preview
vercel deploy --prod # production
```

_Vercel detects FastAPI, reads_ `[[tool.vercel.subscribers]]`_, and builds your subscribers into a queue-triggered function. On Vercel,_ `send()` _authenticates with the deployment's OIDC token automatically._

Push deliveries go to the publishing deployment by default, so orders queued before a release still run through that deployment's subscribers.

Old deployments can receive retries until messages are acknowledged, expire, or reach a delivery cap. Keep them available while work remains because promoting a release does not move the backlog to new code.

Open your project's **Observability** tab and select **Queues**. Each queue shows **Max Message Age** and **Retry Depth** by consumer group.

Rising Retry Depth means at least one message is being delivered repeatedly. Check the affected group's logs for handler errors or a downstream outage. With Observability Plus, the **Queue Retry Depth** alert preset can notify you when retries cross its threshold.

## Use cases

The following patterns build on the checkout example. The CRM, audit-log, and import snippets are integration excerpts: supply your own `crm`, `audit_log`, and `loader` clients before adding them to a subscriber module.

| Use case                                 | Building blocks                                 | Example          |
| ---------------------------------------- | ----------------------------------------------- | ---------------- |
| Moving slow work out of a request        | `send()` plus one subscriber                    | Steps 2 and 3    |
| One event, several independent reactions | Several consumer groups on one topic            | Step 2           |
| Calling a flaky or rate-limited API      | `RetryAfter`, `max_attempts`, `max_concurrency` | Step 4 and below |
| Work that should happen later            | `send(..., delay=...)`                          | Below            |
| Handling a family of events in one place | Wildcard topics                                 | Below            |
| Large files and imports                  | Binary and streaming payloads                   | Below            |
| Always-on workers on Vercel Daemons      | `poll()` with `ALL_DEPLOYMENTS`                 | Below            |

### Do something later

Set `delay` to postpone delivery and `retention` to leave time for processing afterward. This example delays a reminder for one day and retains it for two:

```python
from datetime import timedelta

from pydantic import BaseModel
from vercel.queue import Topic, send


class CartReminder(BaseModel):
    cart_id: str


reminders = Topic[CartReminder]("cart-reminders")


async def schedule_reminder(cart_id: str) -> None:
    await send(
        reminders,
        CartReminder(cart_id=cart_id),
        delay=timedelta(hours=24),
        retention=timedelta(days=2),
        idempotency_key=f"cart-reminder-{cart_id}",
    )
```

Call `schedule_reminder(cart_id)` when a cart becomes eligible for a reminder. Its subscriber should check whether the purchase has since completed before sending email. Both delay and retention have a seven-day maximum; messages expire when retention ends, even if they have not been processed.

For one follow-up, a delayed message is the simplest tool. When a process has several waits and decisions in sequence, use a [workflow](https://workflow-sdk.dev/docs/getting-started/python) instead (below).

### Protect a downstream API

If a subscriber calls an API with a rate limit, cap how many deliveries run at once:

```python
@subscribe(topic=orders, consumer_group="crm-sync", max_concurrency=5)
async def sync_to_crm(order: Order) -> None:
    await crm.upsert_customer(email=order.email, last_order=order.order_id)
```

This caps concurrent push deliveries for this consumer, but account-wide CRM and requests-per-second limits need separate controls. Coordinate other consumers and follow the provider's retry guidance.

### Handle a family of events

Use a trailing `*` to subscribe to topics with a shared prefix, such as events written to an audit log:

```python
@subscribe(topic="user-*")
async def audit_user_event(message: Message[dict[str, str]]) -> None:
    await audit_log.write(topic=message.metadata.topic, event=message.payload)
```

This handler receives events from `user-created`, `user-deleted`, and any other topic with the `user-` prefix.

### Move large payloads

Topics typed as `bytes` or `str` carry raw content, and streaming topics accept `AsyncIterable[bytes]`. The queue supports payloads up to 100 MB, but the subscriber still needs enough execution time and memory to process them.

For a CSV import, use an incremental loader that buffers incomplete records and handles text characters split across chunks:

```python
from collections.abc import AsyncIterable

from vercel.queue import ByteStreamTransport, Topic, subscribe

imports = Topic[AsyncIterable[bytes]]("csv-imports", transport=ByteStreamTransport())


@subscribe(topic=imports)
async def load_rows(chunks: AsyncIterable[bytes]) -> None:
    async for chunk in chunks:
        await loader.feed(chunk)
```

The SDK uses `ByteStreamTransport` to read and write byte streams. Without it, an asynchronous iterable is treated as JSON and fails serialization.

### Run a polling worker on Vercel Daemons

If you need an always-on worker that pulls messages, run it on [Vercel Daemons](https://vercel.com/kb/guide/vercel-daemons) with [poll mode](https://vercel.com/docs/queues/poll-mode). The SDK reads the OIDC token file provided to each instance.

Move the Step 4 handler to `workers/warehouse.py` and add an empty `workers/__init__.py`. Import `Order` from `queues.orders`, retain `MAX_SHIPMENT_ATTEMPTS` and the other imports, and remove `@subscribe`.

Delete the warehouse handler from `queues/orders.py` and leave the receipt subscriber. This avoids competing push and poll consumers.

Save the polling loop as `worker.py`:

```python
import asyncio
import logging
import signal

from vercel.queue import ALL_DEPLOYMENTS, QueueClient

from queues.orders import orders
from workers.warehouse import notify_warehouse

logger = logging.getLogger(__name__)


async def main() -> None:
    queue = QueueClient(region="iad1", deployment=ALL_DEPLOYMENTS)
    stopping = asyncio.Event()
    asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stopping.set)

    while not stopping.is_set():
        handled = False
        try:
            async for delivery in queue.poll(orders, "warehouse", limit=1):
                handled = True
                try:
                    async with delivery as message:
                        await notify_warehouse(message)
                except Exception:
                    logger.exception("Warehouse delivery failed; left unacknowledged")
        except Exception:
            logger.exception("Queue polling failed")

        if not handled and not stopping.is_set():
            try:
                await asyncio.wait_for(stopping.wait(), timeout=1)
            except TimeoutError:
                pass


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())
```

The delivery context acknowledges a successful handler and handles `RetryAfter` directives. Other exceptions leave the message unacknowledged and propagate, so the loop logs them and continues. The application-level failure threshold from Step 4 still runs because it is part of the handler; decorator options such as `max_attempts` do not govern direct calls from this loop.

#### Select the deployment scope and region

Daemon instances do not receive `VERCEL_DEPLOYMENT_ID`. This worker uses `ALL_DEPLOYMENTS` so it can process retained orders across production rollouts. That scope also includes preview deployments, so use a separate topic for preview traffic. Set the client region to the region where the producer sends its messages.

#### Allow time to finish a delivery

The loop requests one message at a time and checks for `SIGTERM` before the next poll. Vercel allows 30 seconds for shutdown by default. Configure `terminationGracePeriodSeconds` and provider request timeouts so the current delivery can finish within that window; otherwise it remains available for retry after its lease expires.

`vercel dev` does not run Daemons. Build and test the worker container separately before deploying it, using the [Daemon guide](https://vercel.com/kb/guide/vercel-daemons) for its service configuration.

Moving an existing Celery app? Vercel documents how to [run Celery tasks on Vercel Queues](https://vercel.com/docs/frameworks/backend/celery).

### When to use something else

| Requirement                                        | Use                                                                                                                              |
| -------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- |
| Several steps with waits or decisions between them | [Vercel Workflows](https://workflow-sdk.dev/docs/getting-started/python) to persist progress and resume after a wait or approval |
| Work on a fixed schedule                           | [Cron Jobs](https://vercel.com/docs/cron-jobs), optionally publishing queued jobs for each scheduled batch                       |
| Work required to produce the current response      | Keep it in the request so the caller receives the result                                                                         |

## Next steps

- Read the full [Python SDK reference](https://vercel.com/docs/queues/python-sdk) for transports, manual polling, leases, and every error class
  
- Learn how delivery, retention, and deployments interact in [Queues concepts](https://vercel.com/docs/queues/concepts)
  
- Watch throughput and retries with [Queues observability](https://vercel.com/docs/queues/observability)
  
- Check [pricing and limits](https://vercel.com/docs/queues/pricing) before you send at volume

## Related Resources

- [Backends on Vercel: Python, containers, and Daemons](https://vercel.com/blog/backends-on-vercel-python-containers-and-daemons): Explore how Python, container images, and background workers fit into the Vercel platform.