CoAP Protocol — RESTful Communication for Constrained IoT Devices Explained

In this tutorial, you'll learn about CoAP Protocol. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

CoAP (Constrained Application Protocol) is a specialized UDP-based protocol designed for IoT devices with limited memory, processing power, and battery, implementing a RESTful request-response model similar to HTTP.

Why CoAP Matters

MQTT dominates the pub/sub space, but not every IoT use case fits the pub/sub pattern. Sometimes you need to read from or write to a specific resource on a device — get the current temperature, turn on a specific light, update a configuration value. This is the RESTful request-response model, and CoAP brings it to constrained devices with 10KB of RAM and 8-bit microcontrollers. CoAP runs over UDP, avoiding TCP's connection overhead, and uses binary encoding that's far more compact than HTTP's text headers.

Plain-Language Explanation

Imagine HTTP as sending a formal letter. You write the address, a greeting, the message, a closing, and your signature. The recipient acknowledges receipt and sends a reply with their own formal letter. It's reliable but verbose.

CoAP is like sending a text message. "Temp?" and the reply is "23.5". Short, fast, and if the reply doesn't arrive, you send "Temp?" again. Same intent as HTTP — read a resource (GET /.well-known/core), create a resource (POST /sensors), update a resource (PUT /actuators/light), delete a resource (DELETE /actuators/light). But the overhead is dramatically smaller.

graph LR
    Client[CoAP Client
Dashboard] -->|CON GET /temperature| Server[CoAP Server
ESP8266 Sensor]
    Server -->|ACK 2.05 Content "23.5"| Client
    Client -->|CON PUT /actuators/light| Server
    Server -->|ACK 2.04 Changed| Client
    subgraph "CoAP Message Types"
        CON[Confirmable]
        NON[Non-Confirmable]
        ACK[Acknowledgment]
        RST[Reset]
    end
    style Server fill:#3498db,color:#fff
    style Client fill:#27ae60,color:#fff

CoAP vs HTTP Mapping

Resource Discovery

CoAP devices advertise their resources through /.well-known/core, returning a link format listing available endpoints:

# CoAP resource discovery response
# Request:  GET coap://sensor.local/.well-known/core
# Response: </temperature>;rt="temperature-celsius";if="sensor",
#           </humidity>;rt="percent";if="sensor",
#           </actuators/light>;rt="light-switch";if="actuator"

Python CoAP Server Example

# coap_server.py
from aiocoap import *
import asyncio, random, json

class TemperatureResource(Resource):
    async def render_get(self, request):
        temp = round(random.uniform(18.0, 30.0), 1)
        payload = json.dumps({"temperature": temp, "unit": "C"}).encode()
        return Message(payload=payload, code=Code.CONTENT)

    async def render_put(self, request):
        # Update configuration (threshold, etc.)
        print(f"Received config update: {request.payload}")
        return Message(code=Code.CHANGED)

class LightResource(Resource):
    def __init__(self):
        super().__init__()
        self.state = False  # Off

    async def render_get(self, request):
        payload = json.dumps({"state": "on" if self.state else "off"}).encode()
        return Message(payload=payload, code=Code.CONTENT)

    async def render_put(self, request):
        data = json.loads(request.payload)
        self.state = data.get("state", False)
        print(f"Light turned {'ON' if self.state else 'OFF'}")
        return Message(code=Code.CHANGED)

async def main():
    root = Site()
    root.add_resource(['temperature'], TemperatureResource())
    root.add_resource(['actuators', 'light'], LightResource())

    await Context.create_server_context(root, bind=('0.0.0.0', 5683))
    print("CoAP server running on coap://0.0.0.0:5683")
    print("Resources:")
    print("  GET /temperature")
    print("  PUT /actuators/light")
    await asyncio.get_event_loop().run_forever()

if __name__ == "__main__":
    asyncio.run(main())

# coap_client.py
from aiocoap import *
import asyncio, json

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

    # GET temperature
    request = Message(code=GET, uri='coap://localhost:5683/temperature')
    response = await protocol.request(request).response
    data = json.loads(response.payload)
    print(f"Temperature: {data['temperature']}°C")

    # PUT light on
    payload = json.dumps({"state": True}).encode()
    request = Message(code=PUT, uri='coap://localhost:5683/actuators/light', payload=payload)
    response = await protocol.request(request).response
    print(f"Light control: {response.code}")

if __name__ == "__main__":
    asyncio.run(main())

Expected output:

Temperature: 24.7°C
Light control: 2.04 Changed

CoAP Observe (Subscription)

Instead of polling, a client can "observe" a resource and receive notifications when it changes:

# Observe temperature changes
request = Message(code=GET, uri='coap://localhost:5683/temperature')
observation = await protocol.request(request).observation

async for response in observation:
    data = json.loads(response.payload)
    print(f"Updated temperature: {data['temperature']}°C")
    # Exit after 5 updates

CoAP vs MQTT

When CoAP wins: You need to read/write specific resources on individual devices. You're building a device-to-device mesh. UDP-only networks (some LPWAN technologies).

When MQTT wins: You need pub/sub event streaming. Devices communicate through a cloud broker. You need many-to-many communication patterns.

Common Mistakes

1. Treating CoAP exactly like HTTP: CoAP runs over UDP, so packets can be lost. Use CON (confirmable) messages for important requests and implement retry logic.

2. Not implementing resource discovery: CoAP's discovery is one of its best features. Always expose /.well-known/core so clients can discover available resources.

3. Ignoring block-wise transfer: CoAP has a 64KB message limit. For larger payloads (firmware updates), use block-wise transfer (Block1 and Block2 options).

4. No security: DTLS (Datagram TLS) secures CoAP (CoAPS on port 5684). Without it, anyone on the network can read or write resources.

5. Polling instead of observing: Don't poll GET /temperature every second. Use the observe pattern to receive push updates.

Practice Questions

1. What transport protocol does CoAP use and why? UDP. TCP's connection overhead and three-way handshake are too expensive for constrained devices. UDP is lightweight but requires CoAP's CON/NON reliability mechanism.

2. How does CoAP achieve reliability over UDP? CON (confirmable) messages require an ACK (acknowledgment). If no ACK arrives, the sender retransmits with exponential backoff. NON (non-confirmable) messages have no acknowledgment.

3. What is the difference between CoAP observe and MQTT subscribe? Both provide push updates, but CoAP observe is per-resource (client observes a specific URI) while MQTT subscribe uses topic hierarchies with wildcards.

4. How does CoAP resource discovery work? Clients send GET /.well-known/core and the server responds with a link format listing available resources, their types, and interfaces.

5. What is CoAP block-wise transfer? A mechanism to split large payloads into multiple blocks, each sent in a separate CoAP message. Required for firmware updates or large configuration files.

Mini Project

Build a CoAP-controlled LED simulator:

from aiocoap import *
import asyncio, json, time

class LEDResource(Resource):
    colors = {"red": (255,0,0), "green": (0,255,0), "blue": (0,0,255)}
    active_color = "off"

    async def render_get(self, request):
        return Message(
            payload=json.dumps({"state": self.active_color}).encode(),
            code=Code.CONTENT
        )

    async def render_put(self, request):
        data = json.loads(request.payload)
        color = data.get("color", "off")
        if color in self.colors or color == "off":
            self.active_color = color
            status = "on" if color != "off" else "off"
            print(f"LED: {color.upper()} ({status})")
            return Message(code=Code.CHANGED)
        return Message(code=Code.BAD_REQUEST, payload=b"Invalid color")

async def main():
    root = Site()
    root.add_resource(['led'], LEDResource())
    await Context.create_server_context(root, bind=('0.0.0.0', 5683))
    print("LED simulator running. Try:")
    print("  coap://localhost:5683/led")
    await asyncio.get_event_loop().run_forever()

asyncio.run(main())

Cross-References

• MQTT
• IoT Overview
• ESP32
• IoT Security
• Industrial IoT