Implement Vercel Queue SDK

[elprans]

Introduce vercel-queue, a Python client for publishing to and consuming from
the Vercel Queues API.

Features:

• Simple API surface: send(), @subscribe, and poll()
• Customizable Clients: configurable QueueQlient
• Full async/sync support: async first, import the same API through
  vercel.queue.sync
• Type Safety: Topic[T], typed messages, and optional Pydantic validation
• Zero-copy streaming: subscribers can be typed to receive
  AsyncIterator[bytes] in which case the byte stream is passed along
  from the incoming message HTTP stream directly without copying or
  extra buffering
• Automatic message lifecycle: messages are automatically
  acknowledged on successful handler exit; message lease is
  automatically extended while the handler is running
• Customizable Serialization: Built-in JSON, text, binary, and streaming
  transports, ability to define and use custom transports
• Local Development Support: complete local queue server emulator that can
  be used both in-process and as a standalone local server process for
  testing and development.

See README.md for more details.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
vercel-py	Ready	Preview	Jul 8, 2026 6:14pm

[socket-security]

Review the following changes in direct dependencies. Learn more about Socket for GitHub.

Diff	Package	Supply Chain Security	Vulnerability	Quality	Maintenance	License
	pypi/​tox@​4.56.1
	pypi/​uvicorn@​0.49.0
	pypi/​python-multipart@​0.0.32
	pypi/​tox-uv-bare@​1.35.2
	pypi/​time-machine@​2.19.0
	pypi/​ty@​0.0.55
	pypi/​ruff@​0.15.12 ⏵ 0.15.20	+1
	pypi/​fastapi@​0.138.1

View full report

[ricardo-agz]

I think we need a first class primitive for running the subscriber outside of vercel in poll mode, something like:

from vercel.queue import subscribe, subscriber

@subscribe(topic="orders", retry_after=60)
async def fulfill(order: dict[str, str]) -> None:
    await process(order)

if __name__ == "__main__":
    subscriber.poll(region="iad1")

You shouldn't need anything more than those 2 lines to migrate out of vercel

[ricardo-agz]

I also don't think consumer groups should be required, this should be the minimal example:

from vercel.queue import subscribe, subscriber

@subscribe(topic="orders")
async def fulfill(order: dict[str, str]) -> None:
    await process(order)

And in cases where you do want to explicitly set it, like for poll mode, the examples shouldn't require basing it off of the file name and sanitizing it. The DX I'd want is:

from vercel.queue import subscribe, subscriber

@subscribe(topic="orders", consumer="a")
async def fulfill(order: dict[str, str]) -> None:
    await process(order)

[ricardo-agz]

I think exposing app = get_asgi_app as a user-facing thing is ugly. The vercel-workers runtime hid it away by automatically bootstrapping the subscriber's asgi app

Ideally, I'd want the user-facing code to look like this:

worker.py

from vercel.queue import subscribe, Subscriber

@subscribe(topic="orders")
async def fulfill_order(order: dict[str, str]) -> None:
    await process_order(order)

# app = asgi_app() # <- leaks implementation details

subscriber = Subscriber() # <- this can have a get_asgi_app() method that we can look for in the runtime

if __name__ == "__main__":
    subscriber.poll(region="iad1")

And config can look like:

[[tool.vercel.subscribers]]
entrypoint = "worker:subscriber"

or if we don't like subscriber = Subscriber() then the following should work:
worker.py

from vercel.queue import subscribe, subscriber

@subscribe(topic="orders")
async def fulfill_order(order: dict[str, str]) -> None:
    await process_order(order)

# app = asgi_app() # <- not required

config:

[[tool.vercel.subscribers]]
entrypoint = "worker.py" # we bootstrap the asgi app in the runtime
                         # worker.py just needs to import all the subscribe tasks

[ricardo-agz]

It's good that this works and it definitely should, but let's not teach the LLMs to do this, I'd rather commit to [[tool.vercel.subscribers]] from the beginning and use that in the README and docs and examples

[ricardo-agz]

also functions is a footgun since it creates a function for every entry (and it can be a glob)
Ideally we'd want to be able to control how we pack subscribers into functions

[elprans]

Yeah, I oscillated between publishing something that is less optimal and works now and can be tested vs something that is better but doesn't work in this PR. The plan is to release vercel.queue with proper documentation for subscribers (and coordinate the release with the landing of that).

[elprans]

I think exposing app = get_asgi_app as a user-facing thing is ugly. The vercel-workers runtime hid it away by automatically bootstrapping the subscriber's asgi app

The asgi_app thing is temporary so that we can test this impl with plain api/ functions and not get blocked on support for subscribers.

[elprans]

I think we need a first class primitive for running the subscriber outside of vercel in poll mode, something like:

Added poll_and_handle(subscriber). I'm wary of adding a wholesale "poll all subscribers" helper as that can be a bit of a foot gun given that topics must be polled individually and so making a convenient/implicit helper would encourage a lot of polling. Additionally, this would not work for wildcard subs at all.

[elprans]

I also don't think consumer groups should be required, this should be the minimal example:

I adjusted most examples to avoid setting explicit group names. For plain api/ functions they are still required, but I made sanitation internal/automatic, so you can just subscriber_group=f"api/{__name__}.py" now