從五篇文章看懂 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。這就是「入口不同、服務邏輯共用」的核心。
三、型別標註讓程式意圖更清楚
例如 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。
六、模組與匯入讓專案各司其職
從單一檔案逐步發展成服務時，可以依責任拆成模組：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。citeturn21search10
錯誤處理也應依邊界分層：輸入格式錯誤回傳驗證訊息；網路中斷記錄連線錯誤並重試；設備命令逾時回報未確認狀態；Agent 無法使用模型時顯示可理解的錯誤。不要把密碼、API key 或內部例外堆疊直接回傳給使用者。
十、從語言特性走到完整系統
- 先練函式和資料結構：用普通函式讀取、驗證和處理感測資料。
- 再練 TCP 與 JSON：理解 bytes、編碼、訊息邊界和連線生命週期。
- 加入 asyncio：把等待網路資料的程式改成 coroutine，觀察多連線下事件迴圈如何切換工作。
- 建立 FastAPI API：用資料模型和裝飾器將共用函式提供給網站或 App。
- 建立 MCP tools：用型別標註和工具描述，讓 MCP Client 能探索服務能力。
- 接上 Agent：讓模型依任務選工具；把權限、核准與業務規則留在 host 和服務層。
- 延伸至 AIoT：用 MQTT 接設備資料，將讀取與控制功能包在受控服務層，讓 REST 和 MCP 共用。
十一、結語
這五篇文章展示的是同一條 Python 學習路徑：從函式與資料開始，再透過 asyncio 處理網路等待，用 FastAPI 和 MCP SDK 將功能轉成不同介面，最後由 Agent 選用工具並連到 AIoT 設備。真正讓系統容易擴充的，是把 Python 語言特性用在清楚的責任分工上：函式承載規則、型別標註說明介面、裝飾器完成註冊、非同步處理 I/O、模組整理專案、資料模型驗證輸入、context manager 管理資源。學會這些基礎後，換成新的 Python 框架或協定，也更容易看懂程式背後的運作方式。
沒有留言:
張貼留言