Trace CoAP methods and Observe on local UDP

Discover a local CoAP resource, read and change values, then correlate two Observe notifications by token and actual UDP bytes.

Broker Bex: I want you to match each resource action and notification to its wire evidence., your practice guide

Broker Bex: I want you to match each resource action and notification to its wire evidence.
Predict the reading, then compare it with the measurement.

Python 3 + aiocoap

Python

Discover a local CoAP resource, read and change values, then correlate two Observe notifications by token and actual UDP bytes.

Tier 2 · Python · install required · No account

Version tested: Python 3.9.21; aiocoap 0.4.7; local UDP; Xfce Terminal 1.0.4 on Xvfb. Date: 2026-10-08.

Download the three files, create a Python virtual environment, install requirements.txt, and run main.py.

Open the Python run guide (new tab)

Get the files

Download all three prepared files into one folder. No account or paid service is required.

main.py

6,183 bytes · Python program

Download

"""Real localhost CoAP GET, PUT and Observe traffic through a UDP byte tap."""
import argparse
import asyncio

import aiocoap
from aiocoap import resource

PROXY_PORT = 17680
SERVER_PORT = 17681

class ByteTap(asyncio.DatagramProtocol):
    def __init__(self):
        self.transport = None
        self.client = None
        self.frames = []

def connection_made(self, transport):
        self.transport = transport

def datagram_received(self, data, address):
        if address[1] == SERVER_PORT:
            self.frames.append(("S>C", data))
            if self.client:
                self.transport.sendto(data, self.client)
        else:
            self.client = address
            self.frames.append(("C>S", data))
            self.transport.sendto(data, ("127.0.0.1", SERVER_PORT))

def describe(frame):
    direction, data = frame
    token_len = data[0] & 15
    code = f"{data[1] >> 5}.{data[1] & 31:02d}"
    return (f"UDP {direction} {len(data)} bytes hex={data.hex()}\n"
            f"CoAP ver={data[0] >> 6} type={(data[0] >> 4) & 3} "
            f"code={code} mid=0x{int.from_bytes(data[2:4], 'big'):04x} "
            f"token={data[4:4 + token_len].hex()}")

class Discovery(resource.Resource):
    async def render_get(self, request):
        response = aiocoap.Message(payload=b'</temperature>;obs,</target>;rt="setpoint"')
        response.opt.content_format = 40
        return response

class Temperature(resource.ObservableResource):
    def __init__(self):
        super().__init__()
        self.celsius = 21.5

async def render_get(self, request):
        response = aiocoap.Message(payload=f"{self.celsius:.1f}".encode())
        response.opt.content_format = 0
        return response

class Target(resource.Resource):
    def __init__(self):
        super().__init__()
        self.celsius = 20.0

async def render_get(self, request):
        return aiocoap.Message(payload=f"{self.celsius:.1f}".encode())

async def render_put(self, request):
        try:
            value = float(request.payload.decode())
            if not 5 <= value <= 35:
                raise ValueError()
        except ValueError:
            return aiocoap.Message(code=aiocoap.BAD_REQUEST, payload=b"target must be 5..35 C")
        self.celsius = value
        return aiocoap.Message(code=aiocoap.CHANGED, payload=f"{value:.1f}".encode())

async def run(step):
    site = resource.Site()
    temperature = Temperature()
    site.add_resource(("sensors",), Discovery())
    site.add_resource(("temperature",), temperature)
    site.add_resource(("target",), Target())
    server = await aiocoap.Context.create_server_context(site, bind=("127.0.0.1", SERVER_PORT))
    loop = asyncio.get_running_loop()
    transport, tap = await loop.create_datagram_endpoint(ByteTap, local_addr=("127.0.0.1", PROXY_PORT))
    client = await aiocoap.Context.create_client_context()
    base = f"coap://127.0.0.1:{PROXY_PORT}"
    outputs = {}
    try:
        for number, method, path, payload in [
            (1, aiocoap.GET, "sensors", b""),
            (2, aiocoap.GET, "temperature", b""),
            (3, aiocoap.PUT, "target", b"22.0"),
            (4, aiocoap.GET, "target", b""),
        ]:
            start = len(tap.frames)
            request = aiocoap.Message(code=method, uri=f"{base}/{path}", payload=payload)
            response = await asyncio.wait_for(client.request(request).response, 3)
            await asyncio.sleep(.04)
            frames = tap.frames[start:]
            outputs[number] = [
                f"STEP {number}: {method} /{path}",
                *(describe(frame) for frame in frames[:2]),
                f"application response={response.code} payload={response.payload.decode()}",
                "Bytes above came from this run's UDP datagrams, not a packet example.",
            ]
        start = len(tap.frames)
        request = aiocoap.Message(code=aiocoap.GET, uri=f"{base}/temperature", observe=0)
        requester = client.request(request)
        first = await asyncio.wait_for(requester.response, 3)
        notifications = []
        pending = asyncio.Queue()
        requester.observation.register_callback(pending.put_nowait)
        for value in (22.0, 22.5):
            temperature.celsius = value
            temperature.updated_state()
            notifications.append(await asyncio.wait_for(pending.get(), 3))
        await asyncio.sleep(.04)
        observe_frames = tap.frames[start:]
        requester.observation.cancel()
        outputs[5] = [
            "STEP 5: GET /temperature Observe=0, then two server updates",
            *(describe(frame) for frame in observe_frames[:2]),
            f"initial response={first.payload.decode()} observe={first.opt.observe}",
            f"notification 1={notifications[0].payload.decode()} observe={notifications[0].opt.observe}",
            f"notification 2={notifications[1].payload.decode()} observe={notifications[1].opt.observe}",
            "Notifications above came from this run's CoAP server over local UDP.",
        ]
        response_frames = [frame for frame in observe_frames if frame[0] == "S>C"]
        tokens = [frame[1][4:4 + (frame[1][0] & 15)].hex() for frame in response_frames]
        outputs[6] = [
            "STEP 6: correlate the Observe token and UDP frames",
            f"server datagrams={len(response_frames)}",
            *(f"response {n}: token={token} code={response_frames[n - 1][1][1] >> 5}.{response_frames[n - 1][1][1] & 31:02d} hex={response_frames[n - 1][1].hex()}" for n, token in enumerate(tokens, 1)),
            f"same token across initial response and updates={len(set(tokens)) == 1}",
            "Hex rows are original aiocoap UDP datagrams relayed by the byte tap.",
        ]
        for number in ([step] if step else sorted(outputs)):
            print("\n".join(outputs[number]), flush=True)
    finally:
        transport.close()
        await client.shutdown()
        await server.shutdown()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--step", type=int, choices=range(1, 7))
    args = parser.parse_args()
    asyncio.run(run(args.step))

requirements.txt

15 bytes · Pinned dependencies

Download

README.md

1,107 bytes · Run guide

Download

# CoAP methods and Observe on local UDP

Use Python 3.9 or newer. In this folder:

```bash
python3 -m venv .venv
. .venv/bin/activate
python -m pip install -r requirements.txt
python main.py
python main.py --step 5
```

Steps 1–4 run discovery, GET and PUT, then GET the updated target. Step 5 subscribes to the observable temperature and triggers two changes. Step 6 checks that notifications retain the CoAP token. Each `--step` reruns the whole short scenario to start from a known state, then prints only that step. The program runs a real aiocoap server, client and UDP relay on 127.0.0.1 ports 17681 and 17680. The relay records the actual datagram bytes; the values are fictional. No radio link, cloud service or public endpoint is used. Stop another service using these ports before starting the lab.

After the full run, choose an investigation: compare the GET and PUT method codes in steps 2–4, or follow the Observe token and notification sequence in steps 5–6. The aiocoap client/server show application behavior; the standard-library UDP relay shows the corresponding transmitted bytes.

Put main.py, requirements.txt, and README.md in the same folder.
Create and activate a Python virtual environment, then install requirements.txt.
Run python3 main.py and compare its output with each observed result below.

Steps

Screens captured against Python 3 + aiocoap Python 3.9.21; aiocoap 0.4.7; local UDP; Xfce Terminal 1.0.4 on Xvfb on 2026-10-08; the tool may have moved on — the text steps are the contract.

Step 1
Do
In the terminal window panel, run `python3 main.py --step 1` to GET `/sensors` and inspect the returned resource links.
You will see
The local UDP trace contains a 2.05 Content response with `</temperature>;obs,</target>;rt="setpoint"`.
Why it matters
Discovery names the observable reading and the writable target before clients use either resource.
Step 2
Do
In the terminal window panel, run `python3 main.py --step 2` to GET `/temperature`; compare the datagram hex, CoAP code, and decoded reading.
You will see
The actual response has `2.05 Content` and payload `21.5`; the output prints request and response datagram bytes.
Why it matters
The resource value can be tied to a specific UDP request and response rather than inferred from a page example.
Step 3
Do
In the terminal window panel, run `python3 main.py --step 3` to PUT `22.0` to `/target`; inspect the response code.
You will see
The PUT payload is `22.0` and the application response is `2.04 Changed` with payload `22.0`.
Why it matters
A change operation has a different method and success code from a read.
Step 4
Do
In the terminal window panel, run `python3 main.py --step 4` to GET `/target` after the PUT.
You will see
The follow-up GET returns `2.05 Content` with payload `22.0`.
Why it matters
A second read confirms that this run changed the target state.
Step 5
Do
In the terminal window panel, run `python3 main.py --step 5` to register Observe and trigger two local temperature updates.
You will see
The initial response is `21.5 observe=0`; notifications report `22.0 observe=1` and `22.5 observe=2`.
Why it matters
The changing Observe values and payloads show server-initiated updates after one registration.
Step 6
Do
In the terminal window panel, run `python3 main.py --step 6` and compare the response token in all three UDP datagrams.
You will see
Three server datagrams share one response token; `same token across initial response and updates=True`.
Why it matters
The token links the initial response and subsequent notifications to one observer request.

Chapter checks

These questions refer to the chapter’s examples. Use the return links to review their answers.

A CoAP Observe design replaces polling for a fast-changing sensor. What evidence should be reviewed before accepting it?
Return to the chapter’s knowledge check
A client sends GET /temperature with Observe: 0 and Token: 0xAB12. The server responds with Observe: 47. What does the value 47 in the server's response represent?
Return to the chapter’s knowledge check

Return to CoAP Observe: Subscriptions and Updates · Browse Labs

Python 3 + aiocoap

Get the files

main.py

requirements.txt

README.md

Steps

1 Step 1

2 Step 2

3 Step 3

4 Step 4

5 Step 5

6 Step 6

Chapter checks

Step 1

Step 2

Step 3

Step 4

Step 5

Step 6