Pico W: async web server with uasyncio, the modern pattern
Serve a real async web API from a Pico W using uasyncio, with background sensor tasks and non-blocking request handlers. The right way after the synchronous server gets too small.
The synchronous web server tutorial gets you 80% of the way. You bind a socket, accept a connection, send a response, close. It works for one request at a time. The moment you have a second request, the first one stalls, and the moment you want to read a sensor while you are serving, the sensor read blocks the response.
This tutorial is the upgrade: uasyncio, async def handlers, background
tasks, and the gc.collect() pattern that keeps the Pico W from running
out of RAM.
What you need
- Raspberry Pi Pico W
- MicroPython firmware 1.20+ (the
uasynciomodule is standard) - A sensor (the BME280 is a good all-in-one) if you want to do the background-task example
What uasyncio is
uasyncio is MicroPython’s version of asyncio, the Python standard
library’s cooperative multitasking library. Instead of threads, you have
coroutines. Instead of locks, you have await. A coroutine runs until it
hits an await, then yields control, then resumes when the awaited thing
is done.
This is the right model for microcontrollers. You do not have the memory for threads. You have one CPU. Cooperative multitasking on a single CPU is enough.
Sync vs async, the difference that matters
Synchronous server (the old pattern):
import socket
addr = socket.getaddrinfo("0.0.0.0", 80)[0][-1]
s = socket.socket()
s.bind(addr)
s.listen(5)
while True:
cl, addr = s.accept()
req = cl.recv(1024) # blocks here
cl.send(response) # blocks here
cl.close()
The recv and send block. If the client is slow, the server is slow.
You cannot read a sensor while you are serving. The server is busy
waiting on a network call.
Async server (the new pattern):
import asyncio
import uasyncio as asyncio
async def handle(reader, writer):
req = await reader.read(1024) # does not block the event loop
writer.write(response)
await writer.drain()
writer.close()
async def main():
server = await asyncio.start_server(handle, "0.0.0.0", 80)
await server.serve_forever()
asyncio.run(main())
While the await reader.read() is in flight, the event loop is free to
run other coroutines. A second client can connect. A background sensor
task can publish.
The request handler structure
Every async request handler is a function that takes a reader and a
writer. You read the request, write the response, close the writer.
async def handle(reader, writer):
try:
request_line = await reader.readline()
# The request line is "GET /path HTTP/1.1\r\n"
# Parse it
method, path, _ = request_line.decode().split(" ", 2)
# Read headers until empty line
while True:
line = await reader.readline()
if line == b"\r\n":
break
# Build a response
if path == "/":
body = "Hello from the Pico W"
elif path == "/temp":
body = "23.5 C"
else:
body = "Not found"
response = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/plain\r\n"
f"Content-Length: {len(body)}\r\n"
"Connection: close\r\n"
"\r\n"
+ body
)
writer.write(response.encode())
await writer.drain()
finally:
writer.close()
await writer.wait_closed()
The try / finally makes sure the connection closes even if the
request raises an exception. A slow client that disconnects mid-request
will not leak the writer.
Handling multiple connections
asyncio.start_server accepts connections in a loop. For each new
connection, the library creates a task running your handle function.
Multiple clients can be connected at the same time. The total number is
limited by the Pico W’s RAM (about 20 KB free after MicroPython starts).
Plan for 3-4 concurrent connections.
The gc.collect() gotcha
MicroPython’s garbage collector does not run as often as CPython’s. On a
Pico W, after a few hundred requests, you can see MemoryError even
though the heap has plenty of free space. The memory is fragmented.
The fix: call gc.collect() periodically, especially between requests
in a long-running server.
import gc
async def handle(reader, writer):
try:
# ... request handling ...
pass
finally:
writer.close()
await writer.wait_closed()
gc.collect() # run after every request
I put gc.collect() in the finally of every handler. It is not free
(it pauses the server for a few milliseconds), but the alternative is a
crash at request 500.
Running sensors in background tasks
The pattern I use in every Pico W web project: spawn a background task
with asyncio.create_task(), and have it update a shared state object
that the request handlers read.
import uasyncio as asyncio
import machine
import gc
# Shared state, updated by the sensor task
state = {"temp": 0.0, "humidity": 0.0, "updated": False}
async def sensor_task():
while True:
# Read your sensor here
# For demo, simulate
await asyncio.sleep(2)
state["temp"] = 23.5
state["humidity"] = 50.0
state["updated"] = True
gc.collect()
async def handle(reader, writer):
try:
request_line = await reader.readline()
# ... read rest of headers ...
if b"GET /temp" in request_line:
body = f'{state["temp"]} C'
else:
body = "OK"
response = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/plain\r\n"
f"Content-Length: {len(body)}\r\n"
"Connection: close\r\n"
"\r\n"
+ body
)
writer.write(response.encode())
await writer.drain()
finally:
writer.close()
await writer.wait_closed()
gc.collect()
async def main():
asyncio.create_task(sensor_task())
server = await asyncio.start_server(handle, "0.0.0.0", 80)
await server.serve_forever()
asyncio.run(main())
The sensor runs in the background, updates state, and the request
handler reads from state. The two never block each other.
server.serve_forever() vs server.start()
serve_forever() is a coroutine you await to block forever. The event
loop runs in the same task as the server. Use this when you have nothing
else to do at the top level.
start() schedules the server to run in the background and returns
immediately. Use this when you have other work to do at the top level
(e.g. a status LED blink):
async def main():
server = await asyncio.start_server(handle, "0.0.0.0", 80)
asyncio.create_task(server.serve_forever())
asyncio.create_task(blink_task()) # your other task
await asyncio.sleep_forever()
asyncio.sleep_forever() is a coroutine that never returns. It exists
so the event loop has something to do at the top level.
Comparing to ESP32 AsyncWebServer
The ESP32 has a similar pattern but with two key differences:
- The ESP32 has a lot more RAM (300+ KB free after Wi-Fi connects), so the
gc.collect()gotcha is less common. - The ESP32 Arduino core has an
AsyncWebServerlibrary that uses callbacks instead of coroutines. It is more performant but harder to read.
The Pico W aysncio pattern is easier to write and reason about. The ESP32 pattern scales better. For “Pico W serving a few clients,” go with MicroPython.
When blocking is fine
A blocking server is the right choice when:
- The server handles one request at a time (e.g. a configuration endpoint you hit once a day).
- The server is on a private network with trusted clients.
- You need it working in 20 minutes, not 40.
The async pattern is the right choice when:
- Multiple clients can connect at once.
- The handler has to do work that might block (sensor reads, file I/O).
- You are building a real API, not a config page.
For a “set the Wi-Fi SSID” endpoint, sync is fine. For a “stream sensor data” endpoint, async is the right tool.
What you learned
uasynciois MicroPython’s cooperative multitasking library. One CPU, no threads, coroutines all the way down.- The async server can handle multiple connections at once without one blocking another.
gc.collect()between requests is the price you pay for the Pico W’s small heap.- Background tasks update shared state, request handlers read from it.
When something breaks
MemoryError after a few hundred requests. Garbage collection. Add
gc.collect() in the handler’s finally block.
The server stops responding. Watchdog timer. Add a soft reset after N requests, or move the watchdog feeding into a background task.
The sensor reads always return 0. The sensor task never ran. Check
that asyncio.create_task(sensor_task()) is in main(), before the
await on the server.
Two requests at the same time corrupt the state. Add a lock:
asyncio.Lock() around the read-and-write of the shared state object.
OSError: [Errno 12] ENOMEM. The heap is exhausted. Reduce the
response size, or reduce the number of concurrent connections, or restart
the Pico W.
What to build next
- A weather station: BME280 in a background task, a
/weatherendpoint serving JSON. - A motor controller: a
/move?speed=50endpoint, background task reading encoders. - A real-time dashboard: WebSocket endpoint (covered in a separate tutorial on this site).
The weather station is the natural first project. The dashboard is the “this is actually useful” project.