Note
All examples in this document are using Server in debug mode.
This mode is useful for development, but it is not recommended to use it in production.
More about Debug mode at the end of Examples section.
Different ways of starting the server
There are several ways to start the server on CircuitPython, mostly depending on the device you are using and whether you have access to external network.
Functionally, all of them are the same, not features of the server are limited or disabled in any way.
Below you can find examples of different ways to start the server:
CPython usage
Library can also be used in CPython, no changes other than changing the socket_source are necessary.
1# SPDX-FileCopyrightText: 2024 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socket
6
7from adafruit_httpserver import Server, Request, Response
8
9
10pool = socket
11server = Server(pool, "/static", debug=True)
12
13
14@server.route("/")
15def base(request: Request):
16 """
17 Serve a default static plain text message.
18 """
19 return Response(request, "Hello from the CircuitPython HTTP Server!")
20
21
22# Ports below 1024 are reserved for root user only.
23# If you want to run this example on a port below 1024, you need to run it as root (or with `sudo`).
24server.serve_forever("0.0.0.0", 5000)
Serving static files
It is possible to serve static files from the filesystem.
In this example we are serving files from the /static directory.
In order to save memory, we are unregistering unused MIME types and registering additional ones. More about MIME types.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5
6import socketpool
7import wifi
8
9from adafruit_httpserver import Server, MIMETypes
10
11
12MIMETypes.configure(
13 default_to="text/plain",
14 # Unregistering unnecessary MIME types can save memory
15 keep_for=[".html", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".ico"],
16 # If you need to, you can add additional MIME types
17 register={".foo": "text/foo", ".bar": "text/bar"},
18)
19
20pool = socketpool.SocketPool(wifi.radio)
21server = Server(pool, "/static", debug=True)
22
23# You don't have to add any routes, by default the server will serve files
24# from it's root_path, which is set to "/static" in this example.
25
26# If you don't set a root_path, the server will not serve any files.
27
28server.serve_forever(str(wifi.radio.ipv4_address))
You can also serve a specific file from the handler.
By default FileResponse looks for the file in the server’s root_path directory
(/default-static-directory in the example below), but you can change it manually in every FileResponse
(to e.g. /other-static-directory, as in example below).
By doing that, you can serve files from multiple directories, and decide exactly which files are accessible.
1# SPDX-FileCopyrightText: 2022 Dan Halbert for Adafruit Industries, Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5
6import socketpool
7import wifi
8
9from adafruit_httpserver import Server, Request, FileResponse
10
11
12pool = socketpool.SocketPool(wifi.radio)
13server = Server(pool, "/default-static-folder", debug=True)
14
15
16@server.route("/home")
17def home(request: Request):
18 """
19 Serves the file /other-static-folder/home.html.
20 """
21
22 return FileResponse(request, "home.html", "/other-static-folder")
23
24
25server.serve_forever(str(wifi.radio.ipv4_address))
1<html lang="en">
2 <head>
3 <meta charset="UTF-8">
4 <meta http-equiv="X-UA-Compatible" content="IE=edge">
5 <meta name="viewport" content="width=device-width, initial-scale=1.0">
6 <title>Adafruit HTTPServer</title>
7 </head>
8 <body>
9 <p>Hello from the <strong>CircuitPython HTTP Server!</strong></p>
10 </body>
11</html>
Tasks between requests
If you want your code to do more than just serve web pages,
use the .start()/.poll() methods as shown in this example.
Between calling .poll() you can do something useful,
for example read a sensor and capture an average or
a running total of the last 10 samples.
.poll() return value can be used to check if there was a request and if it was handled.
1# SPDX-FileCopyrightText: 2022 Dan Halbert for Adafruit Industries
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import (
9 Server,
10 REQUEST_HANDLED_RESPONSE_SENT,
11 Request,
12 FileResponse,
13)
14
15
16pool = socketpool.SocketPool(wifi.radio)
17server = Server(pool, "/static", debug=True)
18
19
20@server.route("/")
21def base(request: Request):
22 """
23 Serve the default index.html file.
24 """
25 return FileResponse(request, "index.html")
26
27
28# Start the server.
29server.start(str(wifi.radio.ipv4_address))
30
31while True:
32 try:
33 # Do something useful in this section,
34 # for example read a sensor and capture an average,
35 # or a running total of the last 10 samples
36
37 # Process any waiting requests
38 pool_result = server.poll()
39
40 if pool_result == REQUEST_HANDLED_RESPONSE_SENT:
41 # Do something only after handling a request
42 pass
43
44 # If you want you can stop the server by calling server.stop() anywhere in your code
45 except OSError as error:
46 print(error)
47 continue
If you need to perform some action periodically, or there are multiple tasks that need to be done,
it might be better to use asyncio module to handle them, which makes it really easy to add new tasks
without needing to manually manage the timing of each task.
asyncio is not included in CircuitPython by default, it has to be installed separately.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5from asyncio import create_task, gather, run, sleep as async_sleep
6import socketpool
7import wifi
8
9from adafruit_httpserver import (
10 Server,
11 REQUEST_HANDLED_RESPONSE_SENT,
12 Request,
13 FileResponse,
14)
15
16
17pool = socketpool.SocketPool(wifi.radio)
18server = Server(pool, "/static", debug=True)
19
20
21@server.route("/")
22def base(request: Request):
23 """
24 Serve the default index.html file.
25 """
26 return FileResponse(request, "index.html")
27
28
29# Start the server.
30server.start(str(wifi.radio.ipv4_address))
31
32
33async def handle_http_requests():
34 while True:
35 # Process any waiting requests
36 pool_result = server.poll()
37
38 if pool_result == REQUEST_HANDLED_RESPONSE_SENT:
39 # Do something only after handling a request
40 pass
41
42 await async_sleep(0)
43
44
45async def do_something_useful():
46 while True:
47 # Do something useful in this section,
48 # for example read a sensor and capture an average,
49 # or a running total of the last 10 samples
50 await async_sleep(1)
51
52 # If you want you can stop the server by calling server.stop() anywhere in your code
53
54
55async def main():
56 await gather(
57 create_task(handle_http_requests()),
58 create_task(do_something_useful()),
59 )
60
61
62run(main())
Server with MDNS
It is possible to use the MDNS protocol to make the server accessible via a hostname in addition to an IP address. It is worth noting that it takes a bit longer to get the response from the server when accessing it via the hostname.
In this example, the server is accessible via the IP and http://custom-mdns-hostname.local:5000/.
On some routers it is also possible to use http://custom-mdns-hostname:5000/, but this is not guaranteed to work.
1# SPDX-FileCopyrightText: 2022 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import mdns
6import socketpool
7import wifi
8
9from adafruit_httpserver import Server, Request, FileResponse
10
11
12mdns_server = mdns.Server(wifi.radio)
13mdns_server.hostname = "custom-mdns-hostname"
14mdns_server.advertise_service(service_type="_http", protocol="_tcp", port=5000)
15
16pool = socketpool.SocketPool(wifi.radio)
17server = Server(pool, "/static", debug=True)
18
19
20@server.route("/")
21def base(request: Request):
22 """
23 Serve the default index.html file.
24 """
25
26 return FileResponse(request, "index.html", "/www")
27
28
29server.serve_forever(str(wifi.radio.ipv4_address))
Get CPU information
You can return data from sensors or any computed value as JSON. That makes it easy to use the data in other applications.
If you want to use the data in a web browser, it might be necessary to enable CORS. More info: https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS
1# SPDX-FileCopyrightText: 2022 Dan Halbert for Adafruit Industries
2#
3# SPDX-License-Identifier: Unlicense
4
5import microcontroller
6import socketpool
7import wifi
8
9from adafruit_httpserver import Server, Request, JSONResponse
10
11
12pool = socketpool.SocketPool(wifi.radio)
13server = Server(pool, debug=True)
14
15# (Optional) Allow cross-origin requests.
16server.headers = {
17 "Access-Control-Allow-Origin": "*",
18}
19
20
21@server.route("/cpu-information", append_slash=True)
22def cpu_information_handler(request: Request):
23 """
24 Return the current CPU temperature, frequency, and voltage as JSON.
25 """
26
27 data = {
28 "temperature": microcontroller.cpu.temperature,
29 "frequency": microcontroller.cpu.frequency,
30 "voltage": microcontroller.cpu.voltage,
31 }
32
33 return JSONResponse(request, data)
34
35
36server.serve_forever(str(wifi.radio.ipv4_address))
Handling different methods
On every server.route() call you can specify which HTTP methods are allowed.
By default, only GET method is allowed.
You can pass a list of methods or a single method as a string.
It is recommended to use the the values in adafruit_httpserver.methods module to avoid typos and for future proofness.
If you want to route a given path with and without trailing slash, use append_slash=True parameter.
In example below, handler for /api and /api/ route will be called when any of GET, POST, PUT, DELETE methods is used.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, JSONResponse, GET, POST, PUT, DELETE
9
10
11pool = socketpool.SocketPool(wifi.radio)
12server = Server(pool, debug=True)
13
14objects = [
15 {"id": 1, "name": "Object 1"},
16]
17
18
19@server.route("/api", [GET, POST, PUT, DELETE], append_slash=True)
20def api(request: Request):
21 """
22 Performs different operations depending on the HTTP method.
23 """
24
25 # Get objects
26 if request.method == GET:
27 return JSONResponse(request, objects)
28
29 # Upload or update objects
30 if request.method in [POST, PUT]:
31 uploaded_object = request.json()
32
33 # Find object with same ID
34 for i, obj in enumerate(objects):
35 if obj["id"] == uploaded_object["id"]:
36 objects[i] = uploaded_object
37
38 return JSONResponse(
39 request, {"message": "Object updated", "object": uploaded_object}
40 )
41
42 # If not found, add it
43 objects.append(uploaded_object)
44 return JSONResponse(
45 request, {"message": "Object added", "object": uploaded_object}
46 )
47
48 # Delete objects
49 if request.method == DELETE:
50 deleted_object = request.json()
51
52 # Find object with same ID
53 for i, obj in enumerate(objects):
54 if obj["id"] == deleted_object["id"]:
55 del objects[i]
56
57 return JSONResponse(
58 request, {"message": "Object deleted", "object": deleted_object}
59 )
60
61 # If not found, return error
62 return JSONResponse(
63 request, {"message": "Object not found", "object": deleted_object}
64 )
65
66 # If we get here, something went wrong
67 return JSONResponse(request, {"message": "Something went wrong"})
68
69
70server.serve_forever(str(wifi.radio.ipv4_address))
Change NeoPixel color
There are several ways to pass data to the handler function:
In your handler function you can access the query/GET parameters using
request.query_paramsYou can also access the POST data directly using
request.bodyor if you data is in JSON format, you can userequest.json()to parse it into a dictionaryAlternatively for short pieces of data you can use URL parameters, which are described later in this document For more complex data, it is recommended to use JSON format.
All of these approaches allow you to pass data to the handler function and use it in your code.
For example by going to /change-neopixel-color?r=255&g=0&b=0 or /change-neopixel-color/255/0/0
you can change the color of the NeoPixel to red.
Tested on ESP32-S2 Feather.
1# SPDX-FileCopyrightText: 2022 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import board
6import neopixel
7import socketpool
8import wifi
9
10from adafruit_httpserver import Server, Route, as_route, Request, Response, GET, POST
11
12
13pool = socketpool.SocketPool(wifi.radio)
14server = Server(pool, "/static", debug=True)
15
16pixel = neopixel.NeoPixel(board.NEOPIXEL, 1)
17
18
19# This is the simplest way to register a route. It uses the Server object in current scope.
20@server.route("/change-neopixel-color", GET)
21def change_neopixel_color_handler_query_params(request: Request):
22 """Changes the color of the built-in NeoPixel using query/GET params."""
23
24 # e.g. /change-neopixel-color?r=255&g=0&b=0
25
26 r = request.query_params.get("r") or 0
27 g = request.query_params.get("g") or 0
28 b = request.query_params.get("b") or 0
29
30 pixel.fill((int(r), int(g), int(b)))
31
32 return Response(request, f"Changed NeoPixel to color ({r}, {g}, {b})")
33
34
35# This is another way to register a route. It uses the decorator that converts the function into
36# a Route object that can be imported and registered later.
37@as_route("/change-neopixel-color/form-data", POST)
38def change_neopixel_color_handler_post_form_data(request: Request):
39 """Changes the color of the built-in NeoPixel using POST form data."""
40
41 data = request.form_data # e.g. r=255&g=0&b=0 or r=255\r\nb=0\r\ng=0
42 r, g, b = data.get("r", 0), data.get("g", 0), data.get("b", 0)
43
44 pixel.fill((int(r), int(g), int(b)))
45
46 return Response(request, f"Changed NeoPixel to color ({r}, {g}, {b})")
47
48
49def change_neopixel_color_handler_post_json(request: Request):
50 """Changes the color of the built-in NeoPixel using JSON POST body."""
51
52 data = request.json() # e.g {"r": 255, "g": 0, "b": 0}
53 r, g, b = data.get("r", 0), data.get("g", 0), data.get("b", 0)
54
55 pixel.fill((r, g, b))
56
57 return Response(request, f"Changed NeoPixel to color ({r}, {g}, {b})")
58
59
60# You can always manually create a Route object and import or register it later.
61# Using this approach you can also use the same handler for multiple routes.
62post_json_route = Route(
63 "/change-neopixel-color/json", POST, change_neopixel_color_handler_post_json
64)
65
66
67def change_neopixel_color_handler_url_params(
68 request: Request, r: str = "0", g: str = "0", b: str = "0"
69):
70 """Changes the color of the built-in NeoPixel using URL params."""
71
72 # e.g. /change-neopixel-color/255/0/0
73
74 pixel.fill((int(r), int(g), int(b)))
75
76 return Response(request, f"Changed NeoPixel to color ({r}, {g}, {b})")
77
78
79# Registering Route objects
80server.add_routes(
81 [
82 change_neopixel_color_handler_post_form_data,
83 post_json_route,
84 # You can also register a inline created Route object
85 Route(
86 path="/change-neopixel-color/<r>/<g>/<b>",
87 methods=GET,
88 handler=change_neopixel_color_handler_url_params,
89 ),
90 ]
91)
92
93
94server.serve_forever(str(wifi.radio.ipv4_address))
Templates
With the help of the adafruit_templateengine library, it is possible to achieve somewhat of a
server-side rendering of HTML pages.
Instead of using string formatting, you can use templates, which can include more complex logic like loops and conditionals. This makes it very easy to create dynamic pages, witout using JavaScript and exposing any API endpoints.
Templates also allow splitting the code into multiple files, that can be reused in different places. You can find more information about the template syntax in the adafruit_templateengine documentation.
1{% exec path = context.get("path") %}
2{% exec items = context.get("items") %}
3
4<head>
5 <meta charset="UTF-8">
6 <title>Directory listing for /{{ path }}</title>
7</head>
8
9<body>
10 <h1>Directory listing for /{{ path }}</h1>
11
12 <input type="text" placeholder="Search...">
13
14 <ul>
15 {# Going to parent directory if not alredy in #}
16 {% if path %}
17 <li><a href="?path=/{{ "".join(path.split('/')[:-1]) }}">..</a></li>
18 {% endif %}
19
20 {# Listing items #}
21 {% for item in items %}
22 <li><a href="?path={{ f'/{path}/{item}' if path else f'/{item}' }}">{{ item }}</a></li>
23 {% endfor %}
24
25 </ul>
26
27 {# Script for filtering items #}
28 <script>
29 const search = document.querySelector('input');
30 const items = document.querySelectorAll('li');
31
32 search.addEventListener('keyup', (e) => {
33 const term = e.target.value.toLowerCase();
34
35 items.forEach(item => {
36 const text = item.innerText.toLowerCase();
37
38 if (text.indexOf(term) != -1) {
39 item.style.display = 'list-item';
40 } else {
41 item.style.display = 'none';
42 }
43 });
44 });
45 </script>
46</body>
47
48</html>
1# SPDX-FileCopyrightText: 2023 Michal Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4import os
5import re
6
7import socketpool
8import wifi
9
10from adafruit_httpserver import Server, Request, Response, FileResponse
11
12try:
13 from adafruit_templateengine import render_template
14except ImportError as e:
15 raise ImportError("This example requires adafruit_templateengine library.") from e
16
17
18pool = socketpool.SocketPool(wifi.radio)
19server = Server(pool, "/static", debug=True)
20
21# Create /static directory if it doesn't exist
22try:
23 os.listdir("/static")
24except OSError as e:
25 raise OSError("Please create a /static directory on the CIRCUITPY drive.") from e
26
27
28def is_file(path: str):
29 return (os.stat(path.rstrip("/"))[0] & 0b_11110000_00000000) == 0b_10000000_00000000
30
31
32@server.route("/")
33def directory_listing(request: Request):
34 path = request.query_params.get("path", "").replace("%20", " ")
35
36 # Preventing path traversal by removing all ../ from path
37 path = re.sub(r"\/(\.\.)\/|\/(\.\.)|(\.\.)\/", "/", path).strip("/")
38
39 # If path is a file, return it as a file response
40 if is_file(f"/static/{path}"):
41 return FileResponse(request, path)
42
43 items = sorted(
44 [
45 item + ("" if is_file(f"/static/{path}/{item}") else "/")
46 for item in os.listdir(f"/static/{path}")
47 ],
48 key=lambda item: not item.endswith("/"),
49 )
50
51 # Otherwise, return a directory listing
52 return Response(
53 request,
54 render_template(
55 "directory_listing.tpl.html",
56 context={"path": path, "items": items},
57 ),
58 content_type="text/html",
59 )
60
61
62# Start the server.
63server.serve_forever(str(wifi.radio.ipv4_address))
Form data parsing
Another way to pass data to the handler function is to use form data.
Remember that it is only possible to use it with POST method.
More about POST method.
It is important to use correct enctype, depending on the type of data you want to send.
application/x-www-form-urlencoded- For sending simple text data without any special characters including spaces.If you use it, values will be automatically parsed as strings, but special characters will be URL encoded e.g.
"Hello World! ^-$%"will be saved as"Hello+World%21+%5E-%24%25"
multipart/form-data- For sending text with special characters and filesWhen used, non-file values will be automatically parsed as strings and non plain text files will be saved as
bytes. e.g."Hello World! ^-$%"will be saved as'Hello World! ^-$%', and e.g. a PNG file will be saved asb'\x89PNG\r\n\x1a\n\x00\....
text/plain- For sending text data with special characters.If used, values will be automatically parsed as strings, including special characters, emojis etc. e.g.
"Hello World! ^-$%"will be saved as"Hello World! ^-$%", this is the recommended option.
If you pass multiple values with the same name, they will be saved as a list, that can be accessed using request.form_data.get_list().
Even if there is only one value, it will still get a list, and if there multiple values, but you use request.form_data.get() it will
return only the first one.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, Response, GET, POST
9
10
11pool = socketpool.SocketPool(wifi.radio)
12server = Server(pool, debug=True)
13
14
15FORM_HTML_TEMPLATE = """
16<html lang="en">
17 <head>
18 <title>Form with {enctype} enctype</title>
19 </head>
20 <body>
21 <a href="/form?enctype=application/x-www-form-urlencoded">
22 <button>Load <strong>application/x-www-form-urlencoded</strong> form</button>
23 </a><br />
24 <a href="/form?enctype=multipart/form-data">
25 <button>Load <strong>multipart/form-data</strong> form</button>
26 </a><br />
27 <a href="/form?enctype=text/plain">
28 <button>Load <strong>text/plain</strong> form</button>
29 </a><br />
30
31 <h2>Form with {enctype} enctype</h2>
32 <form action="/form" method="post" enctype="{enctype}">
33 <input type="text" name="something" placeholder="Type something...">
34 <input type="submit" value="Submit">
35 </form>
36 {submitted_value}
37 </body>
38</html>
39"""
40
41
42@server.route("/form", [GET, POST])
43def form(request: Request):
44 """
45 Serve a form with the given enctype, and display back the submitted value.
46 """
47 enctype = request.query_params.get("enctype", "text/plain")
48
49 if request.method == POST:
50 posted_value = request.form_data.get("something")
51
52 return Response(
53 request,
54 FORM_HTML_TEMPLATE.format(
55 enctype=enctype,
56 submitted_value=(
57 f"<h3>Enctype: {enctype}</h3>\n<h3>Submitted form value: {posted_value}</h3>"
58 if request.method == POST
59 else ""
60 ),
61 ),
62 content_type="text/html",
63 )
64
65
66server.serve_forever(str(wifi.radio.ipv4_address))
Chunked response
Library supports chunked responses. This is useful for streaming large amounts of data.
In order to use it, you need pass a generator that yields chunks of data to a ChunkedResponse
constructor.
1# SPDX-FileCopyrightText: 2022 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, ChunkedResponse
9
10
11pool = socketpool.SocketPool(wifi.radio)
12server = Server(pool, debug=True)
13
14
15@server.route("/chunked")
16def chunked(request: Request):
17 """
18 Return the response with ``Transfer-Encoding: chunked``.
19 """
20
21 def body():
22 yield "Adaf"
23 yield b"ruit" # Data chunk can be bytes or str.
24 yield " Indus"
25 yield b"tr"
26 yield "ies"
27
28 return ChunkedResponse(request, body)
29
30
31server.serve_forever(str(wifi.radio.ipv4_address))
URL parameters and wildcards
Alternatively to using query parameters, you can use URL parameters. They are a better choice when you want to perform different actions based on the URL. Query/GET parameters are better suited for modifying the behaviour of the handler function.
Of course it is only a suggestion, you can use them interchangeably and/or both at the same time.
In order to use URL parameters, you need to wrap them inside with angle brackets in Server.route, e.g. <my_parameter>.
All URL parameters values are passed as keyword arguments to the handler function.
Notice how the handler function in example below accepts two additional arguments : device_id and action.
If you specify multiple routes for single handler function and they have different number of URL parameters,
make sure to add default values for all the ones that might not be passed.
In the example below the second route has only one URL parameter, so the action parameter has a default value.
Keep in mind that URL parameters are always passed as strings, so you need to convert them to the desired type. Also note that the names of the function parameters have to match with the ones used in route, but they do not have to be in the same order.
Alternatively you can use e.g. **params to get all the parameters as a dictionary and access them using params['parameter_name'].
It is also possible to specify a wildcard route:
...- matches one path segment, e.g/api/...will match/api/123, but not/api/123/456....- matches multiple path segments, e.g/api/....will match/api/123and/api/123/456
In both cases, wildcards will not match empty path segment, so /api/.../users will match /api/v1/users, but not /api//users or /api/users.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, Response
9
10
11pool = socketpool.SocketPool(wifi.radio)
12server = Server(pool, debug=True)
13
14
15class Device:
16 def turn_on(self): # pylint: disable=no-self-use
17 print("Turning on device.")
18
19 def turn_off(self): # pylint: disable=no-self-use
20 print("Turning off device.")
21
22
23def get_device(device_id: str) -> Device: # pylint: disable=unused-argument
24 """
25 This is a **made up** function that returns a `Device` object.
26 """
27 return Device()
28
29
30@server.route("/device/<device_id>/action/<action>")
31@server.route("/device/emergency-power-off/<device_id>")
32def perform_action(
33 request: Request, device_id: str, action: str = "emergency_power_off"
34):
35 """
36 Performs an "action" on a specified device.
37 """
38
39 device = get_device(device_id)
40
41 if action in ["turn_on"]:
42 device.turn_on()
43 elif action in ["turn_off", "emergency_power_off"]:
44 device.turn_off()
45 else:
46 return Response(request, f"Unknown action ({action})")
47
48 return Response(
49 request, f"Action ({action}) performed on device with ID: {device_id}"
50 )
51
52
53@server.route("/device/<device_id>/status/<date>")
54def device_status_on_date(request: Request, **params: dict):
55 """
56 Return the status of a specified device between two dates.
57 """
58
59 device_id = params.get("device_id")
60 date = params.get("date")
61
62 return Response(request, f"Status of {device_id} on {date}: ...")
63
64
65@server.route("/device/.../status", append_slash=True)
66@server.route("/device/....", append_slash=True)
67def device_status(request: Request):
68 """
69 Returns the status of all devices no matter what their ID is.
70 Unknown commands also return the status of all devices.
71 """
72
73 return Response(request, "Status of all devices: ...")
74
75
76server.serve_forever(str(wifi.radio.ipv4_address))
Authentication
In order to increase security of your server, you can use Basic and Bearer authentication.
Remember that it is not a replacement for HTTPS, traffic is still sent in plain text, but it can be used to protect your server from unauthorized access.
If you want to apply authentication to the whole server, you need to call .require_authentication on Server instance.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, Response, Basic, Token, Bearer
9
10
11# Create a list of available authentication methods.
12auths = [
13 Basic("user", "password"),
14 Token("2db53340-4f9c-4f70-9037-d25bee77eca6"),
15 Bearer("642ec696-2a79-4d60-be3a-7c9a3164d766"),
16]
17
18pool = socketpool.SocketPool(wifi.radio)
19server = Server(pool, "/static", debug=True)
20server.require_authentication(auths)
21
22
23@server.route("/implicit-require")
24def implicit_require_authentication(request: Request):
25 """
26 Implicitly require authentication because of the server.require_authentication() call.
27 """
28
29 return Response(request, body="Authenticated", content_type="text/plain")
30
31
32server.serve_forever(str(wifi.radio.ipv4_address))
On the other hand, if you want to apply authentication to a set of routes, you need to call require_authentication function.
In both cases you can check if request is authenticated by calling check_authentication on it.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, Response, UNAUTHORIZED_401
9from adafruit_httpserver.authentication import (
10 AuthenticationError,
11 Basic,
12 Token,
13 Bearer,
14 check_authentication,
15 require_authentication,
16)
17
18
19pool = socketpool.SocketPool(wifi.radio)
20server = Server(pool, debug=True)
21
22# Create a list of available authentication methods.
23auths = [
24 Basic("user", "password"),
25 Token("2db53340-4f9c-4f70-9037-d25bee77eca6"),
26 Bearer("642ec696-2a79-4d60-be3a-7c9a3164d766"),
27]
28
29
30@server.route("/check")
31def check_if_authenticated(request: Request):
32 """
33 Check if the request is authenticated and return a appropriate response.
34 """
35 is_authenticated = check_authentication(request, auths)
36
37 return Response(
38 request,
39 body="Authenticated" if is_authenticated else "Not authenticated",
40 content_type="text/plain",
41 )
42
43
44@server.route("/require-or-401")
45def require_authentication_or_401(request: Request):
46 """
47 Require authentication and return a default server 401 response if not authenticated.
48 """
49 require_authentication(request, auths)
50
51 return Response(request, body="Authenticated", content_type="text/plain")
52
53
54@server.route("/require-or-handle")
55def require_authentication_or_manually_handle(request: Request):
56 """
57 Require authentication and manually handle request if not authenticated.
58 """
59
60 try:
61 require_authentication(request, auths)
62
63 return Response(request, body="Authenticated", content_type="text/plain")
64
65 except AuthenticationError:
66 return Response(
67 request,
68 body="Not authenticated - Manually handled",
69 content_type="text/plain",
70 status=UNAUTHORIZED_401,
71 )
72
73
74server.serve_forever(str(wifi.radio.ipv4_address))
Redirects
Sometimes you might want to redirect the user to a different URL, either on the same server or on a different one.
You can do that by returning Redirect from your handler function.
You can specify wheter the redirect is permanent or temporary by passing permanent=... to Redirect.
If you need the redirect to preserve the original request method, you can set preserve_method=True.
Alternatively, you can pass a status object directly to Redirect constructor.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import (
9 Server,
10 Request,
11 Response,
12 Redirect,
13 POST,
14 NOT_FOUND_404,
15 MOVED_PERMANENTLY_301,
16)
17
18
19pool = socketpool.SocketPool(wifi.radio)
20server = Server(pool, debug=True)
21
22REDIRECTS = {
23 "google": "https://www.google.com",
24 "adafruit": "https://www.adafruit.com",
25 "circuitpython": "https://circuitpython.org",
26}
27
28
29@server.route("/blinka")
30def redirect_blinka(request: Request):
31 """Always redirect to a Blinka page as permanent redirect."""
32 return Redirect(request, "https://circuitpython.org/blinka", permanent=True)
33
34
35@server.route("/adafruit")
36def redirect_adafruit(request: Request):
37 """Permanent redirect to Adafruit website with explicitly set status code."""
38 return Redirect(request, "https://www.adafruit.com/", status=MOVED_PERMANENTLY_301)
39
40
41@server.route("/fake-login", POST)
42def fake_login(request: Request):
43 """Fake login page."""
44 return Response(request, "Fake login page with POST data preserved.")
45
46
47@server.route("/login", POST)
48def temporary_login_redirect(request: Request):
49 """Temporary moved login page with preserved POST data."""
50 return Redirect(request, "/fake-login", preserve_method=True)
51
52
53@server.route("/<slug>")
54def redirect_other(request: Request, slug: str = None):
55 """
56 Redirect to a URL based on the slug.
57 """
58
59 if slug is None or slug not in REDIRECTS:
60 return Response(request, "Unknown redirect", status=NOT_FOUND_404)
61
62 return Redirect(request, REDIRECTS.get(slug))
63
64
65server.serve_forever(str(wifi.radio.ipv4_address))
Server-Sent Events
All types of responses until now were synchronous, meaning that the response was sent immediately after the handler function returned. However, sometimes you might want to send data to the client at a later time, e.g. when some event occurs. This can be overcomed by periodically polling the server, but it is not an elegant solution. Instead, you can use Server-Sent Events (SSE).
Response is initialized on return, events can be sent using .send_event() method. Due to the nature of SSE, it is necessary to store the
response object somewhere, so that it can be accessed later.
Warning
Because of the limited number of concurrently open sockets, it is not possible to process more than one SSE response at the same time. This might change in the future, but for now, it is recommended to use SSE only with one client at a time.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5from time import monotonic
6import microcontroller
7import socketpool
8import wifi
9
10from adafruit_httpserver import Server, Request, Response, SSEResponse, GET
11
12
13pool = socketpool.SocketPool(wifi.radio)
14server = Server(pool, debug=True)
15
16
17sse_response: SSEResponse = None
18next_event_time = monotonic()
19
20HTML_TEMPLATE = """
21<html lang="en">
22 <head>
23 <title>Server-Sent Events Client</title>
24 </head>
25 <body>
26 <p>CPU temperature: <strong>-</strong>°C</p>
27 <script>
28 const eventSource = new EventSource('/connect-client');
29 const cpuTemp = document.querySelector('strong');
30
31 eventSource.onmessage = event => cpuTemp.textContent = event.data;
32 eventSource.onerror = error => cpuTemp.textContent = error;
33 </script>
34 </body>
35</html>
36"""
37
38
39@server.route("/client", GET)
40def client(request: Request):
41 return Response(request, HTML_TEMPLATE, content_type="text/html")
42
43
44@server.route("/connect-client", GET)
45def connect_client(request: Request):
46 global sse_response # pylint: disable=global-statement
47
48 if sse_response is not None:
49 sse_response.close() # Close any existing connection
50
51 sse_response = SSEResponse(request)
52
53 return sse_response
54
55
56server.start(str(wifi.radio.ipv4_address))
57while True:
58 server.poll()
59
60 # Send an event every second
61 if sse_response is not None and next_event_time < monotonic():
62 cpu_temp = round(microcontroller.cpu.temperature, 2)
63 sse_response.send_event(str(cpu_temp))
64 next_event_time = monotonic() + 1
Websockets
Although SSE provide a simple way to send data from the server to the client, they are not suitable for sending data the other way around.
For that purpose, you can use Websockets. They are more complex than SSE, but they provide a persistent two-way communication channel between the client and the server.
Remember, that because Websockets also receive data, you have to explicitly call .receive() on the Websocket object to get the message.
This is anologous to calling .poll() on the Server object.
The following example uses asyncio, which has to be installed separately. It is not necessary to use asyncio to use Websockets,
but it is recommended as it makes it easier to handle multiple tasks. It can be used in any of the examples, but here it is particularly useful.
Warning
Because of the limited number of concurrently open sockets, it is not possible to process more than one Websocket response at the same time. This might change in the future, but for now, it is recommended to use Websocket only with one client at a time.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5from asyncio import create_task, gather, run, sleep as async_sleep
6import board
7import microcontroller
8import neopixel
9import socketpool
10import wifi
11
12from adafruit_httpserver import Server, Request, Response, Websocket, GET
13
14
15pool = socketpool.SocketPool(wifi.radio)
16server = Server(pool, debug=True)
17
18pixel = neopixel.NeoPixel(board.NEOPIXEL, 1)
19
20websocket: Websocket = None
21
22HTML_TEMPLATE = """
23<html lang="en">
24 <head>
25 <title>Websocket Client</title>
26 </head>
27 <body>
28 <p>CPU temperature: <strong>-</strong>°C</p>
29 <p>NeoPixel Color: <input type="color"></p>
30 <script>
31 const cpuTemp = document.querySelector('strong');
32 const colorPicker = document.querySelector('input[type="color"]');
33
34 let ws = new WebSocket('ws://' + location.host + '/connect-websocket');
35
36 ws.onopen = () => console.log('WebSocket connection opened');
37 ws.onclose = () => console.log('WebSocket connection closed');
38 ws.onmessage = event => cpuTemp.textContent = event.data;
39 ws.onerror = error => cpuTemp.textContent = error;
40
41 colorPicker.oninput = debounce(() => ws.send(colorPicker.value), 200);
42
43 function debounce(callback, delay = 1000) {
44 let timeout
45 return (...args) => {
46 clearTimeout(timeout)
47 timeout = setTimeout(() => {
48 callback(...args)
49 }, delay)
50 }
51 }
52 </script>
53 </body>
54</html>
55"""
56
57
58@server.route("/client", GET)
59def client(request: Request):
60 return Response(request, HTML_TEMPLATE, content_type="text/html")
61
62
63@server.route("/connect-websocket", GET)
64def connect_client(request: Request):
65 global websocket # pylint: disable=global-statement
66
67 if websocket is not None:
68 websocket.close() # Close any existing connection
69
70 websocket = Websocket(request)
71
72 return websocket
73
74
75server.start(str(wifi.radio.ipv4_address))
76
77
78async def handle_http_requests():
79 while True:
80 server.poll()
81
82 await async_sleep(0)
83
84
85async def handle_websocket_requests():
86 while True:
87 if websocket is not None:
88 if (data := websocket.receive(fail_silently=True)) is not None:
89 r, g, b = int(data[1:3], 16), int(data[3:5], 16), int(data[5:7], 16)
90 pixel.fill((r, g, b))
91
92 await async_sleep(0)
93
94
95async def send_websocket_messages():
96 while True:
97 if websocket is not None:
98 cpu_temp = round(microcontroller.cpu.temperature, 2)
99 websocket.send_message(str(cpu_temp), fail_silently=True)
100
101 await async_sleep(1)
102
103
104async def main():
105 await gather(
106 create_task(handle_http_requests()),
107 create_task(handle_websocket_requests()),
108 create_task(send_websocket_messages()),
109 )
110
111
112run(main())
Custom response types e.g. video streaming
The built-in response types may not always meet your specific requirements. In such cases, you can define custom response types and implement the necessary logic.
The example below demonstrates a XMixedReplaceResponse class, which uses the multipart/x-mixed-replace content type to stream video frames
from a camera, similar to a CCTV system.
To ensure the server remains responsive, a global list of open connections is maintained. By running tasks asynchronously, the server can stream video to multiple clients while simultaneously handling other requests.
1# SPDX-FileCopyrightText: 2024 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5try:
6 from typing import Dict, List, Tuple, Union
7except ImportError:
8 pass
9
10from asyncio import create_task, gather, run, sleep
11from random import choice
12
13import socketpool
14import wifi
15
16from adafruit_pycamera import PyCamera
17from adafruit_httpserver import Server, Request, Response, Headers, Status, OK_200
18
19
20pool = socketpool.SocketPool(wifi.radio)
21server = Server(pool, debug=True)
22
23
24camera = PyCamera()
25camera.display.brightness = 0
26camera.mode = 0 # JPEG, required for `capture_into_jpeg()`
27camera.resolution = "1280x720"
28camera.effect = 0 # No effect
29
30
31class XMixedReplaceResponse(Response):
32 def __init__(
33 self,
34 request: Request,
35 frame_content_type: str,
36 *,
37 status: Union[Status, Tuple[int, str]] = OK_200,
38 headers: Union[Headers, Dict[str, str]] = None,
39 cookies: Dict[str, str] = None,
40 ) -> None:
41 super().__init__(
42 request=request,
43 headers=headers,
44 cookies=cookies,
45 status=status,
46 )
47 self._boundary = self._get_random_boundary()
48 self._headers.setdefault(
49 "Content-Type", f"multipart/x-mixed-replace; boundary={self._boundary}"
50 )
51 self._frame_content_type = frame_content_type
52
53 @staticmethod
54 def _get_random_boundary() -> str:
55 symbols = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
56 return "--" + "".join([choice(symbols) for _ in range(16)])
57
58 def send_frame(self, frame: Union[str, bytes] = "") -> None:
59 encoded_frame = bytes(
60 frame.encode("utf-8") if isinstance(frame, str) else frame
61 )
62
63 self._send_bytes(
64 self._request.connection, bytes(f"{self._boundary}\r\n", "utf-8")
65 )
66 self._send_bytes(
67 self._request.connection,
68 bytes(f"Content-Type: {self._frame_content_type}\r\n\r\n", "utf-8"),
69 )
70 self._send_bytes(self._request.connection, encoded_frame)
71 self._send_bytes(self._request.connection, bytes("\r\n", "utf-8"))
72
73 def _send(self) -> None:
74 self._send_headers()
75
76 def close(self) -> None:
77 self._close_connection()
78
79
80stream_connections: List[XMixedReplaceResponse] = []
81
82
83@server.route("/frame")
84def frame_handler(request: Request):
85 frame = camera.capture_into_jpeg()
86
87 return Response(request, body=frame, content_type="image/jpeg")
88
89
90@server.route("/stream")
91def stream_handler(request: Request):
92 response = XMixedReplaceResponse(request, frame_content_type="image/jpeg")
93 stream_connections.append(response)
94
95 return response
96
97
98async def send_stream_frames():
99 while True:
100 await sleep(0.1)
101
102 frame = camera.capture_into_jpeg()
103
104 for connection in iter(stream_connections):
105 try:
106 connection.send_frame(frame)
107 except BrokenPipeError:
108 connection.close()
109 stream_connections.remove(connection)
110
111
112async def handle_http_requests():
113 server.start(str(wifi.radio.ipv4_address))
114
115 while True:
116 await sleep(0)
117
118 server.poll()
119
120
121async def main():
122 await gather(
123 create_task(send_stream_frames()),
124 create_task(handle_http_requests()),
125 )
126
127
128run(main())
HTTPS
Warning
HTTPS on CircuitPython works only on boards with enough memory e.g. ESP32-S3.
When you want to expose your server to the internet or an untrusted network, it is recommended to use HTTPS. Together with authentication, it provides a relatively secure way to communicate with the server.
Note
Using HTTPS slows down the server, because of additional work with encryption and decryption.
Enabling HTTPS is straightforward and comes down to passing the path to the certificate and key files to the Server constructor
and setting https=True.
1# SPDX-FileCopyrightText: 2024 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, Response
9
10
11pool = socketpool.SocketPool(wifi.radio)
12server = Server(
13 pool,
14 root_path="/static",
15 https=True,
16 certfile="cert.pem",
17 keyfile="key.pem",
18 debug=True,
19)
20
21
22@server.route("/")
23def base(request: Request):
24 """
25 Serve a default static plain text message.
26 """
27 return Response(request, "Hello from the CircuitPython HTTPS Server!")
28
29
30server.serve_forever(str(wifi.radio.ipv4_address), 443)
To create your own certificate, you can use the following command:
sudo openssl req -x509 -nodes -days 365 -newkey rsa:2048 -keyout key.pem -out cert.pem
You might have to change permissions of the files, so that the server can read them.
Multiple servers
Although it is not the primary use case, it is possible to run multiple servers at the same time.
In order to do that, you need to create multiple Server instances and call .start() and .poll() on each of them.
Using .serve_forever() for this is not possible because of it’s blocking behaviour.
Each server must have a different port number.
To distinguish between responses from different servers a ‘X-Server’ header is added to each response. This is an optional step, both servers will work without it.
In combination with separate authentication and diffrent root_path this allows creating moderately complex setups.
You can share same handler functions between servers or use different ones for each server.
1# SPDX-FileCopyrightText: 2023 Michał Pokusa
2#
3# SPDX-License-Identifier: Unlicense
4
5import socketpool
6import wifi
7
8from adafruit_httpserver import Server, Request, Response
9
10
11pool = socketpool.SocketPool(wifi.radio)
12
13bedroom_server = Server(pool, "/bedroom", debug=True)
14bedroom_server.headers["X-Server"] = "Bedroom"
15
16office_server = Server(pool, "/office", debug=True)
17office_server.headers["X-Server"] = "Office"
18
19
20@bedroom_server.route("/bedroom")
21def bedroom(request: Request):
22 """
23 This route is registered only on ``bedroom_server``.
24 """
25 return Response(request, "Hello from the bedroom!")
26
27
28@office_server.route("/office")
29def office(request: Request):
30 """
31 This route is registered only on ``office_server``.
32 """
33 return Response(request, "Hello from the office!")
34
35
36@bedroom_server.route("/home")
37@office_server.route("/home")
38def home(request: Request):
39 """
40 This route is registered on both servers.
41 """
42 return Response(request, "Hello from home!")
43
44
45ip_address = str(wifi.radio.ipv4_address)
46
47# Start the servers.
48bedroom_server.start(ip_address, 5000)
49office_server.start(ip_address, 8000)
50
51while True:
52 try:
53 # Process any waiting requests for both servers.
54 bedroom_server.poll()
55 office_server.poll()
56 except OSError as error:
57 print(error)
58 continue
Debug mode
It is highly recommended to disable debug mode in production.
During development it is useful to see the logs from the server.
You can enable debug mode by setting debug=True on Server instance or in constructor,
it is disabled by default.
Debug mode prints messages on server startup, after sending a response to a request and if exception
occurs during handling of the request in .serve_forever().
This is how the logs might look like when debug mode is enabled:
Started development server on http://192.168.0.100:5000
192.168.0.101 -- "GET /" 194 -- "200 OK" 154 -- 96ms
192.168.0.101 -- "GET /example" 134 -- "404 Not Found" 172 -- 123ms
192.168.0.102 -- "POST /api" 1241 -- "401 Unauthorized" 95 -- 64ms
Traceback (most recent call last):
...
File "code.py", line 55, in example_handler
KeyError: non_existent_key
192.168.0.103 -- "GET /index.html" 242 -- "200 OK" 154 -- 182ms
Stopped development server
This is the default format of the logs:
{client_ip} -- "{request_method} {path}" {request_size} -- "{response_status}" {response_size} -- {elapsed_ms}
If you need more information about the server or request, or you want it in a different format you can modify
functions at the bottom of adafruit_httpserver/server.py that start with _debug_....
Note
This is an advanced usage that might change in the future. It is not recommended to modify other parts of the code.