用一顆按鈕控制開發板上的 LED

用一顆按鈕控制開發板上的 LED：Thonny、MicroPython 與 Codex 實作紀錄

在學習 MicroPython 的過程中，一個很適合入門的實驗， 就是使用按鈕控制開發板上的 LED。 這次透過 Thonny 撰寫程式，完成「按一下按鈕，LED 切換亮滅」的功能， 並將開關與 LED 狀態輸出到 Thonny Shell。

硬體接線

本次使用 Raspberry Pi Pico 類型的開發板，接線方式如下：

• 按鈕一端接 GPIO 15
• 按鈕另一端接 GND
• LED 使用開發板上的板載 LED
• 程式使用內建上拉電阻

使用上拉電阻時，按鈕沒有按下時會讀到高電位； 按下按鈕後，GPIO 15 會接地並讀到低電位。

使用 Codex 協助完成的過程

這次實作使用 Codex 協助撰寫程式、修改功能，以及分析硬體測試時遇到的問題。 一開始先向 Codex 說明需求：使用一顆按鈕控制開發板上的板載 LED。

1. 使用 Codex 建立按鈕控制 LED 的 MicroPython 程式。
2. 透過 Thonny 開啟程式，並儲存為 button_led.py。
3. 加入 print()，將按鈕與 LED 狀態輸出到 Thonny Shell。
4. 實際測試時發現 LED 沒有亮，將測試結果回報給 Codex。
5. Codex 協助分析後，發現固定使用 GPIO 25 不一定適用於目前的開發板。
6. 將 LED 設定修改為 Pin("LED", Pin.OUT)。
7. 再次測試後，LED 成功恢復正常亮滅。

一開始遇到的問題

最初使用 GPIO 25 控制板載 LED：

led = Pin(25, Pin.OUT)

程式雖然可以執行，但 LED 卻沒有亮。 原因是不同型號的開發板，板載 LED 可能使用不同的控制方式。 因此，固定使用 GPIO 25 不一定能控制目前使用的板載 LED。

修正方式

將 LED 設定改為使用開發板定義的 LED 名稱：

led = Pin("LED", Pin.OUT)

這樣 MicroPython 會依照目前使用的開發板，自動對應正確的板載 LED。 修正後，LED 就能正常亮滅。

完整 MicroPython 程式

from machine import Pin
from time import sleep

# 使用板子定義的板載 LED 腳位
led = Pin("LED", Pin.OUT)

# 按鈕接在 GPIO 15 與 GND，使用內建上拉電阻
button = Pin(15, Pin.IN, Pin.PULL_UP)

while True:
    if button.value() == 0:
        led.value(not led.value())

        print("開關：ON，LED：", "亮" if led.value() else "滅")

        # 等待按鈕放開，避免一次觸發多次
        while button.value() == 0:
            sleep(0.01)

        # 簡單防止按鍵彈跳
        sleep(0.02)

Thonny Shell 輸出

按下按鈕後，Thonny Shell 會顯示：

開關：ON，LED： 亮

再次按下按鈕時，則會顯示：

開關：ON，LED： 滅

這次實作學到的事情

• 如何使用 MicroPython 設定 GPIO 輸入與輸出。
• 如何使用按鈕控制 LED 的亮滅。
• 如何使用內建上拉電阻簡化按鈕接線。
• 如何使用 print() 在 Thonny Shell 顯示狀態。
• 如何透過等待按鈕放開與延遲處理按鍵彈跳。
• 不同開發板的板載 LED 腳位定義可能不同。
• 如何使用 Codex 協助產生程式、分析錯誤並進行修改。

實作心得

這次實驗讓我了解到，不同開發板的硬體定義可能不完全相同。 使用固定 GPIO 編號前，必須先確認該編號是否真的對應到板載 LED。 使用 Pin("LED", Pin.OUT) 可以讓程式具備更好的相容性。

此外，按鈕程式需要加入等待放開與按鍵防彈跳處理， 否則按一下可能會被程式判斷成多次按鍵。 透過 Codex 的協助，可以快速建立程式、加入新功能， 並根據實際硬體測試結果找出問題。

這個小實驗也可以進一步延伸成長按偵測、LED 閃爍模式， 或加入蜂鳴器與其他感測器。

小提醒：執行程式前，請確認 Thonny 已選擇正確的 MicroPython 開發板與連接埠。

用 Codex 與 Thonny 打造 Pico 2 W 智慧生活專案

Codex × Thonny × Raspberry Pi Pico 2 W

用自然語言打造智慧生活小專案

本文示範如何使用 Codex 協助操作 Thonny，從 Hello World 開始，逐步完成 LED 閃爍，以及可由手機控制的 Pico 2 W 內建網站。

01Hello World
確認開發環境

02LED 閃爍
控制板載硬體

03手機控制
建立內建 AP 網站

Codex 在這個專案中的角色

Codex 可以依照自然語言提示，協助開啟 Thonny、輸入程式、切換到 MicroPython（Raspberry Pi Pico）直譯器，並執行程式。以下畫面是部落格用的操作示意：左側是 Codex 對話，右側是 Thonny 編輯器與 Shell。

範例一：執行 Hello World

Codex 提示詞請在 Thonny 執行 Hello World 程式

Codex 與 Thonny

Codex

請在 Thonny 執行 Hello World 程式

已開啟 Thonny，輸入程式並執行。

Thonny

Thonny — <untitled>

print("Hello World")

>>> Hello World

這個步驟可用來確認 Thonny 與 Python 執行環境正常。

範例二：讓 Pico LED 每秒閃爍

Codex 提示詞用 Thonny 執行控制 Pico 板子上的 LED 每秒鐘閃爍一次

Codex 與 Thonny

Codex

用 Thonny 執行控制 Pico 板子上的 LED 每秒鐘閃爍一次

已切換為 MicroPython（Raspberry Pi Pico），並執行 LED 程式。

Thonny / MicroPython

MicroPython (Raspberry Pi Pico)

from machine import Pin from time import sleep led = Pin("LED", Pin.OUT) while True: led.toggle() sleep(1)

MPY: soft reboot

執行前請確認 Pico 透過 USB 連接電腦，並在 Thonny 選擇 Raspberry Pi Pico 的 MicroPython 直譯器。

範例三：用手機控制 Pico 2 W LED

Codex 提示詞Raspberry Pi Pico 2 W 內建一個網站，可以經由手機來控制板上的 LED ON/OFF

補充提示詞可以內建 AP

Codex 與 Thonny

Codex

可以內建 AP

已建立 Pico 2 W 熱點與 LED 控制網站。

SSID: Pico2W-LED
Password: picoled123
Open http://192.168.4.1

Thonny / Pico 2 W

MicroPython — Raspberry Pi Pico 2 W

ap = network.WLAN(network.AP_IF) ap.config(ssid="Pico2W-LED", password="picoled123") ap.active(True) if b"GET /on" in request: led.on() elif b"GET /off" in request: led.off()

Wi-Fi AP ready
Open http://192.168.4.1

手機連線到 Pico2W-LED，密碼為 picoled123，再開啟 http://192.168.4.1，即可看到 LED ON 與 LED OFF 按鈕。

完整網站程式

import network
import socket
from machine import Pin

led = Pin("LED", Pin.OUT)
ap = network.WLAN(network.AP_IF)
ap.config(ssid="Pico2W-LED", password="picoled123")
ap.active(True)

while not ap.active():
    pass

print("Open http://192.168.4.1")
addr = socket.getaddrinfo("0.0.0.0", 80)[0][-1]
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(addr)
server.listen(1)

while True:
    client, remote = server.accept()
    request = client.recv(1024)
    if b"GET /on" in request:
        led.on()
    elif b"GET /off" in request:
        led.off()
    status = "ON" if led.value() else "OFF"
    page = "<h1>Pico 2 W LED</h1>"
    page += "<p>LED: " + status + "</p>"
    page += "<a href='/on'><button>LED ON</button></a>"
    page += " <a href='/off'><button>LED OFF</button></a>"
    response = "HTTP/1.1 200 OK" + chr(13) + chr(10)
    response += "Content-Type: text/html" + chr(13) + chr(10)
    response += "Connection: close" + chr(13) + chr(10) + chr(13) + chr(10)
    client.send((response + page).encode())
    client.close()

安全提醒：本文的 AP 名稱與密碼是示範用設定。實際使用時請更換密碼，並避免在公開環境中使用簡單密碼。

本文示範 Codex 如何協助操作 Thonny 與 Raspberry Pi Pico 2 W，適合做為 MicroPython 與智慧生活專案的入門教材。

網際網路應用系列｜從 TCP 到 AIoT

從五篇文章認識網際網路技術：TCP、HTTP、MCP、Agent 與 AIoT

五篇文章分別談 asyncio TCP、MCP Server／Client、Agent 呼叫 MCP、FastAPI REST API 與 MCP，以及 AIoT。把它們放在一起看，會發現它們不是五種互相取代的技術，而是網際網路應用中不同層次的工具：底層負責傳輸，應用協定定義訊息，框架提供服務入口，Agent 決定何時使用工具，AIoT 再把網路服務連到真實設備。

一、先從網際網路的分層觀念開始

當瀏覽器連上一個網站，或手機向感測器平台查資料，資料會經過多個層次。每一層只處理自己的一部分工作，並透過明確介面交給上一層。

層次／角色	主要問題	本系列的例子
應用程式	使用者或系統要完成什麼任務？	Agent 回答問題、儀表板顯示魚塭資料
應用層協定	訊息有什麼格式、方法與語意？	HTTP、MCP、MQTT
傳輸層	兩個端點如何傳送資料？	TCP；Python asyncio Streams 提供程式介面
網路層與名稱解析	資料要送到哪台主機？	IP 位址、路由、DNS 名稱
實體與鏈路	如何經由乙太網路、Wi-Fi 或行動網路送出？	校園網路、Wi-Fi、4G／5G、乙太網路

這是教學用的簡化分層圖。實際網路協定堆疊的分類方式不只一種；學習時重要的是辨認每個元件的責任。TCP 是傳輸協定；HTTP、MQTT 和 MCP 定義應用層互動；FastAPI 是 Python Web 框架；Agent 是使用模型和工具的應用程式模式；asyncio 是 Python 的非同步 I/O 函式庫。

二、TCP：可靠傳送位元組，但不認識你的資料格式

TCP 在兩個端點間建立連線，提供可靠、有順序的位元組串流。TCP 不知道一段 bytes 是一句話、一筆 JSON，還是多個應用訊息黏在一起。應用程式必須自行定義訊息邊界，例如每筆訊息以換行結尾、使用固定長度，或先傳送資料長度。

# asyncio TCP Server：以換行符號作為訊息邊界
import asyncio

async def handle_client(reader: asyncio.StreamReader,
                        writer: asyncio.StreamWriter) -> None:
    try:
        line = await reader.readline()
        message = line.decode("utf-8").rstrip("\n")
        writer.write(f"收到：{message}\n".encode("utf-8"))
        await writer.drain()
    finally:
        writer.close()
        await writer.wait_closed()

async def main() -> None:
    server = await asyncio.start_server(
        handle_client, "127.0.0.1", 8888
    )
    async with server:
        await server.serve_forever()

asyncio.run(main())

asyncio.start_server() 建立 TCP 服務；StreamReader 和 StreamWriter 提供收發資料的高階介面。await 表示程式在等網路資料或寫入完成時，事件迴圈可以去處理其他連線。這改善了 I/O 等待時的資源利用方式，但不會改變 TCP 本身，也不會讓 CPU 密集工作自動平行化。Python 文件將 asyncio 定位為以 async／await 撰寫並行程式的函式庫，常見用途包括網路和 Web 伺服器。

要記住：TCP 負責送達 bytes；換行符、JSON、指令名稱等內容格式是應用程式自己定義的。若需要一般 Web API 的請求方法、標頭、狀態碼和內容格式，通常會在 TCP 之上使用 HTTP。

三、HTTP 與 REST：讓一般程式用一致方式呼叫服務

HTTP 以 request／response 方式溝通。Client 傳送方法、路徑、標頭和可能的 body；Server 回傳狀態碼、標頭和回應內容。REST 是設計 HTTP API 的一種架構風格，常以資源路徑和 HTTP 方法表達操作，例如用 GET /api/sensors/pond-01 讀取設備資料。

FastAPI 是 Python Web 框架，可依函式參數、型別標註和資料模型建立 HTTP 路由與 API schema。它會依 OpenAPI 標準產生 API 描述和互動文件。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Reading(BaseModel):
    device_id: str
    temperature: float

@app.get("/api/sensors/{device_id}", response_model=Reading)
async def get_sensor(device_id: str):
    # 真正應用中，這裡呼叫共用 service 或資料庫
    return Reading(device_id=device_id, temperature=28.6)

• @app.get(...) 把 Python 函式註冊成 HTTP GET 路由。
• 路徑參數 device_id 對應 URL 中的設備名稱。
• Pydantic model 描述 JSON 回應欄位，幫助驗證與產生文件。
• 瀏覽器或手機 App 不必理解 TCP framing，只要依 HTTP 和 JSON 規則呼叫。

四、MCP：讓 AI host 探索和呼叫工具的協定

MCP（Model Context Protocol）處理的是 AI 應用如何連接外部資料與工具。MCP Server 可公布 tools、resources 或 prompts；MCP Client 代表 AI host 與 Server 溝通。以工具為例，Client 可以取得工具清單，再傳送工具名稱與參數要求 Server 執行。MCP 的訊息語意由協定定義，傳輸可以使用 stdio 或 Streamable HTTP 等方式。

from mcp.server import MCPServer

mcp = MCPServer("Pond Monitor")

@mcp.tool()
def get_latest_temperature(device_id: str) -> str:
    """讀取指定魚塭最新的水溫。"""
    reading = sensor_service.latest(device_id)
    return f"{device_id} 水溫：{reading.temperature}°C"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

@mcp.tool() 把一般 Python 函式註冊為工具；函式名稱、說明、參數型別會協助 Client 認識可用能力。Agent 並不是 MCP Server：Agent 通常住在 AI host 中，MCP Client 負責協定往返，MCP Server 負責提供實際工具。MCP 也不等於 REST API；雖然 MCP 可透過 HTTP 傳輸，MCP request 仍遵照 MCP 的方法、工具描述和回應格式。

五、Agent：應用程式中的任務規劃與工具選擇

Agent 是應用層的工作流程：接收使用者目標，利用模型理解語意，判斷是否需要資料或工具，接著透過 MCP Client 呼叫工具，最後把工具結果整理成回覆。Agent 不是新的底層網路協定；網路傳輸仍由 MCP Client／Server、HTTP、stdio 或其他 transport 處理。

1. 使用者提出問題：「昨天晚上魚塭水溫有沒有異常？」
2. Agent 根據任務及工具描述，選擇歷史讀值工具。
3. MCP Client 把工具名稱和參數送給 MCP Server。
4. MCP Server 呼叫後端 service 或資料庫取得讀值。
5. Agent 依結果產生自然語言說明，並可在不確定時追問。

若工具會改變外部狀態，例如啟動抽水馬達，Agent 的「選擇呼叫」不能取代後端授權、操作限制、人工核准和設備端保護。工具定義只說明能力，不會自動替能力加上安全機制。

六、MQTT：IoT 設備的發布／訂閱通訊

MQTT 常用在設備遙測與控制。設備可以把讀值 publish 到 Topic，後端依 Topic subscribe；Broker 負責把訊息送給符合訂閱條件的 Client。這種發布／訂閱模式讓生產資料的設備不必知道每一個儀表板或分析程式的位址。MQTT 5.0 規格將其定義為 Client／Server 的發布／訂閱訊息傳輸協定。

# 概念示意：感測器發布一筆遙測資料
topic = "pond/pond-01/telemetry"
payload = {
    "temperature": 28.6,
    "dissolved_oxygen": 6.4
}

# 感測器 publish；後端訂閱 pond/+/telemetry
mqtt_client.publish(topic, json.dumps(payload), qos=1)

MQTT 專注在設備訊息流；FastAPI 適合被網站、App 或其他服務主動呼叫；MCP 適合將經過選擇的功能提供給 AI host。把三者放在一起，就能同時照顧設備資料流、一般軟體 API 和 Agent 工具呼叫。

七、循序圖一：AIoT 設備資料如何進入網路服務

感測器透過 MQTT 發布資料；Broker 轉交給已訂閱的 Python 後端；後端檢查並存入資料庫。一般應用程式之後使用 FastAPI 查詢，而不是直接連到感測器。

MQTT 遙測資料通常是設備主動上報；REST 查詢則由 Client 發出請求。兩種資料流可以同時存在，後端服務層負責把它們接到相同資料。

八、循序圖二：Agent 如何透過 MCP 工具完成查詢

使用者提出問題後，Agent 先取得工具描述並選擇工具；MCP Server 呼叫共用服務層取得資料，再把工具結果交回 Agent。這是 AI 應用的協定呼叫流程，並非模型直接連接 TCP socket 或感測器。

MCP 連線管理和握手步驟會依協定版本、SDK 與傳輸模式而異；圖中聚焦在工具清單、工具呼叫和結果回傳等主要應用語意。

九、五個主題各自扮演什麼角色？

技術	它是什麼	最適合負責
TCP	傳輸層協定，提供可靠、有順序的 byte stream。	兩個程式間的底層資料通道；應用程式要另定訊息邊界。
asyncio	Python 非同步 I/O 函式庫與程式寫法。	在網路等待期間處理其他 coroutine；不是新的網路協定。
HTTP／REST	HTTP 是應用協定；REST 是 API 設計風格。	網站、手機 App、一般程式明確呼叫服務。
FastAPI	Python Web 框架。	建立 HTTP 路由、驗證請求資料及產生 OpenAPI 文件。
MCP	AI 應用探索工具、資源和提示的協定。	讓支援 MCP 的 host／Agent 用共同格式使用外部能力。
Agent	使用模型理解任務並選擇工具的應用程式。	依上下文決定何時查詢或呼叫工具，組織回覆。
MQTT	發布／訂閱訊息協定。	設備遙測上報、狀態通知和命令傳遞。

十、把五篇文章整合成 AIoT 網際網路應用

以智慧魚塭為例，完整資料和控制路徑可以這樣設計：

水質感測器→MQTT Broker→Python service／資料庫←FastAPI REST Client

使用者問題→Agent→MCP Client→MCP Server→共用 service

感測資料透過 MQTT 進入平台；管理儀表板經 FastAPI 查詢歷史趨勢；Agent 經 MCP 查詢某個魚塭的讀值，或根據使用者授權提出開啟增氧設備的要求。REST endpoint 與 MCP tool 都呼叫同一服務層，讓權限、數值檢查和操作紀錄集中管理。

實作順序

1. 先做單一 TCP Server／Client，理解主機、埠號、連線和訊息邊界。
2. 再用 asyncio Streams 改寫網路 I/O，認識 coroutine、await 和事件迴圈。
3. 使用 FastAPI 建立 REST endpoint，確認 HTTP 方法、狀態碼、JSON 和 OpenAPI 文件。
4. 將已存在的服務功能註冊成 MCP tool，讓 MCP Inspector 或 AI host 探索並呼叫。
5. 加入 Agent 作為任務編排者，並把需要模型 API 的部分與一般 API／設備通訊分開管理。
6. 最後接 MQTT 感測設備，再把控制命令設計成受限制、可稽核、可人工核准的工具。

十一、部署時別忽略網路與安全

• 確認連線位置：127.0.0.1 只能讓本機連線；區網或雲端服務要設定可達的網卡位址、DNS、路由和防火牆。
• 分清通訊埠：TCP 範例、FastAPI、MCP Streamable HTTP 和 MQTT Broker 可使用不同 port；開放 port 前先確認網路範圍和必要性。
• 使用 TLS 與身份驗證：正式環境避免明文傳送密碼、Token 或感測資料；設備、REST API、MCP endpoint 都應依角色限制權限。
• 保護 API 金鑰：金鑰應放在伺服器環境變數或秘密管理服務，不要寫進前端 JavaScript、公開原始碼或部落格。401 常代表認證資料錯誤；429 可能是速率或額度限制，需查看服務端回應與帳務設定。
• 控制工具要守規則：Agent 可建議或請求動作；設備服務必須再次檢查範圍、權限、互鎖和核准，不要讓模型直接控制任意命令。
• 設計失聯策略：網路、Broker 或模型 API 中斷時，設備端仍需有本地保護和安全預設狀態。

關鍵原則：AIoT 的安全和可靠性必須由整條路徑共同負責：設備韌體、網路與 Broker、後端 service、API／MCP 權限，以及 Agent host 的核准流程。

十二、結語：不要只記名詞，要看清楚資料怎麼走

TCP 解決可靠的位元組傳輸；asyncio 讓 Python 在 I/O 等待期間更有效率地安排工作；HTTP／REST 提供一般軟體熟悉的請求與回應方式；FastAPI 幫助 Python 開發 HTTP 服務；MCP 讓 AI host 能探索並呼叫工具；Agent 根據任務選用工具；MQTT 則把設備訊息送到後端。把每一層的責任分清楚，再用共用服務層連接它們，就能從簡單的 TCP 範例逐步發展到 REST API、Agent 工具和 AIoT 應用。

系列文章：
Python asyncio TCP Server／Client｜ Python MCP Server／Client｜ Python Agent 呼叫 MCP Server｜ FastAPI REST API 與 MCP Server｜ FastAPI、MCP、AIoT 與 Agent

官方參考資料：IETF RFC 9293 TCP、Python asyncio Streams、FastAPI First Steps、Model Context Protocol、MCP Python SDK、OASIS MQTT 5.0。

Python 網路程式設計系列｜程式語言特性總整理

從五篇文章看懂 Python：串起 TCP、MCP、Agent、FastAPI 與 AIoT

從 asyncio TCP 的收發資料，到 FastAPI 建立 REST API、MCP Server 公布工具、Agent 選用工具，再到 AIoT 接收感測資料與控制設備，表面上看起來是五種不同技術，其實都能用同一套 Python 基礎概念來理解。這篇文章回到 Python 程式語言本身，整理函式、型別標註、裝飾器、非同步、模組、例外處理與資源管理，說明它們如何讓不同技術接在一起。

一、五篇文章，其實是一條逐步擴大的程式路徑

主題	程式要處理的問題	主要 Python 概念
asyncio TCP Server／Client	如何在網路連線等待資料時，繼續服務其他連線？	async def、await、coroutine、事件迴圈、StreamReader／StreamWriter
MCP Server／Client	如何用共同協定探索工具並傳送工具參數？	函式、型別標註、模組、非同步連線、序列化資料
Agent 呼叫 MCP Server	如何讓 Agent 理解任務並選擇工具？	物件組合、工具函式、裝飾器、非同步呼叫、例外處理
FastAPI REST API 與 MCP	如何同時服務一般程式與 AI host？	型別標註、裝飾器、資料模型、模組化、共用服務函式
FastAPI、MCP、Agent 與 AIoT	如何把感測資料、一般 API、Agent 工具和設備控制接起來？	資料結構、函式重用、非同步 I/O、事件驅動、資源生命週期

理解這些程式時，可以先分成三層：Python 語言負責表達資料與流程；框架或 SDK 負責 HTTP、MCP、TCP 等協定細節；應用程式則決定要提供什麼功能。FastAPI、MCP SDK 和 Agent SDK 不是 Python 語言本身，而是利用 Python 語言特性建構出來的工具。

二、函式是共用功能的基本單位

在 TCP、REST、MCP 和 AIoT 程式裡，函式都負責一段清楚的工作。把實際商業邏輯放進一般 Python 函式，再由不同入口呼叫，可以避免 REST 和 MCP 各自複製一套判斷規則。

def check_temperature(temperature: float) -> str:
    """依溫度回傳設備狀態建議。"""
    if temperature >= 35:
        return "溫度偏高，建議檢查通風設備"
    return "溫度在設定範圍內"

REST endpoint 可以呼叫 check_temperature() 並回傳 JSON；MCP tool 也可以呼叫相同函式，再把結果交給 Agent。這就是「入口不同、服務邏輯共用」的核心。

Python 的可組合性：函式可以被傳入其他函式、放在模組中匯入，也可以被框架註冊成路由或工具。框架接手協定工作，程式設計者仍然可以用普通函式描述領域功能。

三、型別標註讓程式意圖更清楚

例如 temperature: float 表示預期輸入是浮點數，-> str 表示預期回傳文字。型別標註讓 IDE、自動補全、靜態檢查工具和框架能了解程式介面。FastAPI 會讀取型別標註與 Pydantic 模型來驗證 HTTP 資料；MCP SDK 也會依工具函式的參數和型別建立輸入描述。

from pydantic import BaseModel, Field

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

def summarize(reading: SensorReading) -> str:
    return f"{reading.device_id}：{reading.temperature}°C"

SensorReading 不只說明欄位名稱，也把數值限制寫在資料模型裡。API 接收到不符合條件的資料時，FastAPI/Pydantic 可以回報驗證錯誤。要留意的是，Python 執行環境本身不會自動強制所有型別標註；真正的執行期驗證需要 Pydantic 或自行撰寫檢查。

四、裝飾器把一般函式註冊成路由或工具

裝飾器語法以 @ 開頭，常用來替函式加上框架提供的行為。函式本身仍是 Python 函式，裝飾器則告訴框架如何使用它。

# FastAPI：把函式註冊成 HTTP GET 路由
@app.get("/api/sensors/{device_id}")
async def read_sensor(device_id: str):
    return await sensor_service.latest(device_id)

# MCP SDK：把函式註冊成 Agent 可探索的工具
@mcp.tool()
def get_latest_temperature(device_id: str) -> str:
    """讀取指定設備的最新溫度。"""
    return sensor_service.temperature_text(device_id)

第一個裝飾器把函式和 HTTP 路徑關聯起來；第二個把函式及說明註冊到 MCP Server。Agent 可以透過 MCP Client 探索工具名稱與輸入欄位，再決定是否呼叫它。裝飾器讓宣告方式簡潔，但除錯時仍要知道函式何時被框架呼叫、參數由誰提供，以及錯誤會如何回傳。

五、async／await：等待 I/O 時讓出執行權

一般同步程式在等待網路回應時，執行緒通常停在該處。使用 async def 定義 coroutine，並在等待非同步 I/O 的地方使用 await，事件迴圈就能先安排其他等待中的工作。Python 官方文件說明 asyncio 用於以 async／await 撰寫並行程式，適合 I/O 密集的網路程式。

import asyncio

async def handle_client(reader, writer):
    try:
        data = await reader.readline()
        message = data.decode("utf-8").strip()
        writer.write(f"收到：{message}\n".encode("utf-8"))
        await writer.drain()
    finally:
        writer.close()
        await writer.wait_closed()

async def main():
    server = await asyncio.start_server(
        handle_client, "127.0.0.1", 8888
    )
    async with server:
        await server.serve_forever()

asyncio.run(main())

• async def 宣告非同步函式，呼叫後會產生 coroutine。
• await reader.readline() 表示目前工作要等待資料；等待期間事件迴圈可以處理其他連線。
• async with 管理 server 的生命週期；finally 確保連線結束時關閉 writer。
• asyncio.run() 建立事件迴圈並執行最上層 coroutine。

不是所有程式都要 async：asyncio 特別適合網路、檔案或資料庫等 I/O 等待很多的工作。CPU 密集運算、阻塞式函式若直接放進 async 函式，仍會卡住事件迴圈；需要改用非同步版本、執行緒或程序。citeturn21search0

六、模組與匯入讓專案各司其職

從單一檔案逐步發展成服務時，可以依責任拆成模組：TCP Server 管理連線，FastAPI 模組管理 HTTP 路由，MCP 模組註冊工具，Agent 模組負責任務流程，AIoT service 模組管理感測資料和設備規則。

project/
├── service.py       # 共用業務邏輯、資料存取
├── tcp_server.py    # asyncio TCP 入口
├── api_server.py    # FastAPI REST 入口
├── mcp_server.py    # MCP tools 入口
├── agent_client.py  # Agent／MCP Client
└── mqtt_adapter.py  # MQTT 設備訊息接收與命令發布

例如 from service import latest_reading 把共用函式匯入 API 或 MCP 模組。這種結構讓程式可以分別啟動、替換或測試；到了多程序或多台主機部署時，共用資料應放到資料庫或訊息系統，而不是期待不同程序共用 Python 記憶體。

七、資料結構與 JSON：把 Python 物件送過網路

網路協定傳送的是位元組，不是 Python 物件。常見流程是先把設備資料整理成 dict，轉成 JSON 字串，再編碼成 UTF-8 bytes。接收方則反向解碼和解析。REST、MCP 與 MQTT 都需要明確的資料格式。

import json

reading = {
    "device_id": "pond-01",
    "temperature": 28.6,
    "dissolved_oxygen": 6.4,
}

payload = json.dumps(reading, ensure_ascii=False)
wire_bytes = payload.encode("utf-8")

received_text = wire_bytes.decode("utf-8")
received_data = json.loads(received_text)
print(received_data["temperature"])

TCP 本身只提供連續的 bytes stream，不會替應用程式標記每一則訊息的邊界；文章中的 TCP 範例需要自行約定換行、固定長度或長度前綴等 framing 方式。HTTP 和 MCP 則在協定層定義了訊息如何包裝。理解「應用程式資料結構」和「網路傳輸格式」是兩個不同層次，可避免把 TCP 封包誤當成一筆完整 JSON 訊息。

八、循序圖：同一套 Python 服務，從資料到 Agent 工具

循序圖聚焦在使用端的兩種請求方式：一般程式透過 FastAPI 呼叫 REST endpoint；Agent 則透過 MCP Client 探索工具，再由 MCP Server 呼叫同一個 Python service。感測資料可由 MQTT adapter 先寫入共用儲存層。

循序圖重點不是讓一支 Python 程式包辦所有工作，而是讓不同介面透過明確的函式和資料結構合作。對一般程式來說，FastAPI 路由是明確入口；對 Agent 來說，MCP tool 是描述清楚的功能入口；底層服務仍維持可重用、可驗證。

九、Python 的資源管理與錯誤處理

網路連線、檔案、資料庫交易和伺服器工作階段都需要正確開啟和關閉。同步程式常用 with；非同步程式使用 async with。在 asyncio TCP 中，連線關閉可放入 finally；在 FastAPI 整合 MCP 時，應用程式 lifespan 可負責啟動和結束 MCP session manager。Python 的 contextlib 提供 asynccontextmanager，用來建立非同步 context manager。citeturn21search10

錯誤處理也應依邊界分層：輸入格式錯誤回傳驗證訊息；網路中斷記錄連線錯誤並重試；設備命令逾時回報未確認狀態；Agent 無法使用模型時顯示可理解的錯誤。不要把密碼、API key 或內部例外堆疊直接回傳給使用者。

十、從語言特性走到完整系統

1. 先練函式和資料結構：用普通函式讀取、驗證和處理感測資料。
2. 再練 TCP 與 JSON：理解 bytes、編碼、訊息邊界和連線生命週期。
3. 加入 asyncio：把等待網路資料的程式改成 coroutine，觀察多連線下事件迴圈如何切換工作。
4. 建立 FastAPI API：用資料模型和裝飾器將共用函式提供給網站或 App。
5. 建立 MCP tools：用型別標註和工具描述，讓 MCP Client 能探索服務能力。
6. 接上 Agent：讓模型依任務選工具；把權限、核准與業務規則留在 host 和服務層。
7. 延伸至 AIoT：用 MQTT 接設備資料，將讀取與控制功能包在受控服務層，讓 REST 和 MCP 共用。

安全提醒：AI Agent 的工具呼叫可能造成外部副作用。對泵浦、加熱器、投餌機、電鎖等設備，應採用明確的允許清單、範圍檢查、身分授權、人工確認、操作紀錄與設備端安全互鎖。語言模型不能取代設備控制器的即時保護。

十一、結語

這五篇文章展示的是同一條 Python 學習路徑：從函式與資料開始，再透過 asyncio 處理網路等待，用 FastAPI 和 MCP SDK 將功能轉成不同介面，最後由 Agent 選用工具並連到 AIoT 設備。真正讓系統容易擴充的，是把 Python 語言特性用在清楚的責任分工上：函式承載規則、型別標註說明介面、裝飾器完成註冊、非同步處理 I/O、模組整理專案、資料模型驗證輸入、context manager 管理資源。學會這些基礎後，換成新的 Python 框架或協定，也更容易看懂程式背後的運作方式。

系列文章：
Python asyncio TCP Server／Client｜ Python MCP Server／Client｜ Python Agent 呼叫 MCP Server｜ FastAPI REST API 與 MCP Server｜ FastAPI、MCP、AIoT 與 Agent

Python 官方文件： asyncio、asyncio streams、typing、contextlib。

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 與協定會更新，實作前請依目前官方文件確認版本與部署設定。

Python 網路程式設計｜REST API × MCP

FastAPI REST API 與 MCP Server 比較：同一功能，兩種介面

FastAPI REST API 與 MCP Server 都能把 Python 功能提供給其他程式使用，但面向的使用者和互動方式不同。本文用同一個 echo_message 功能，分別建立 REST endpoint 與 MCP tool，說明如何選用、如何共用商業邏輯，以及兩者如何在同一個專案中合作。

一、先看核心差異：它們解決不同層次的問題

FastAPI 是用來建立 HTTP API 的 Python Web 框架；REST 是一種以資源、HTTP 方法與狀態碼組織 API 的設計風格。MCP 則是讓 AI 應用以標準方式探索和使用工具、資源及提示的協定。MCP 可以使用 Streamable HTTP 傳輸，因此它也會經由 HTTP 網路通訊；但 MCP 訊息與 REST endpoint 的請求格式和互動流程並不相同。

瀏覽器／手機／一般程式→FastAPI REST endpoint→Python 功能

AI Agent／MCP host→MCP Client→MCP Server tool→Python 功能

一句話記憶：REST API 通常由呼叫端指定要呼叫哪個 URL 和 HTTP 方法；MCP Server 會描述可用的工具，讓 MCP Client／Agent 能先探索工具，再以工具名稱與參數呼叫。FastAPI 會依 API 定義產生 OpenAPI schema 和互動文件；MCP SDK 則負責 MCP 工具、資源、提示與協定傳輸。FastAPI 官方文件｜MCP Python SDK 官方文件

二、比較表

比較面向	FastAPI REST API	MCP Server
主要目的	讓網站、手機 App、IoT 後端或其他服務呼叫應用程式功能。	讓支援 MCP 的 AI host／Agent 以共同協定使用工具、資源和提示。
呼叫方式	呼叫端指定 URL、HTTP 方法和 JSON 請求內容，例如 POST /echo。	Client 先探索工具，再以工具名稱和參數發出 MCP tools/call。
介面描述	常以 OpenAPI 描述路由、參數、回應和狀態碼。	工具以名稱、說明和輸入 schema 描述；也可提供 resources 和 prompts。
典型呼叫端	瀏覽器、前端、手機程式、一般 Python／JavaScript client、後端服務。	支援 MCP 的 Agent、AI 桌面應用、IDE 或自製 MCP Client。
是否需要 AI	不需要。呼叫端按程式邏輯指定 API。	MCP 本身不要求 server 內有模型；通常由 Agent／host 判斷要不要呼叫工具。
常見傳輸	HTTP／HTTPS。	stdio、Streamable HTTP 等標準傳輸。
適合情境	建立應用程式後端、資料 CRUD、前端服務、公開或內部 API。	讓 AI 工具使用者能探索並呼叫既有系統功能。

三、先寫共用的 Python 功能

無論呼叫端是 REST client 還是 MCP Client，建議把真正的業務邏輯放在共同模組，而不是各寫一份。這樣兩種介面都呼叫同一個函式，結果和規則才不會逐漸不一致。

# service.py

def echo_message(message: str) -> str:
    """回傳 server 收到的訊息。"""
    return f"Server received: {message}"

service.py 不處理 HTTP，也不處理 MCP。它只負責應用程式本身的功能。REST 和 MCP 是兩個不同的入口，最後都委派給這個函式。

四、用 FastAPI 建立 REST API

# rest_server.py
from fastapi import FastAPI
from pydantic import BaseModel

from service import echo_message

app = FastAPI(title="Echo REST API")

class EchoRequest(BaseModel):
    message: str

class EchoResponse(BaseModel):
    reply: str

@app.post("/echo", response_model=EchoResponse)
def echo(request: EchoRequest) -> EchoResponse:
    reply = echo_message(request.message)
    return EchoResponse(reply=reply)

REST API 程式解說

1. FastAPI() 建立 Web API 應用程式。
2. EchoRequest 描述 client 送來的 JSON 欄位；EchoResponse 描述 server 回覆格式。FastAPI 會使用型別資訊驗證資料並產生 API 文件。
3. @app.post("/echo") 建立 POST /echo endpoint。REST client 知道路徑和方法後即可呼叫。
4. endpoint 把訊息交給 echo_message()，再以 JSON 回傳。

安裝並啟動：

python -m pip install fastapi uvicorn
uvicorn rest_server:app --reload --host 127.0.0.1 --port 8000

可用命令列呼叫 REST API：

curl -X POST "http://127.0.0.1:8000/echo" `
  -H "Content-Type: application/json" `
  -d "{\"message\":\"Hello REST\"}"

預期回傳 JSON：

{"reply":"Server received: Hello REST"}

FastAPI 也會依 OpenAPI 規格產生互動式文件，預設可到 http://127.0.0.1:8000/docs 查看 endpoint 和測試請求。

五、用 MCP SDK 建立 MCP Server

# mcp_server.py
from mcp.server import MCPServer

from service import echo_message

mcp = MCPServer(
    "Echo MCP Server",
    instructions="提供 echo_message 工具，回傳收到的文字。",
)

@mcp.tool()
def echo_message_tool(message: str) -> str:
    """回傳 server 收到的訊息。"""
    return echo_message(message)

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=8001,
    )

MCP Server 程式解說

1. MCPServer(...) 建立 MCP Server，並可提供用途說明。
2. @mcp.tool() 把 Python 函式註冊為 MCP 工具。工具名稱、說明和參數 schema 會提供給 MCP Client 探索。
3. 工具函式呼叫共用的 echo_message()，因此 REST 與 MCP 介面使用相同業務規則。
4. mcp.run(transport="streamable-http") 以 Streamable HTTP 啟動服務。本例使用 8001 埠，endpoint 為 http://127.0.0.1:8001/mcp。

安裝 MCP Python SDK v2 後啟動：

python -m pip install "mcp[cli]>=2,<3"
python mcp_server.py

MCP Client 連線後可先用 tools/list 取得工具清單，再以 tools/call 呼叫工具。MCP 的工具定義包含名稱、描述及輸入 schema；MCP Python SDK 支援 stdio、Streamable HTTP 等傳輸。citeturn15search1turn15search2turn15search4

六、同一個功能如何被兩種 client 使用？

REST 呼叫	MCP 呼叫
送出 POST /echo。	從工具清單找到 echo_message_tool。
在 JSON body 傳入 {"message":"Hello"}。	在工具 arguments 傳入 {"message":"Hello"}。
取得 HTTP status code 與 JSON response body。	取得 MCP tool result，可能包含文字或結構化內容。
通常由程式碼明確指定要呼叫哪個 URL。	Agent 可讀取工具說明，依使用者任務選擇要呼叫哪個工具。

重要觀念：MCP Server 不是「內建 AI 的 Server」。它提供可被呼叫的工具；Agent 或 MCP host 負責理解任務、選擇工具，再呼叫 MCP Server。若一般程式已清楚知道要做什麼，直接呼叫 REST API 通常更簡單。

七、循序圖：REST 與 MCP 的呼叫流程

循序圖由上往下表示時間先後。左半部是一般程式直接呼叫 REST；右半部是 Agent 透過 MCP Client 探索工具、選擇工具，再由 MCP Server 執行。兩條流程最後都會呼叫同一個共用服務函式。

1. REST 流程：Client 依 API 文件直接送出 POST /echo，FastAPI 驗證 JSON，執行 endpoint，再回傳 HTTP 狀態碼與 JSON。
2. MCP 初始化：MCP Client 與 Server 建立連線並協商協定能力。實際傳輸可能是 stdio 或 Streamable HTTP。
3. 探索工具：Client 使用 tools/list 取得 MCP Server 公布的工具名稱、描述和輸入 schema，Agent 因此能判斷有哪些能力可用。
4. 呼叫與回覆：Agent 選定工具後，Client 用 tools/call 傳入工具名稱及參數；Server 執行工具並回傳 tool result，Agent 再把結果整理成給使用者的回答。

REST 的主要往返通常是「呼叫 endpoint、收到 HTTP response」；MCP 多了協定初始化與工具探索的階段。若 Client 已經知道固定要呼叫哪個 API，REST 直接明確；若 Agent 要根據任務從可用工具中選擇，MCP 的工具描述和探索流程就有幫助。

八、可以讓 REST 與 MCP 共存嗎？

可以。常見架構是把業務功能放在共用服務層，再分別提供 REST 與 MCP 入口：

Web／手機 App ──HTTP POST──> FastAPI /echo ─┐
                                             ├──> service.echo_message()
Agent ──MCP tools/call──────> MCP Server ────┘

這樣網站、手機 App、其他後端程式可以使用 REST API；AI Agent 則使用 MCP 工具。兩者共用驗證後的業務邏輯、資料庫存取和紀錄程式。初學時分別跑在不同埠號較容易觀察；系統成熟後，再依部署架構考慮掛載到同一個 ASGI 應用程式或透過反向代理管理。

九、何時選哪一種？

• 建立網站或手機 App 後端：先選 FastAPI REST API。它的 URL、HTTP 方法和 JSON 格式適合一般 client，OpenAPI 文件也方便測試與產生 client 程式碼。
• 把資料或功能提供給支援 MCP 的 Agent：建立 MCP Server，明確描述可用工具、參數與用途。
• 同時有一般 client 和 Agent：保留 REST 與 MCP 兩個入口，共用同一套 service layer。
• 純 REST API 要讓模型呼叫：也能用 Agent SDK 的 function tool 封裝 API 呼叫，不一定非要自己架 MCP；若希望多種 MCP host 以共同協定探索並使用工具，MCP 會較合適。
• 涉及敏感操作：REST 和 MCP 都要做身分驗證、授權、輸入檢查與紀錄。不要因為使用 MCP 就假設工具自動安全；只開放必要能力，修改或刪除等操作應加上核准機制。

十、結論

FastAPI REST API 和 MCP Server 不需要二選一。REST API 面向一般軟體 client，讓呼叫端依 URL 和 HTTP 方法呼叫服務；MCP Server 面向 AI host／Agent，讓 client 透過標準協定探索和呼叫工具。實務上，可把業務邏輯寫一次，再同時提供 REST 與 MCP 介面。選擇時先看呼叫端是一般應用程式還是 AI Agent，再決定要暴露哪一種介面。

參考資料：FastAPI First Steps、FastAPI OpenAPI Docs、MCP Python SDK、MCP Tools Specification。範例使用 FastAPI 和 MCP Python SDK 的官方文件所述介面；套件版本更新時，請以官方文件為準。

Python Agent × MCP｜實作教學

讓 Python Agent 呼叫 MCP Server：以 OpenAI Agents SDK 為例

MCP Server 已經提供工具後，還可以讓 AI Agent 自己判斷何時呼叫嗎？可以。本文將既有的 Python MCP Echo Server 接到 OpenAI Agents SDK，讓 Agent 理解使用者要求、選擇 MCP 工具、傳入參數，再根據工具結果回答。

一、從直接呼叫到 Agent 自行選工具

直接呼叫 MCP 的程式會明確寫出 client.call_tool("echo_message", ...)，由程式決定要呼叫哪個工具。Agent 的作法不同：我們把 MCP Server 提供的工具交給 Agent，再以自然語言描述任務。Agent 會依照指示與使用者要求選擇工具，並將工具結果整理成回答。

使用者提出要求→Agent 判斷可用工具→MCP Client 呼叫工具→MCP Server 執行→Agent 回覆

架構要點：本例使用 OpenAI Agents SDK 在本機執行 Agent 程式。Agent 透過 MCP Streamable HTTP 連到本機 server；模型負責理解需求與選擇工具，Agents SDK 負責把工具呼叫送到 MCP Server 並帶回結果。OpenAI 官方 Agents SDK 支援連線至 Streamable HTTP MCP Server。官方 MCP 整合文件

這仍是 MCP 工具呼叫，並不代表模型直接連入您的電腦。執行中的 Agent 程式負責連線和轉送工具呼叫。因此，Agent 程式必須能連到 MCP Server 的 URL。

二、Agent Client 與 MCP Server 的運作循序圖

下圖呈現一次完整的工具呼叫。流程包含使用者提出任務、Agent 取得工具資訊、模型選擇工具、MCP Server 執行工具，以及 Agent 把結果交回模型整理回答。

實線表示請求或呼叫；虛線表示回覆。工具清單可能由 SDK 快取；圖中列出首次連線時的探索流程。

循序圖逐步解讀

1. 使用者提出任務：例如「請把『你好』傳給 echo_message」。Agent Client 將任務交給 Agent 執行。
2. Client 連接 MCP Server：Agents SDK 透過 Streamable HTTP 連至 MCP endpoint，完成 MCP 初始化並取得 Server 資訊與能力。
3. 探索工具：Client 讀取 MCP Server 提供的工具清單與參數規格。Agent 才能知道工具名稱、用途和需要哪些輸入。
4. 模型選擇工具：Agent Client 把使用者任務和工具資訊交給 OpenAI 模型 API。模型判斷應呼叫 echo_message，並提供 message 參數。
5. Client 呼叫工具：Agent Client 按模型指定的工具名稱和參數，送出 MCP tools/call 請求。
6. Server 執行函式：MCP Server 執行 Python 的 echo_message()，並將函式結果回傳給 Agent Client。
7. 模型產生回答：Client 把工具結果交回模型，模型依結果組織最後回答，再由 Agent Client 顯示給使用者。

責任分工：模型負責理解需求、選擇工具和整理回答；MCP Server 負責執行實際工具程式。模型不會直接連入使用者電腦。Agent Client 負責模型 API 與 MCP Server 之間的協調，因此兩段連線都必須正常。

三、先準備 MCP Server

沿用前一篇建立的 mcp_server.py。它透過 Streamable HTTP 在本機埠號 8000 提供下列工具：

• echo_message(message)：把訊息記錄在 server 終端機，並回傳文字。
• get_server_info()：回傳 server 名稱、主機、埠號及傳輸方式。

Server 需使用 Python MCP SDK v2 API，形式如下：

from mcp.server import MCPServer

HOST = "127.0.0.1"
PORT = 8000
mcp = MCPServer("Python MCP Echo Server")

@mcp.tool()
def echo_message(message: str) -> str:
    """Return the message sent by the client."""
    print(f"Client message: {message}")
    return f"Server received: {message}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host=HOST, port=PORT)

此 server 的 MCP endpoint 是 http://127.0.0.1:8000/mcp。SDK 透過 MCP 工具定義產生名稱、說明及輸入規格，Agents SDK 連線後便能取得這些工具。

四、安裝 Agent 所需套件

在同一個 Python 虛擬環境安裝 OpenAI Agents SDK 與 MCP SDK：

python -m pip install openai-agents "mcp>=2,<3"

OpenAI Agents SDK 的 MCP 說明支援 MCP Python 套件 v1 與 v2；此範例的 server 是以 v2 API 撰寫，因此以 mcp>=2,<3 限定主要版本。實際套件相依性請以安裝時的套件解析結果為準。

五、Agent Client 程式

import asyncio

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

MCP_URL = "http://127.0.0.1:8000/mcp"

async def main() -> None:
    async with MCPServerStreamableHttp(
        name="Python MCP Echo Server",
        params={"url": MCP_URL},
        cache_tools_list=True,
    ) as mcp_server:
        agent = Agent(
            name="MCP 示範助理",
            instructions=(
                "你可以使用 MCP Server 提供的工具。"
                "當使用者要求傳送或回覆文字時，使用 echo_message 工具。"
                "根據工具回傳結果，用繁體中文簡潔回答。"
            ),
            mcp_servers=[mcp_server],
        )

        result = await Runner.run(
            agent,
            "請呼叫 echo_message 工具，傳送：你好，這是 Agent 透過 MCP 傳來的訊息。",
        )
        print(result.final_output)

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

六、逐段解說 Agent 程式

1. from agents import Agent, Runner 匯入 Agent 定義與執行器。Agent 設定名稱、行為指示及工具來源；Runner 執行一輪或多輪的 Agent 工作。
2. MCPServerStreamableHttp 是 Agents SDK 提供的 MCP 連線介面。它會連上指定的 MCP endpoint，取得工具清單，並在 Agent 呼叫工具時送出 MCP 請求。
3. params={"url": MCP_URL} 指向既有 server。若 Agent 程式與 server 在同一台電腦，可使用 127.0.0.1；若在不同電腦，請改用 server 的區網 IP 或可安全存取的遠端網址。
4. async with ... as mcp_server 管理連線生命週期。離開區塊後會清理 MCP 連線。
5. cache_tools_list=True 代表工具定義在連線期間可快取，減少重複查詢；若 server 的工具會頻繁變動，則應考慮不快取或重新整理工具清單。
6. mcp_servers=[mcp_server] 把 MCP Server 提供的工具交給 Agent 使用。Agent 會根據使用者的要求和 instructions 選擇工具。
7. Runner.run(agent, "...") 提交一項自然語言任務。這裡刻意明確要求呼叫 echo_message，方便觀察完整流程。
8. result.final_output 是 Agent 完成工具呼叫後產生的最後回答。

七、啟動順序與預期結果

1. 先在第一個終端機啟動 MCP Server：python mcp_server.py。
2. 確認 server 正在監聽本機 8000 埠。
3. 在第二個終端機設定 API 金鑰，並執行 Agent 程式。

# Windows PowerShell：在目前終端機工作階段設定
$env:OPENAI_API_KEY = "你的有效 API 金鑰"
python mcp_agent.py

成功時，MCP Server 終端機會記錄 Agent 傳給 echo_message 的內容；Agent 終端機會印出依工具結果整理的回覆。模型的實際回覆文字可能略有不同。

八、常見錯誤：API 金鑰與額度

錯誤訊息	代表狀況	處理方式
401 invalid_api_key	Agent 呼叫模型時使用的 API 金鑰無效、錯誤或程式讀到另一把金鑰。	確認 OPENAI_API_KEY 在執行 Agent 的同一個終端機中設定，並確認 API Platform 的金鑰有效。不要把金鑰貼到對話、程式碼或公開儲存庫。
429 insufficient_quota credit_balance_exhausted	金鑰已辨識，但該 API 組織沒有可用點數，或帳務額度不足。	前往 API Platform 帳務頁檢查正確組織的餘額、付款方式和支出限制；購買點數後等待帳務頁更新，再重試。
連線錯誤／Connection refused	Agent 無法連到 MCP Server endpoint。	確認 server 已啟動，URL 路徑是 /mcp，埠號相符；若跨電腦連線，檢查 IP、防火牆及網路路由。

兩段不同的連線：Agent 先連 OpenAI API 取得模型推理服務，同時由本機 Python 程式連 MCP Server 執行工具。API 的 401/429 與 MCP endpoint 的連線錯誤，是不同位置的問題。ChatGPT 訂閱與 API 用量分別計費；使用 ChatGPT Plus 不會自動提供 API 點數。可參閱 OpenAI 計費說明 與 API 預付額度說明。

九、適用情境與安全原則

• 適合：用自然語言要求 Agent 搜尋資料、查詢系統、整理文件，或執行由 MCP Server 明確公開的功能。
• 工具設計：工具名稱和 docstring 要清楚描述用途、參數與回傳內容，避免模糊工具讓 Agent 難以選擇。
• 權限：只公開 Agent 任務需要的工具；具有刪除、寄送、付款或修改資料等副作用的工具，應設計人工確認或權限檢查。
• 遠端服務：不要把開發用的無驗證 HTTP endpoint 直接暴露到公網。遠端部署時應使用 HTTPS、身分驗證、授權和適當的網路限制。
• 金鑰：使用環境變數或受控的秘密儲存服務，不要把 API 金鑰寫進原始碼或貼到錯誤訊息中。

OpenAI Agents SDK 的官方 MCP 文件也提醒，MCP 工具可能存取資料或執行操作，因此應只連接可信任的 server，採用最小必要權限，並為敏感工具設定核准流程。官方 MCP 整合與安全說明

結語

將 MCP Server 接到 Agent 後，開發者提供工具與行為規則，Agent 則負責理解任務並選用工具。前一個直接 MCP Client 範例由程式固定呼叫工具；Agent 版本則加入模型推理，能依自然語言需求選擇工具並組織回覆。遇到問題時，可先判斷錯誤發生在模型 API、MCP 傳輸，還是工具執行階段，再針對該層排查。

參考資料：OpenAI Agents SDK：Model Context Protocol、管理 ChatGPT 與 API 計費、API 預付計費。程式介面依官方文件撰寫；套件與服務細節可能隨版本更新。