CoAP Protocol — RESTful Communication for Constrained IoT Devices Explained

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