Python 網路程式設計系列｜FastAPI × MCP × AIoT

從 FastAPI 與 MCP 延伸到 AIoT：讓 Agent 看懂感測資料並安全操作設備

前一篇比較了 FastAPI REST API 與 MCP Server。這篇把兩者放進 AIoT 架構：感測器透過 MQTT 傳送資料，FastAPI 提供一般程式使用的 REST API，MCP Server 把安全且清楚描述的設備功能交給 AI Agent。透過循序圖和 Python 範例，從資料上傳一路看到 Agent 查詢狀態與提出設備控制。

一、AIoT 是什麼？

AIoT 是 AI 與 IoT 的結合。IoT 負責連接感測器、設備和網路；AI 可分析資料、辨識狀況或協助使用者理解問題。真正的系統還需要把資料送進服務、保存與呈現，必要時再把控制命令送回設備。

用魚塭或溫室來想像：感測器量測水溫、溶氧或空氣溫濕度；閘道器把讀值送到伺服器；管理者用儀表板查看資料；AI Agent 則能回答「目前溫度是否偏高？」或在授權和安全規則允許時，呼叫工具啟動通風設備。

這裡的 MQTT、FastAPI、MCP 各自處理不同工作：MQTT 適合設備和後端交換小型訊息；FastAPI 提供一般應用程式可直接呼叫的 HTTP API；MCP 把有名稱、用途與參數說明的工具提供給支援 MCP 的 Agent 或 host。AI 模型不直接接線或控制 GPIO，設備命令仍由受控的服務層執行。

二、整體架構：資料、API 和 Agent 各走自己的入口

元件	在 AIoT 系統中的工作	常見例子
感測器／控制器	讀取環境資料，接收合法的控制命令。	ESP32、Raspberry Pi、溫濕度或水質感測器
MQTT Broker	依 Topic 接收和轉送設備訊息，讓設備不必直接知道每個應用程式的位置。	Mosquitto、雲端 MQTT Broker
AIoT 後端與資料庫	驗證、儲存讀值，套用規則，並連接設備通訊。	Python service layer、SQLite／PostgreSQL
FastAPI REST API	提供網站、手機 App 和一般程式讀取資料或送出管理命令。	GET /api/telemetry/latest
MCP Server 與 Agent	把核准的查詢或控制能力描述成工具，供 Agent 依任務選擇呼叫。	get_latest_reading、set_ventilation

REST 和 MCP 可以共用同一套服務邏輯。REST 的呼叫端依 URL 和 HTTP 方法取用功能；Agent 透過 MCP 看見工具清單及參數，再決定要不要呼叫。MCP 不會自動替設備加上 AI，也不會自動提供授權或安全控制。

三、循序圖一：感測資料從設備進入系統

以下以溫室感測為例。設備發布 MQTT 訊息後，Broker 依 Topic 把資料送給後端；後端檢查欄位並存入資料庫。管理畫面之後透過 REST API 讀取最新資料。

1. 感測節點把資料轉成 JSON，例如 {"device_id":"greenhouse-01","temperature":31.2,"humidity":68.0}，發布到 greenhouse/01/telemetry。
2. MQTT Broker 將訊息交給訂閱該 Topic 的後端。後端檢查裝置身分、欄位型別、數值範圍和時間戳，再寫入資料庫。
3. 一般 Client 呼叫 FastAPI 的讀取 endpoint，取得 JSON 回應。FastAPI 的互動 API 文件可用來檢視和試呼叫 endpoint。

四、循序圖二：Agent 透過 MCP 查詢並操作設備

Agent 收到「看看現在溫室是否太熱，必要時開啟通風」的要求後，先從 MCP Server 取得工具描述，再呼叫讀值工具。模型判斷是否需要操作時，MCP Server 把控制要求交給服務層；服務層檢查限制後，才透過 MQTT 發送命令。

循序圖重點：Agent 只提出工具呼叫；設備控制服務才負責驗證、執行和回報。MCP 的連線生命週期會依協定版本與傳輸方式不同，本圖聚焦在工具探索、工具呼叫和設備命令流程。

五、用共用服務層保存讀值並執行安全規則

REST endpoint 和 MCP tool 都應呼叫同一個 service function。以下用記憶體儲存「最新一筆資料」方便教學；程式重新啟動後資料會消失。實際 AIoT 系統應使用資料庫或共享儲存，並保留歷史紀錄、設備識別和資料時間。

# service.py
from datetime import datetime, timezone
from threading import RLock

_lock = RLock()
_latest = None
_ventilation_enabled = False

def record_reading(device_id: str, temperature: float,
                   humidity: float) -> dict:
    global _latest
    if not -20 <= temperature <= 80:
        raise ValueError("temperature 超出允許範圍")
    if not 0 <= humidity <= 100:
        raise ValueError("humidity 必須介於 0 到 100")
    reading = {
        "device_id": device_id,
        "temperature": temperature,
        "humidity": humidity,
        "received_at": datetime.now(timezone.utc).isoformat(),
    }
    with _lock:
        _latest = reading
    return reading

def latest_reading():
    with _lock:
        return dict(_latest) if _latest else None

def set_ventilation(enabled: bool) -> dict:
    global _ventilation_enabled
    with _lock:
        if _latest is None:
            return {"ok": False, "message": "尚無感測資料"}
        if not enabled and _latest["temperature"] >= 35:
            return {"ok": False, "message": "溫度過高，依安全規則拒絕關閉通風"}
        _ventilation_enabled = enabled
        return {"ok": True, "ventilation_enabled": enabled}

• record_reading() 對資料做基本數值檢查，再加上伺服器接收時間。
• RLock 保護共用資料，避免 MQTT callback 和 API request 同時讀寫造成不一致。
• set_ventilation() 示範設備操作前的規則檢查。實際系統還要驗證呼叫者權限、設備狀態、操作範圍，並留下稽核紀錄。

六、用 FastAPI 提供一般程式使用的 REST API

設備閘道或測試程式可以送資料；儀表板、手機 App 則查詢最新讀值。這個範例讓 REST API 和 MCP Server 掛在同一個 ASGI 應用程式，並共用 service.py。

# main.py（REST + MCP 最小示範）
from contextlib import asynccontextmanager
import json

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from starlette.routing import Mount
from mcp.server import MCPServer

import service

mcp = MCPServer("Greenhouse AIoT")

@mcp.tool()
def get_latest_reading() -> str:
    """讀取溫室最新一筆溫度與濕度。"""
    reading = service.latest_reading()
    return json.dumps(reading or {"message": "尚無感測資料"}, ensure_ascii=False)

@mcp.tool()
def set_ventilation(enabled: bool) -> str:
    """依安全規則開啟或關閉溫室通風設備。"""
    result = service.set_ventilation(enabled)
    return json.dumps(result, ensure_ascii=False)

# 先建立 MCP ASGI app，才能在 host lifespan 啟動 session manager。
mcp_app = mcp.streamable_http_app(streamable_http_path="/")

@asynccontextmanager
async def lifespan(app):
    async with mcp.session_manager.run():
        yield

app = FastAPI(title="Greenhouse AIoT API", lifespan=lifespan)

class ReadingInput(BaseModel):
    device_id: str = Field(min_length=1, max_length=64)
    temperature: float = Field(ge=-20, le=80)
    humidity: float = Field(ge=0, le=100)

@app.post("/api/telemetry")
def receive_reading(payload: ReadingInput):
    return service.record_reading(**payload.model_dump())

@app.get("/api/telemetry/latest")
def read_latest():
    reading = service.latest_reading()
    if reading is None:
        raise HTTPException(status_code=404, detail="尚無感測資料")
    return reading

# 掛載後 MCP endpoint 為 /mcp/；FastAPI routes 仍由同一個 app 提供。
app.mount("/mcp", mcp_app)

@mcp.tool() 將 Python 函式註冊成 Agent 可探索的工具；工具描述和型別資訊會讓 Client 知道用途和輸入參數。MCP SDK 的 ASGI app 可以掛載到 FastAPI；掛載時，最外層應用程式要負責啟動 MCP session manager。範例因此使用 FastAPI lifespan 管理它。

安裝並啟動：

python -m pip install "fastapi[standard]" mcp paho-mqtt
uvicorn main:app --reload

REST 文件在 http://127.0.0.1:8000/docs；REST 最新資料 endpoint 是 http://127.0.0.1:8000/api/telemetry/latest；MCP Streamable HTTP endpoint 掛載在 http://127.0.0.1:8000/mcp/。可先用 REST POST 測試資料：

curl -X POST http://127.0.0.1:8000/api/telemetry \
  -H "Content-Type: application/json" \
  -d '{"device_id":"greenhouse-01","temperature":31.2,"humidity":68.0}'

七、MQTT 如何接進這個服務？

正式設備通常不會一直呼叫 REST POST，而是透過 MQTT 發布資料。後端訂閱指定 Topic，解析 JSON 後呼叫剛才的 service.record_reading()，後續 REST 和 MCP 查詢就能取得同一份狀態。

# mqtt_adapter.py：概念範例，接收端應與 service 共用儲存層
import json
from paho.mqtt import client as mqtt
import service

def on_message(client, userdata, message):
    payload = json.loads(message.payload.decode("utf-8"))
    service.record_reading(
        device_id=payload["device_id"],
        temperature=float(payload["temperature"]),
        humidity=float(payload["humidity"]),
    )

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="aiot-backend")
client.on_message = on_message
client.connect("127.0.0.1", 1883, keepalive=60)
client.subscribe("greenhouse/+/telemetry", qos=1)
client.loop_forever()

部署提醒：上面的 MQTT callback 是獨立程式片段。若它和 FastAPI／MCP 分成不同作業程序，Python 記憶體中的 _latest 不會自動共享；請改用共同的資料庫、Redis 或訊息佇列。真實設備命令也要加入 MQTT TLS、帳號權限、Topic ACL、命令有效期限和設備回報確認。

八、Agent 怎麼使用 MCP 工具？

把 MCP endpoint 設定到支援 Streamable HTTP 的 MCP host，Agent 就能看到 get_latest_reading 和 set_ventilation。使用者問「目前溫室狀況如何？」時，Agent 可先讀值，再把感測數字用自然語言說明；若要求操作設備，host 可要求使用者確認，MCP tool 再進行服務層檢查。

1. Host／MCP Client 連線至 MCP endpoint，取得可用工具及其說明。
2. Agent 根據任務選擇讀值工具，Client 傳送工具名稱和參數。
3. MCP Server 呼叫 service layer，將結果交回 Agent。
4. 若要改變設備狀態，先依產品政策要求使用者確認，再呼叫控制工具。
5. 服務層驗證授權與控制規則，送出設備命令，並等待或查詢設備回報。

若使用 OpenAI Agents SDK 或其他雲端模型，Agent 的模型 API 呼叫會需要有效金鑰和可用額度；這與 MQTT 感測資料上傳及 FastAPI 本身是不同的服務。模型不可用時，固定的本地控制規則仍可獨立執行，避免把即時安全控制完全交給雲端模型。

九、AIoT 控制需要哪些安全設計？

• 先分開查詢和控制：讀值工具可以唯讀；控制工具只提供必要動作，不要開放任意 Python、Shell 或任意 GPIO 命令。
• 在 service layer 執行硬性規則：檢查數值範圍、設備狀態、操作頻率和安全門檻。模型的文字判斷不能覆蓋這些規則。
• 對實體副作用加核准：開門、抽水、加熱、投餌等操作可能影響人身、設備或生物，應依風險要求人工確認或採雙重條件。
• 使用身份驗證和最小權限：REST 與 MCP endpoint 都要設定身份、授權和加密傳輸；遠端 MCP 服務依官方部署文件設定允許的 Host／Origin 和 OAuth 或其他驗證。
• 記錄完整操作鏈：保存提出者、Agent／Client、工具名稱、參數、規則判斷、命令 ID、設備回應和時間戳，方便追蹤誤動作。
• 確保設備失聯時可安全運作：設備端保留超時保護、手動開關和本地安全模式；雲端、網路或 LLM 中斷時，重要控制仍由本地規則接手。

重要：MCP 只是工具呼叫協定，並不代表 Agent 呼叫必定安全，也不會自動替每個動作要求人工核准。授權、人工確認和設備安全互鎖，必須由 MCP host、後端服務和設備端共同設計。

十、建置順序建議

1. 先做資料上傳：用模擬器或 ESP32 發布一筆溫度與濕度資料。
2. 再建立資料入口：以 MQTT 接收資料，驗證後存入資料庫。
3. 提供 REST 查詢：先做只讀 endpoint 和簡單儀表板，確認資料正確。
4. 加入 MCP 讀取工具：先讓 Agent 讀取狀態、摘要趨勢，不要一開始就開放控制。
5. 最後加入受控操作：先在模擬設備測試，新增權限、人工確認、規則、稽核與回報，再連接真實致動器。

十一、結語

在這個 AIoT 架構中，MQTT 負責設備訊息傳遞，FastAPI 提供一般程式明確呼叫的 REST API，MCP 則把經過挑選的系統能力描述給 Agent 使用。三者可以共存，核心是讓 REST 和 MCP 都使用同一套資料與服務邏輯。先把資料收集、驗證、儲存做好，再逐步加入 Agent 查詢和受控操作，AI 才能建立在可靠的設備資料之上。

官方參考資料：
FastAPI Tutorial｜ MCP Python SDK（v2）｜ MCP SDK：整合到既有 ASGI 應用｜ MCP SDK 部署與傳輸安全｜ Eclipse Paho MQTT Python Client。SDK 與協定會更新，實作前請依目前官方文件確認版本與部署設定。