from fastapi import FastAPI, Request

from vercel.queue import QueueClient

app =FastAPI()

queue =QueueClient(region="sfo1")

@app.post("/api/orders")

asyncdefcreate_order(request: Request):

body =await request.json()

message_id =await queue.send("orders", body)

return{"messageId": message_id}

Client options

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}")

Publishing messages

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.

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

Topic types and message formats

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.

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.

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

classOrder(BaseModel):

order_id:str

total_cents:int

orders = Topic[Order]("orders")

asyncdefqueue_order() ->None:

awaitsend(orders, Order(order_id="ord_123", total_cents=2500))

@subscribe(topic=orders)

asyncdefprocess_typed_order(order: Order) ->None:

awaitprocess_order(order)

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.queueimport subscribe

@subscribe(topic="user-*")

asyncdefhandle_user_event(event: dict[str,str]) ->None:

awaitprocess_user_event(event)

Push message metadata

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.

from fastapi import Request

from vercel.queue import accept_and_handle

asyncdefhandle_queue_callback(request: Request) ->None:

body =await request.body()

awaitaccept_and_handle(body, request.headers, lease_duration=300)

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

Consuming messages with polling loops

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.

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.

Polling loop options

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

Polling regions

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.

Consuming messages with manual polling

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

CONSUMER_GROUP ="fulfillment"

queue =QueueClient(region="iad1")

asyncdefprocess_batch() ->None:

asyncfor delivery in queue.poll("orders", CONSUMER_GROUP):

message = delivery.accept()

await queue.extend_lease(message, 600)

awaitprocess_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.

Retries and redelivery

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 subscribe

from vercel.queue.embedded import embedded_queue_service

@subscribe(topic="emails")

asyncdefhandle_email(email: dict[str,str]) ->None:

awaitrecord_email(email)

asyncdefsend_one() ->None:

asyncwithembedded_queue_service()as service:

client = service.get_async_client()

await client.send("emails", {"subject": "Hi"})

python -mvercel.queue.devserver--port8000

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.

Environment variables

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