Python SDK Reference

The official Vercel Python SDK lets Python apps publish and consume Vercel Queues messages. It includes async and sync clients, typed topics, push delivery helpers, automatic polling loops, manual polling, and transports for JSON, text, binary, and streaming payloads.

For JavaScript and TypeScript, see the JS SDK Reference.

uv add vercel

pip install vercel

This installs the full Vercel SDK and is recommended when you use other Vercel products. For a minimalist installation that includes only the Queues SDK, use vercel-queue instead.

uv add vercel-queue

pip install vercel-queue

The above command installs just the Queues SDK with optional features disabled. If your application relies Pydantic-backed typed message payloads, install vercel-queue[typed].

Import top-level helpers from vercel.queue for route handlers, short scripts, polling workers, and tasks. Operation helpers such as send, poll, poll_and_handle, and accept_and_handle create a default async client for the call, so you don't need to construct a client first.

from vercel.queue import (
    Message,
    QueueClient,
    Topic,
    accept_and_handle,
    poll,
    poll_and_handle,
    send,
    subscribe,
)

Create a QueueClient when you need to configure region, headers, timeouts, deployment partitioning, or a custom queue service URL.

from vercel.queue import QueueClient
 
queue = QueueClient(region="sfo1")

Clients are lightweight and hold no open connections, so create one at module scope and share it across requests. QueueClient is not a context manager.

from fastapi import FastAPI, Request
from vercel.queue import QueueClient
 
app = FastAPI()
queue = QueueClient(region="sfo1")
 
 
@app.post("/api/orders")
async def create_order(request: Request):
    body = await request.json()
    message_id = await queue.send("orders", body)
    return {"messageId": message_id}

Set deployment to a deployment ID string to pin requests to one deployment, or to ALL_DEPLOYMENTS to send and poll across all deployments. The default targets the current deployment from VERCEL_DEPLOYMENT_ID.

The base_url value can be a fixed URL, a template containing a {region} placeholder, or a callable that takes a region name and returns a URL.

from vercel.queue import QueueClient
 
queue = QueueClient(base_url="https://proxy.example/queues/{region}")

Use send to publish a message to a topic. When you pass a Topic[T], the SDK uses the topic's payload type and transport. Otherwise, the SDK infers a serializer from the payload type.

from fastapi import FastAPI, Request
from vercel.queue import send
 
app = FastAPI()
 
 
@app.post("/api/orders")
async def create_order(request: Request):
    body = await request.json()
    message_id = await send(
        "orders",
        {"orderId": body["orderId"], "action": "process"},
    )
    return {"messageId": message_id}

from datetime import timedelta
from vercel.queue import send
 
message_id = await send(
    "orders",
    payload,
    retention=timedelta(hours=1),
    delay=60,
    idempotency_key="order-123",
    headers={"x-trace-id": "abc-123"},
)

send returns the created message ID. It returns None when the service accepted the message but deferred ingestion. Deferred messages are still delivered.

The message format is part of the topic contract. Use Topic[T] to declare the payload type for a topic. send(), poll(), and subscribers use topic declarations or handler annotations to choose the normal transport automatically.

from typing import TypedDict
from vercel.queue import Topic, send, subscribe
 
 
class Email(TypedDict):
    to: str
    subject: str
 
 
emails = Topic[Email]("emails")
 
 
async def queue_email(email: Email) -> None:
    await send(emails, email)
 
 
@subscribe(topic=emails)
async def receive_email(email: Email) -> None:
    await send_email(email)

You can specify a transport explicitly on the topic when the topic is untyped or when you need custom serialization. This keeps send and receive using the same message format.

from collections.abc import AsyncIterable, AsyncIterator
from vercel.queue import ByteStreamTransport, Topic, send, subscribe
 
large_file = Topic[AsyncIterable[bytes]](
    "large-file",
    transport=ByteStreamTransport(),
)
 
 
async def file_chunks() -> AsyncIterator[bytes]:
    with open("large.bin", "rb") as file:
        while chunk := file.read(1024 * 1024):
            yield chunk
 
 
async def send_file() -> None:
    await send(large_file, file_chunks())
 
 
@subscribe(topic=large_file)
async def archive_file(chunks: AsyncIterable[bytes]) -> None:
    async for chunk in chunks:
        await write_chunk(chunk)

Pydantic models can be sent directly. A typed topic infers typed JSON validation and deserialization on receive.

from pydantic import BaseModel
from vercel.queue import Topic, send, subscribe
 
 
class Order(BaseModel):
    order_id: str
    total_cents: int
 
 
orders = Topic[Order]("orders")
 
 
async def queue_order() -> None:
    await send(orders, Order(order_id="ord_123", total_cents=2500))
 
 
@subscribe(topic=orders)
async def process_typed_order(order: Order) -> None:
    await process_order(order)

Push mode is the default for Python subscribers deployed to Vercel. Register a function with @subscribe, then declare its Python module import path under [[tool.vercel.subscribers]] in pyproject.toml.

from vercel.queue import Message, Topic, subscribe
 
orders = Topic[dict[str, object]]("orders")
 
 
@subscribe(topic=orders)
async def fulfill_order(message: Message[dict[str, object]]) -> None:
    await process_order(message.payload)

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

At build time, Vercel imports the entrypoint module, reads every subscription it registers, and compiles the subscriber into a private queue-triggered function. You don't need to configure experimentalTriggers in vercel.json.

When consumer_group is omitted, the SDK derives a stable group name from the subscriber's fully qualified Python name. Multiple subscribers for the same topic create separate consumer groups, and each consumer group receives a copy of every message.

The SDK accepts the delivery, renews the processing lease while your handler is running, and acknowledges the message when the handler returns. If the handler raises an exception, Vercel redelivers the message according to queue retry behavior.

The entrypoint value is a Python module import path. Use a dotted module path such as queues.orders for queues/orders.py. Don't use a filesystem path or include the .py suffix.

Each [[tool.vercel.subscribers]] entry identifies a Python module that registers one or more subscribers. The module:object format is also supported when you need to identify an object in the module.

When a module registers several subscribers, the generated function consumes all of them. Add a topics filter to split the module's subscribers across separate generated functions.

[[tool.vercel.subscribers]]
entrypoint = "worker"
topics = ["orders"]
 
[[tool.vercel.subscribers]]
entrypoint = "refunds"
topics = ["refunds"]

Keep delivery configuration on the @subscribe decorator. Vercel reads topic, consumer_group, retry_after, initial_delay, max_concurrency, and max_attempts when it generates the queue trigger.

Consumer group names can be any non-empty string. The SDK escapes them into queue-safe names automatically. Use sanitize_name to compute the stored name yourself, and pass a SanitizedName when a value is already queue-safe and must not be escaped again.

Subscribe to multiple topics with a wildcard pattern:

from vercel.queue import subscribe
 
 
@subscribe(topic="user-*")
async def handle_user_event(event: dict[str, str]) -> None:
    await process_user_event(event)

Subscribers receive a Message[T]. Use message.payload for the deserialized payload, message.message_id for the message ID, and message.metadata for delivery metadata.

Use accept_and_handle when you need to route a callback through an existing ASGI framework. It accepts the callback body as bytes, a byte iterable, or a framework response object, plus the callback request headers. Pass lease_duration to change the processing timeout used while handlers run.

from fastapi import Request
from vercel.queue import accept_and_handle
 
 
async def handle_queue_callback(request: Request) -> None:
    body = await request.body()
    await accept_and_handle(body, request.headers, lease_duration=300)

It raises UnhandledMessageError when no registered subscription matches the delivered topic.

Use poll_and_handle() to run a subscriber outside push mode, such as in a self-hosted worker, local process, or long-running script. The helper uses the subscriber's @subscribe metadata to pick the topic, consumer group, payload type, and receive transport.

import asyncio
from vercel.queue import poll_and_handle, subscribe
 
 
@subscribe(topic="orders")
async def fulfill_order(order: dict[str, str]) -> None:
    await process_order(order)
 
 
async def main() -> None:
    async with asyncio.TaskGroup() as task_group:
        poller = task_group.create_task(
            poll_and_handle(fulfill_order, interval=1.0),
        )
        try:
            await wait_for_shutdown_signal()
        finally:
            poller.cancel()
 
 
asyncio.run(main())

poll_and_handle() polls each configured topic until no messages are available, then sleeps for interval before checking again. The loop acknowledges a message when the subscriber returns. If the subscriber raises, the SDK leaves the message unacknowledged so Vercel Queues can redeliver it according to retry behavior.

Wildcard subscribers can run in a polling loop, but you must pass concrete topic names because wildcard topic patterns cannot be polled directly.

from vercel.queue import poll_and_handle, subscribe
 
 
@subscribe(topic="events-*")
async def handle_event(event: dict[str, str]) -> None:
    await record_event(event)
 
 
await poll_and_handle(
    handle_event,
    topics=["events-user", "events-system"],
    interval=1.0,
)

Use QueueClient.poll_and_handle() when the polling worker needs explicit region, authentication, headers, deployment partitioning, or a custom queue service URL.

import asyncio
from vercel.queue import QueueClient, subscribe
 
 
@subscribe(topic="orders")
async def fulfill_order(order: dict[str, str]) -> None:
    await process_order(order)
 
 
async def main() -> None:
    queue = QueueClient(region="iad1")
    async with asyncio.TaskGroup() as task_group:
        poller = task_group.create_task(
            queue.poll_and_handle(fulfill_order, interval=1.0),
        )
        try:
            await wait_for_shutdown_signal()
        finally:
            poller.cancel()
 
 
asyncio.run(main())

Messages can only be received from the region they were sent to. Use a fixed region, such as iad1, for both sending and polling. Avoid a changing runtime region for polling workers because that can distribute messages across regions unpredictably.

Use poll() when you need direct control over delivery lifecycles or explicit consumer group behavior. poll() polls once for a specified consumer group and yields up to limit Delivery[T] objects. It can return no deliveries, so long-running workers usually use poll_and_handle() or add their own loop around poll().

from vercel.queue import QueueClient, Topic
 
orders = Topic[dict[str, object]]("orders")
CONSUMER_GROUP = "fulfillment"
 
 
queue = QueueClient(region="iad1")
 
 
async def poll_once() -> None:
    async for delivery in queue.poll(
        orders,
        CONSUMER_GROUP,
        limit=10,
        lease_duration=300,
    ):
        async with delivery as message:
            await process_order(message.payload)

Entering a delivery starts automatic lease renewal and returns the deserialized Message[T]. A clean exit acknowledges the message, while an exception leaves it available for retry according to queue behavior.

Use the same consumer group name in multiple pollers when those pollers should compete for work. Use different consumer group names when each group should receive its own copy of every message.

Use acknowledge(), extend_lease(), and retry_after() when you intentionally manage a delivery lifecycle yourself. Call delivery.accept() to take ownership of the message. Accepted deliveries skip automatic lease renewal and acknowledgement.

from vercel.queue import QueueClient
 
CONSUMER_GROUP = "fulfillment"
 
queue = QueueClient(region="iad1")
 
 
async def process_batch() -> None:
    async for delivery in queue.poll("orders", CONSUMER_GROUP):
        message = delivery.accept()
        await queue.extend_lease(message, 600)
        await process_order(message.payload)
        await queue.acknowledge(message)

Pass zero to extend_lease() to release a message back to the queue immediately. Use retry_after() to schedule redelivery after a delay when your code owns the lifecycle. Handlers should usually raise RetryAfter instead.

Vercel Queues delivers messages at least once. When a subscriber raises an exception, the SDK leaves the message unacknowledged, and the message becomes visible again after the retry_after interval configured on @subscribe. Redelivery continues until the handler succeeds or the message expires.

Raise RetryAfter from a subscriber to control the next delivery time directly. The SDK stops lease renewal, makes the message visible again after the delay, and treats the delivery as handled. The default delay is 60 seconds, and a delay of zero requests immediate redelivery.

from vercel.queue import Message, RetryAfter, subscribe
 
 
@subscribe(topic="orders")
async def fulfill_order(message: Message[dict[str, str]]) -> None:
    try:
        await process_order(message.payload)
    except TemporaryError as exc:
        delay = min(300, 2**message.metadata.delivery_count * 5)
        raise RetryAfter(delay) from exc

Use message.metadata.delivery_count to add exponential backoff, as shown above.

Raise Handoff when your handler passed the delivery to another system that owns the rest of its lifecycle. The SDK stops lease renewal and leaves the lease open. The external system must acknowledge the message or change its visibility with the original message metadata, or the message is redelivered when the lease expires.

RetryAfter and Handoff both extend QueueDirective. They work in push mode, in poll_and_handle() loops, and inside entered Delivery context managers.

The sync API mirrors the async send and manual polling APIs under vercel.queue.sync. Use vercel.queue.sync.QueueClient.poll_and_handle() to run a subscriber in a background polling thread.

from vercel.queue import subscribe
from vercel.queue.sync import QueueClient
 
 
@subscribe(topic="events")
def handle_event(event: dict[str, str]) -> None:
    process_event(event)
 
 
queue = QueueClient(region="iad1")
queue.send("events", {"type": "user.created"})
poller = queue.poll_and_handle(handle_event, interval=1.0)
try:
    wait_for_shutdown_signal()
finally:
    poller.cancel()

The sync polling loop runs in a daemon thread and returns a concurrent.futures.Future[None]. Calling cancel() asks the polling thread to stop. Calling result() surfaces polling errors, or raises CancelledError after cancellation.

The Python SDK exports typed error classes from vercel.queue.

from vercel.queue import DuplicateIdempotencyKeyError, QueueError, send
 
try:
    await send("orders", payload, idempotency_key="order-123")
except DuplicateIdempotencyKeyError:
    # The idempotency key was already used.
    pass
except QueueError:
    # Handle other queue service failures.
    raise

Common errors include:

Use the embedded queue service to exercise the full send, dispatch, lease renewal, and acknowledgement path in one process without deploying.

from vercel.queue import subscribe
from vercel.queue.embedded import embedded_queue_service
 
 
@subscribe(topic="emails")
async def handle_email(email: dict[str, str]) -> None:
    await record_email(email)
 
 
async def send_one() -> None:
    async with embedded_queue_service() as service:
        client = service.get_async_client()
        await client.send("emails", {"subject": "Hi"})

For pytest, enable the bundled plugin and use the embedded_queue_server fixture. Each test gets isolated queue state.

pytest_plugins = ["vercel.queue.testing.pytest"]

async def test_send(embedded_queue_server):
    client = embedded_queue_server.get_async_client()
    message_id = await client.send("emails", {"subject": "Hi"}, retention=60)
    assert message_id is not None

For cross-process or cross-runtime local integration, install the devserver extra (vercel-queue[devserver]) and run a standalone queue API server:

python -m vercel.queue.devserver --port 8000

The command prints a JSON baseUrl for the local queue API. When --port is omitted, it picks a random available port. Point clients at the printed URL with the VERCEL_QUEUE_BASE_URL environment variable or the base_url client option.

The SDK reads these variables when you don't pass the matching option explicitly:

• JS SDK Reference
• API reference
• Poll mode