0Pricing
AI Agents · Lezione

Protocollo MQTT per l’integrazione degli agenti

Configurazione del broker, sottoscrizione ai topic e attivazione degli agenti guidata dai messaggi.

Protocollo MQTT per l’integrazione degli agenti è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 1 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.

MQTT e agenti IoT

MQTT (Message Queuing Telemetry Transport) è un protocollo leggero publish-subscribe progettato per gli ambienti IoT con larghezza di banda ridotta e latenza elevata. Gli agenti possono sottoscriversi ai topic MQTT per ricevere dati dei sensori in tempo reale e pubblicare comandi sui dispositivi.

Installazione di paho-mqtt

La libreria paho-mqtt è il client MQTT standard per Python. La installi con pip. Supporta MQTT 3.1.1 e 5.0, la crittografia TLS e tutti e tre i livelli QoS.

# pip install paho-mqtt

import paho.mqtt.client as mqtt

# Create a client instance
client = mqtt.Client(client_id='agent_001', protocol=mqtt.MQTTv311)

# Optional: set credentials if broker requires authentication
client.username_pw_set(username='agent_user', password='YOUR_PASSWORD')

# Optional: enable TLS for secure connections
# client.tls_set('/path/to/ca.crt')

print('MQTT client created:', client._client_id)

Connessione al broker

Si connetta a un broker MQTT (ad es. Mosquitto, HiveMQ, EMQX o un broker cloud). La chiamata connect non è bloccante; utilizzi loop_start() per eseguire il ciclo di rete in un thread in background.

import paho.mqtt.client as mqtt
import time

BROKER_HOST = 'broker.hivemq.com'  # public test broker
BROKER_PORT = 1883
KEEP_ALIVE_SECONDS = 60

def on_connect(client, userdata, flags, rc):
    status = {
        0: 'Connected successfully',
        1: 'Refused: wrong protocol',
        2: 'Refused: client ID rejected',
        3: 'Refused: server unavailable',
        4: 'Refused: bad credentials',
        5: 'Refused: not authorised'
    }
    print(f'Connect result: {status.get(rc, f"Unknown code {rc}")}')

client = mqtt.Client(client_id='agent_001')
client.on_connect = on_connect
client.connect(BROKER_HOST, BROKER_PORT, keepalive=KEEP_ALIVE_SECONDS)
client.loop_start()  # background thread
time.sleep(1)  # wait for connection

Sottoscrizione a un topic

I topic sono stringhe gerarchiche come sensors/temperature o factory/line1/pressure. Utilizzi # come carattere jolly per tutti i sotto-topic oppure + come carattere jolly per un singolo livello. Colleghi un callback on_message per elaborare i messaggi in arrivo.

import json

class FakeMsg:
    def __init__(self, topic, payload):
        self.topic = topic
        self.payload = payload

class FakeMQTTClient:
    def __init__(self):
        self.on_message = None
        self.userdata = {}
    def user_data_set(self, userdata):
        self.userdata = userdata
    def subscribe(self, topic, qos=0):
        print(f'Subscribed to {topic} (qos={qos})')
    def simulate_message(self, topic, payload_bytes):
        self.on_message(self, self.userdata, FakeMsg(topic, payload_bytes))

class Agent:
    def process_sensor_data(self, topic, payload):
        print(f'Agent processing {topic} -> {payload}')

def on_message(client, userdata, msg):
    topic = msg.topic
    try:
        payload = json.loads(msg.payload.decode('utf-8'))
    except (json.JSONDecodeError, UnicodeDecodeError):
        payload = msg.payload.decode('utf-8', errors='replace')

    print(f'Received on {topic}: {payload}')
    userdata['agent'].process_sensor_data(topic, payload)

agent_state = {'agent': Agent()}
client = FakeMQTTClient()
client.user_data_set(agent_state)
client.on_message = on_message

client.subscribe('sensors/temperature', qos=1)
client.subscribe('sensors/+/humidity', qos=1)
client.subscribe('factory/#', qos=1)

client.simulate_message('sensors/temperature', b'{"value": 21.5}')

Livelli QoS spiegati

MQTT dispone di tre livelli di Quality of Service: QoS 0 — invia e dimentica (il più veloce, ma i messaggi possono andare persi), QoS 1 — almeno una volta (messaggio confermato, ma può essere duplicato), QoS 2 — esattamente una volta (garantito, ma il più lento). Scelga in base alla tolleranza per la perdita di dati rispetto alla latenza.

# QoS level guidelines for IoT agents:

# QoS 0 — temperature readings updated every second
# (loss of one reading is acceptable)
client.subscribe('sensors/temperature', qos=0)

# QoS 1 — alert notifications (must arrive, duplicates are OK)
client.subscribe('sensors/alerts', qos=1)

# QoS 2 — billing/counting events (each event must be processed exactly once)
client.subscribe('meters/energy_consumed', qos=2)

# When publishing, specify QoS:
client.publish(
    topic='agents/response',
    payload='{"action": "turn_on_cooling"}',
    qos=1,
    retain=False
)
print('Published command with QoS 1')

Messaggi retained

Un messaggio retained è l'ultimo valore pubblicato per un topic, che il broker memorizza e invia immediatamente a ogni nuovo sottoscrittore. È ideale per i topic relativi allo stato dei sensori: una nuova istanza dell'agente che entra nella rete conosce immediatamente il valore corrente del sensore, senza dover attendere il prossimo aggiornamento.

# Publishing with retain=True persists the last value on the broker
client.publish(
    topic='sensors/thermostat/current_temp',
    payload='{"value": 22.5, "unit": "celsius"}',
    qos=1,
    retain=True  # broker stores this message
)

# Any new subscriber will receive this immediately on subscribe,
# even if it was published hours ago.

# To clear a retained message, publish empty payload:
client.publish(
    topic='sensors/thermostat/current_temp',
    payload='',  # empty payload clears retention
    retain=True
)
print('Retained message cleared')

Creazione di un agente MQTT per sensori

Un agente per sensori si sottoscrive ai topic dei dati grezzi dei sensori, convalida i dati in arrivo e decide se attivare un'azione. La logica decisionale utilizza l'LLM solo per i casi complessi; i semplici controlli sulle soglie vengono gestiti direttamente in Python per ottenere maggiore velocità.

import anthropic

class SensorAgent:
    def __init__(self, mqtt_client, llm_api_key: str):
        self.client = mqtt_client
        self.llm = anthropic.Anthropic(api_key=llm_api_key)
        self.readings = []

    def process_sensor_data(self, topic: str, payload: dict):
        value = payload.get('value')
        if value is None:
            return
        self.readings.append({'topic': topic, 'value': value})
        # Fast path: simple threshold
        if topic == 'sensors/temperature' and value > 35:
            self._trigger_action('COOLING_ON', f'Temperature {value}C exceeds threshold')
        # Slow path: complex reasoning via LLM
        elif len(self.readings) >= 10:
            self._llm_analyze()

    def _trigger_action(self, action: str, reason: str):
        payload = '{"action": "' + action + '", "reason": "' + reason + '"}'
        self.client.publish('agents/actions', payload, qos=1)
        print(f'Action triggered: {action} — {reason}')

    def _llm_analyze(self):
        summary = str(self.readings[-10:])
        result = self.llm.messages.create(
            model='claude-opus-4-5', max_tokens=128,
            messages=[{'role': 'user', 'content':
                f'Sensor readings: {summary}. Any anomalies?'}]
        )
        print('LLM analysis:', result.content[0].text)
        self.readings = []

MQTT su WebSocket

MQTT su WebSocket (porta 8083 o 8084 per TLS) consente a dashboard e agenti basati su browser di connettersi ai broker MQTT senza una connessione TCP nativa. Configuri paho-mqtt per utilizzare il trasporto WebSocket con l'opzione transport='websockets'.

import paho.mqtt.client as mqtt

# MQTT over WebSocket configuration
ws_client = mqtt.Client(
    client_id='dashboard_agent',
    transport='websockets',  # use WS instead of TCP
    protocol=mqtt.MQTTv311
)

# WebSocket path (broker-specific)
ws_client.ws_set_options(path='/mqtt', headers=None)

# TLS over WebSocket (WSS, port 8084)
# ws_client.tls_set()  # uses system CA store

ws_client.connect(
    host='broker.hivemq.com',
    port=8884,  # WSS port
    keepalive=60
)
ws_client.loop_start()
print('Connected via WebSocket')

Last Will and Testament

Il Last Will and Testament (LWT) di MQTT consente al broker di pubblicare un messaggio per conto di un client se quest'ultimo si disconnette in modo imprevisto. Lo utilizzi per informare gli altri agenti che un agente per sensori è andato offline, così che possano passare a una modalità di fallback.

import paho.mqtt.client as mqtt

client = mqtt.Client(client_id='agent_001')

# Set LWT before connecting
client.will_set(
    topic='agents/status/agent_001',
    payload='{"status": "offline", "reason": "unexpected_disconnect"}',
    qos=1,
    retain=True  # retain so new subscribers see the last known status
)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        # Publish 'online' status on connect (overrides LWT retain)
        client.publish(
            'agents/status/agent_001',
            '{"status": "online"}',
            qos=1, retain=True
        )

client.on_connect = on_connect
client.connect('broker.hivemq.com', 1883, keepalive=60)
client.loop_start()

Buone pratiche per la progettazione dei topic

Una buona progettazione dei topic rende il codice degli agenti più facile da gestire e scalabile. Utilizzi una gerarchia: location/device_type/device_id/measurement. Eviti gli spazi e i caratteri speciali. Mantenga i topic brevi: aggiungono overhead a ogni messaggio.

# Good topic hierarchy examples:
# factory/line1/sensor_042/temperature
# home/living_room/thermostat_01/setpoint
# agents/agent_001/commands/turn_on
# agents/agent_001/status

TOPIC_SCHEMA = {
    'sensor_data': '{location}/{device_type}/{device_id}/{measurement}',
    'agent_command': 'agents/{agent_id}/commands/{action}',
    'agent_status': 'agents/{agent_id}/status',
    'alert': 'alerts/{severity}/{location}'
}

def build_topic(schema_key: str, **kwargs) -> str:
    template = TOPIC_SCHEMA[schema_key]
    return template.format(**kwargs)

# Usage:
topic = build_topic('sensor_data',
                    location='factory', device_type='temp_sensor',
                    device_id='042', measurement='celsius')
print(topic)  # factory/temp_sensor/042/celsius

Disconnessione e riconnessione corrette

Gli agenti negli ambienti IoT devono gestire con la massima affidabilità le interruzioni di rete. Utilizzi la logica di riconnessione integrata di paho-mqtt: imposti reconnect_on_failure=True e ripeta la sottoscrizione a ogni riconnessione, poiché le sottoscrizioni non vengono mantenute per impostazione predefinita (a meno che non si utilizzino sessioni persistenti).

import paho.mqtt.client as mqtt
import time

SUBSCRIPTIONS = [
    ('sensors/temperature', 1),
    ('sensors/humidity', 1),
    ('agents/commands', 2)
]

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print('Connected — re-subscribing to topics')
        for topic, qos in SUBSCRIPTIONS:
            client.subscribe(topic, qos=qos)
    else:
        print(f'Connection failed with code {rc}')

client = mqtt.Client(client_id='agent_reliable', clean_session=True)
client.on_connect = on_connect
client.reconnect_delay_set(min_delay=1, max_delay=30)
client.connect_async('broker.hivemq.com', 1883, keepalive=60)
client.loop_start()

# Graceful shutdown:
# client.disconnect()
# client.loop_stop()

Verifica delle conoscenze

Quale livello QoS garantisce che un messaggio venga consegnato esattamente una volta?

Riepilogo: MQTT per l'integrazione degli agenti

Ottimo! In questa lezione ha imparato:

  • paho-mqtt: connect(), loop_start(), subscribe(), publish()
  • Livelli QoS: 0 = al massimo una volta, 1 = almeno una volta, 2 = esattamente una volta
  • Messaggi retained: il broker memorizza l'ultimo valore; i nuovi sottoscrittori lo ricevono immediatamente
  • LWT: il broker pubblica un messaggio offline in caso di disconnessione imprevista
  • Riconnessione: ripeta la sottoscrizione in on_connect; utilizzi reconnect_delay_set

Prossimo argomento: elaborazione dei flussi di dati delle serie temporali — finestre mobili, medie mobili e rilevamento dei picchi.

Domande Frequenti

La lezione «Protocollo MQTT per l’integrazione degli agenti» è gratuita?

Sì — il testo completo di «Protocollo MQTT per l’integrazione degli agenti» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.

Cosa imparerò in «Protocollo MQTT per l’integrazione degli agenti»?

Configurazione del broker, sottoscrizione ai topic e attivazione degli agenti guidata dai messaggi. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Agents?

Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 1 di 4.

Quanto tempo richiede la lezione «Protocollo MQTT per l’integrazione degli agenti»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Agents?

Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Protocollo MQTT per l’integrazione degli agenti
  2. Elaborazione dei dati di serie temporali negli agenti
  3. Risposta automatizzata agli eventi dei sensori
  4. Distribuzione edge di agenti leggeri
← Torna a AI Agents