Implementing CoAP Observe Option Using Python

In the last tutorial, we started implementing a CoAP server that allows a client to update the status of an alarm resource. In this tutorial we will use the CoAP Observe option to extend the server so that other clients can register to be notified when the resource changes.

The Observe option is an extension of the GET request. More precisely, it is an optional field in the GET request header. When a client queries the server with an Observe option, it basically asking for the current status of the alarm and also to be notified if it changes in the future.

Note that the server does not have to honour the observe request. For example, if a CoAP resource doesn’t support Observers or it has reached the maximum registered observers. In this case, the Observe option will just be ignored and the request will default to a plain GET request.

The server may also periodically send the current state of the resource to all registered observers. If it doesn’t hear anything back from any observer then that observer will be removed from the resource’s registered observers list. This is one mechanism the server uses to clean up the observers list in the event any client silently disappears.

The Implementation

Let’s carry on with

Instead of inheriting from Aiocoap Resource class, AlarmResource will now inherit from Aiocoap ObservableResource. This will manage the observers for us, we just need to handle what to send and when to send it.

We will update the handling of the PUT request so that when the status of the alarm is updated, it will set a flag to indicate the server to notify observers.

We will add notify_observers_check() method which continuously loop and check to see if the notify_observers flag is set or not. If it is set then the server will send an update to each observer by calling render_get().


import aiocoap.resource as resource
import aiocoap
import threading
import logging
import asyncio

class AlarmResource(resource.ObservableResource):
    """This resource supports the GET and PUT methods and is observable.
    GET: Return current state of alarm
    PUT: Update state of alarm and notify registered observers"""

    def __init__(self):

        self.status = "OFF"
        self.has_observers = False
        self.notify_observers = False

    # Ensure observers are notify if required
    def notify_observers_check(self):
        while True:
            if self.has_observers and self.notify_observers:
                print('Notifying observers')
                self.notify_observers = False

    # Observers change event callback
    def update_observation_count(self, count):
        if count:
            self.has_observers = True
            self.has_observers = False

    # Handles GET request or observer notify
    async def render_get(self, request):
        print('Return alarm state: %s' % self.status)
        payload = b'%s' % self.status.encode('ascii')

        return aiocoap.Message(payload=payload)

    # Handles PUT request
    async def render_put(self, request):
        self.status = request.payload.decode('ascii')
        print('Update alarm state: %s' % self.status)
        self.notify_observers = True

        return aiocoap.Message(code=aiocoap.CHANGED, payload=b'%s' % self.status.encode('ascii'))

In the main(), we updated it so that it will spawn a separate daemon thread with the sole purpose is to call notify_observers_check() to handle the notify_observers flag event.


def main():
    # Resource tree creation
    root = resource.Site()
    alarmResource = AlarmResource()
    root.add_resource(['alarm'], alarmResource)
    asyncio.Task(aiocoap.Context.create_server_context(root, bind=('localhost', 5683)))

    # Spawn a daemon to notify observers when alarm status changes
    observers_notifier = threading.Thread(target=alarmResource.notify_observers_check)
    observers_notifier.daemon = True


if __name__ == "__main__":

To test the Observe option we will create another client that will start observing the alarm status. This observe client will also use the Aiocoap python library. Whenever a notification is received, observe_callback() is called.


import logging
import asyncio

from aiocoap import *


def observe_callback(response):
    if response.code.is_successful():
        print("Alarm status: %s" % (response.payload.decode('ascii')))
        print('Error code %s' % response.code)

async def main():
    context = await Context.create_client_context()

    request = Message(code=GET)
    request.opt.observe = 0
    observation_is_over = asyncio.Future()

        context_request = context.request(request)
        response = await context_request.response
        exit_reason = await observation_is_over
        print('Observation is over: %r' % exit_reason)
        if not context_request.response.done():
        if not context_request.observation.cancelled:

if __name__ == "__main__":

Below is a demo showing a client (left) which randomly set the alarm to “ON” or “OFF” each time it is called. The server in the middle and another client (right) which has been registered with the server to get notified.

Categories: Internet of things, PythonTags: , , ,

Leave a Reply

Fill in your details below or click an icon to log in: Logo

You are commenting using your account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s

%d bloggers like this: