MQTT için MCP Sunucusu: Yapay Zeka Asistanları Cihazlarınızı Okusun ve Kontrol Etsin

MQTT için bir MCP sunucusu, yapay zeka asistanlarına canlı cihaz verilerine bakmanın ve siz izin verdiğinizde komut göndermenin yapılandırılmış, yetkilendirilmiş bir yolunu sunar. Bu eğitimde Python ile bir MQTT broker'a abone olan, her topic için en son mesajı önbelleğe alan ve küçük bir araç seti sunan bir Model Context Protocol sunucusu yazacak, ardından onu MCP destekli bir asistana bağlayıp cihazlarınız hakkında sade bir dille sorular soracaksınız.

MCP, Bir MQTT Sistemine Ne Katar?

Model Context Protocol, yapay zeka uygulamalarını harici araçlara ve verilere bağlamak için geliştirilmiş açık bir protokoldür. Bir MCP sunucusu araçlar (tipli parametrelere ve açıklamalara sahip fonksiyonlar) tanımlar; genellikle bir masaüstü asistanı ya da IDE ajanı olan client da modelin bu araçları çağırmasına izin verir. Gerçekte neyin mümkün olduğuna sunucu karar verir.

MQTT ve MCP farklı sorunları çözer ve birlikte iyi çalışır:

MCP sunucusu ikisinin arasında durur: bir tarafta sıradan bir MQTT client'ı, diğer tarafta bir MCP araç sağlayıcısıdır. MQTT push tabanlı, MCP ise pull tabanlı olduğundan sunucu her topic için en son mesajın önbelleğini tutar; böylece araç çağrıları bir sonraki yayını beklemek yerine anında yanıt verir.

Mimari

AI assistant  <--stdio / MCP-->  mqtt_mcp_server.py  <--MQTT-->  broker  <--MQTT-->  devices
                                  |- background MQTT loop
                                  |- latest-value cache
                                  |- tools: list_topics, get_latest, send_command

Sunucu yalnızca tek bir öneke abone olur ve komutlar yalnızca izin listesindeki bir eylemle .../devices/<id>/cmd adresine yayınlanabilir. Model hiçbir zaman rastgele topic'lere ham publish erişimi elde etmez.

Adım 1: Bağımlılıkları Kurun

Bir sanal ortam oluşturun ve resmi MCP Python SDK'sı ile paho-mqtt 2.x'i kurun:

python3 -m venv .venv
source .venv/bin/activate        # on Windows: .venv\Scripts\activate
pip install "mcp[cli]" "paho-mqtt>=2.0"

Adım 2: Birkaç Cihazı Simüle Edin

Sorgulayacak bir şeyiniz olması için, üç cihazın durumunu beş saniyede bir retained olarak yayınlayan ve komutları onaylayan küçük bir simülatör çalıştırın. Bu eğitimin her yerinde 5d81fa değerini kendi rastgele dizinizle değiştirin.

# simulator.py
import json
import random
import time
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/5d81fa"
devices = {
    "pump-1": {"running": True, "flow_lpm": 42.0, "pressure_bar": 2.1},
    "tank-1": {"level_pct": 63.0, "temp_c": 21.5},
    "fan-1": {"running": True, "speed_pct": 50},
}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/devices/+/cmd", qos=1)


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    cmd = json.loads(msg.payload)
    state = devices.get(device)
    result = "unknown_device"
    if state is not None:
        if cmd["action"] in ("start", "stop") and "running" in state:
            state["running"] = cmd["action"] == "start"
            result = "applied"
        elif cmd["action"] == "set_speed" and "speed_pct" in state:
            state["speed_pct"] = max(0, min(100, int(cmd["value"])))
            result = "applied"
        else:
            result = "unsupported"
    ack = {"id": cmd.get("id"), "result": result, "ts": int(time.time() * 1000)}
    client.publish(f"{PREFIX}/devices/{device}/ack", json.dumps(ack), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-5d81fa")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    devices["pump-1"]["flow_lpm"] = round(40 + random.uniform(-3, 3), 1) if devices["pump-1"]["running"] else 0.0
    devices["tank-1"]["level_pct"] = round(min(100, max(0, devices["tank-1"]["level_pct"] + random.uniform(-1, 1))), 1)
    for name, state in devices.items():
        payload = {**state, "ts": int(time.time() * 1000)}
        client.publish(f"{PREFIX}/devices/{name}/state", json.dumps(payload), qos=1, retain=True)
    time.sleep(5)

Adım 3: MQTT için MCP Sunucusunu Yazın

Eğitimin özü burası. Resmi SDK'daki FastMCP, tipli Python fonksiyonlarını MCP araçlarına dönüştürür ve docstring'leri modelin okuduğu açıklamalar olarak kullanır.

# mqtt_mcp_server.py
import json
import logging
import sys
import threading
import time
import uuid

import paho.mqtt.client as mqtt
from mcp.server.fastmcp import FastMCP

# stdio transport uses stdout for protocol messages: log to stderr only
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("mqtt-mcp")

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/5d81fa"
ALLOWED_ACTIONS = {"start", "stop", "set_speed"}

cache: dict[str, dict] = {}
acks: dict[str, dict] = {}
lock = threading.Lock()


def on_connect(client, userdata, flags, reason_code, properties):
    log.info("MQTT connected: %s", reason_code)
    client.subscribe(f"{PREFIX}/#", qos=1)


def on_message(client, userdata, msg):
    text = msg.payload.decode("utf-8", errors="replace")
    with lock:
        cache[msg.topic] = {"payload": text, "retained": msg.retain, "receivedAt": time.time()}
        if msg.topic.endswith("/ack"):
            try:
                ack = json.loads(text)
                if ack.get("id"):
                    acks[ack["id"]] = ack
            except json.JSONDecodeError:
                pass


mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="mcp-bridge-5d81fa")
mqttc.on_connect = on_connect
mqttc.on_message = on_message
mqttc.connect(BROKER, PORT, keepalive=60)
mqttc.loop_start()

mcp = FastMCP("mqtt-devices")


@mcp.tool()
def list_topics(contains: str = "") -> list[str]:
    """List MQTT topics that have received data, optionally filtered by a substring."""
    with lock:
        return sorted(t for t in cache if contains in t)


@mcp.tool()
def get_latest(device_id: str) -> dict:
    """Return the latest state message of a device (for example pump-1) and its age in seconds."""
    topic = f"{PREFIX}/devices/{device_id}/state"
    with lock:
        entry = cache.get(topic)
    if entry is None:
        return {"error": f"no data for {device_id}"}
    try:
        data = json.loads(entry["payload"])
    except json.JSONDecodeError:
        data = entry["payload"]
    return {"device": device_id, "state": data,
            "ageSeconds": round(time.time() - entry["receivedAt"], 1)}


@mcp.tool()
def send_command(device_id: str, action: str, value: float | None = None) -> dict:
    """Send a command to a device. action must be start, stop or set_speed (value 0-100).
    Waits up to 5 seconds for the device acknowledgement."""
    if action not in ALLOWED_ACTIONS:
        return {"error": f"action must be one of {sorted(ALLOWED_ACTIONS)}"}
    if action == "set_speed" and (value is None or not 0 <= value <= 100):
        return {"error": "set_speed requires a value between 0 and 100"}
    with lock:
        known = f"{PREFIX}/devices/{device_id}/state" in cache
    if not known:
        return {"error": f"unknown device {device_id}"}
    cmd_id = str(uuid.uuid4())
    cmd = {"id": cmd_id, "action": action, "value": value, "source": "mcp"}
    mqttc.publish(f"{PREFIX}/devices/{device_id}/cmd", json.dumps(cmd), qos=1)
    deadline = time.time() + 5
    while time.time() < deadline:
        with lock:
            if cmd_id in acks:
                return {"sent": cmd, "ack": acks.pop(cmd_id)}
        time.sleep(0.1)
    return {"sent": cmd, "ack": None, "note": "no acknowledgement within 5 seconds"}


if __name__ == "__main__":
    mcp.run()  # stdio transport by default

Burada üç ayrıntı önemli:

1. Asla stdout'a yazdırmayın. stdio transport'unda stdout, JSON-RPC protokolünü taşır. Gözden kaçmış tek bir print() akışı bozar ve client bir ayrıştırma (parse) hatasıyla bağlantıyı keser. Log'ları stderr'e yazın.
2. MQTT döngüsü kendi thread'inde çalışır. loop_start(), MCP sunucusu araç çağrılarını işlerken önbelleği güncel tutar.
3. Doğrulama sunucudadır. Docstring modele neyin izinli olduğunu söyler, ancak bunu kod uygular; modelin uydurduğu değerler için de.

Adım 4: MCP Inspector ile Test Edin

Simülatörü bir terminalde başlatın, ardından sunucuyu, araçları elle çağırabileceğiniz bir web arayüzü açan inspector altında çalıştırın (Node.js gerektirir):

python simulator.py
mcp dev mqtt_mcp_server.py

Önce list_topics'i, sonra pump-1 ile get_latest'i, ardından fan-1, set_speed, 70 ile send_command'ı çağırın. "result": "applied" içeren bir onay görmelisiniz. Aynı anda ham trafiği izlemek için:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-5d81fa -t "connectmq/demo/5d81fa/#" -v

Adım 5: Bir Yapay Zeka Asistanı Bağlayın

MCP client'ları stdio sunucularını bir yapılandırma dosyasından başlatır. Yaygın mcpServers biçimini kullanan masaüstü asistanları için, sanal ortamınızdaki Python'a ve betiğe giden mutlak yolları içeren bir girdi ekleyin:

{
  "mcpServers": {
    "mqtt-devices": {
      "command": "/home/you/mqtt-mcp/.venv/bin/python",
      "args": ["/home/you/mqtt-mcp/mqtt_mcp_server.py"]
    }
  }
}

Asistanı yeniden başlatın ve şunu sorun: "Tank seviyesi şu anda ne, pompa çalışıyor mu?" Model get_latest'i iki kez çağırır ve canlı değerlerle yanıt verir. Sonra "fan-1'i yüzde 30'a ayarla" deyin: araç çağrısını, onayı ve yanıtı göreceksiniz. Çoğu client araç çağrılarını onaylamanızı ister; cihaz durumunu değiştiren her araç için bu onayı açık tutun.

Modelin Anlayacağı Araç Açıklamaları Yazmak

FastMCP'de fonksiyon adı, parametre adları, tip ipuçları ve docstring, modelin bir araç hakkında bildiği her şeydir. Belirsiz açıklamalar belirsiz araç kullanımı doğurur; bu yüzden onları arayüzün bir parçası olarak ele alın:

1. Aracın yalnızca ne yaptığını değil, ne döndürdüğünü de söyleyin. "Bir cihazın en son durum mesajını ve saniye cinsinden yaşını döndür" ifadesi, modele verinin ne kadar taze olduğu üzerine akıl yürütebileceğini söyler.
2. Geçerli değerleri adlandırın. İzin verilen eylemleri ve aralıkları docstring'de listelemek, kod bunları yine de uygulasa bile başarısız çağrıları azaltır.
3. Topic yollarını değil, alan adlarını kullanın. Model tam bir topic dizesi oluşturmak yerine pump-1 istemelidir. Topic düzenini cihaz kimliklerinin arkasına gizlemek, modelin yolları tahmin etmesini engeller ve topic'leri sonradan yeniden düzenlemenize olanak tanır.
4. Hataları veri olarak döndürün. {"error": "unknown device fan-9"} döndürmek modelin kendini düzeltmesine veya sorunu kullanıcıya açıklamasına olanak tanır; bir istisna fırlatmak ise konuşmayı çoğu zaman genel bir hatayla sonlandırır.
5. Araç listesini kısa tutun. Üç ila altı odaklı araç, birbiriyle çakışan bir düzine araçtan daha iyi çalışır. Daha fazlasına ihtiyacınız varsa, onları amaca göre sunuculara bölün.

Okuma sonuçlarına her değerin yaşını eklemek de işe yarar. "ageSeconds": 900 gören bir model genellikle okumanın eski olduğunu belirtir; arkasındaki cihaz çevrimdışı olabilecekken istediğiniz tam da bu temkinliliktir.

MCP Sunucusunu Üretim İçin Sağlamlaştırmak

Bu eğitim, denemeler için ideal olan ancak herkesin okuyabildiği ücretsiz herkese açık MQTT broker üzerinde çalışır. Bir asistanı gerçek ekipmana yönlendirmeden önce:

Pratik bir desen iki MCP sunucusudur: serbestçe etkinleştirebileceğiniz salt okunur bir sunucu ve yalnızca gerektiğinde etkinleştirdiğiniz bir komut sunucusu. ConnectMQ gibi yönetilen bir MQTT broker'da her sunucu portaldan kendi oluşturulmuş kullanıcı adını, parolasını ve client ID'sini alır; test ederken her iki client'ı da canlı topic gezgininde izleyebilirsiniz. Broker'ın topic bazında izinler sağlamadığını unutmayın; bu yüzden okuma ile komut verme arasındaki ayrım sunucularınızda uygulanmalıdır.

Güvenli araç yüzeylerinin arkasındaki tasarım ilkeleri için IoT cihazları için LLM ajanı oluşturma rehberimizi okuyun; şema doğrulama desenleri için de IoT için LLM function calling yazımıza bakın. Resmi Model Context Protocol dokümantasyonu, sunucuyu uzakta barındırmanız gerekirse streamable HTTP gibi diğer transport'ları açıklar. Büyük resim için yapay zeka ve IoT ana rehberimiz bu parçaları bir araya getiriyor.

ConnectMQ ile Başlayın

MCP sunucunuzu kontrolünüzdeki bir broker'a yönlendirin. ConnectMQ; MQTT 5, 8883 portunda TLS ve test için tarayıcı içi bir client sunan bir bulut MQTT broker sağlar. Kredi kartı gerektirmeden ücretsiz Serverless planla başlayın ve asistanınıza dakikalar içinde canlı cihaz bağlamı kazandırın.

Sık Sorulan Sorular

MCP sunucusunun neden bir önbelleğe ihtiyacı var?

MQTT, cihazlar yayın yaptığında mesajları iter (push); MCP araç çağrıları ise veriyi talep üzerine ister. Önbellek her topic için en son mesajı saklar; böylece bir araç çağrısı bir sonraki yayını beklemek yerine anında yanıt verebilir.

Yapay zeka asistanı topic'lere doğrudan abone olabilir mi?

Akış anlamında MCP araçları üzerinden hayır. Sunucu asistan adına abone olur ve anlık görüntüler sunar; bu da hangi topic'lerin görünür olduğu üzerindeki kontrolü korur.

Bir asistanın MQTT komutları göndermesine izin vermek güvenli mi?

Yalnızca doğrulama, cihaz tarafı limitler, onaylar ve kullanıcı teyidi içeren dar araçlar üzerinden. Bir modele asla genel bir publish aracı açmayın.

Sunucuyu başlatırken client'ım neden JSON parse hatası gösteriyor?

Bir şey stdout'a yazıyor; stdio transport'u ise stdout'u protokol mesajlarına ayırır. Tüm log ve hata ayıklama çıktılarını stderr'e gönderin.