網際網路應用系列｜從 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 預付計費。程式介面依官方文件撰寫；套件與服務細節可能隨版本更新。

Python 網路程式設計｜MCP 實作

用 Python MCP 實作 Server／Client：從工具註冊到互動呼叫

前一篇以 TCP socket 與 asyncio 建立 echo server/client。本文改用 Model Context Protocol（MCP），讓 server 將功能註冊為工具，client 透過標準協定探索並呼叫工具。範例採 Streamable HTTP，適合兩個獨立程式在本機或區網通訊。

一、MCP 和 TCP 的關係

TCP 是傳輸層通訊協定，負責把位元組可靠地送到對方。MCP 則定義 client 和 server 之間的互動方式，例如初始化、列出工具、呼叫工具，以及回傳內容。MCP 仍需要傳輸方式承載資料；本例使用 Streamable HTTP，底層網路仍會使用 TCP。

本例的差別：一般 echo TCP 程式通常由 client 送一串文字，server 回傳文字。MCP client 會先連線並查詢 server 提供的工具，再用工具名稱與參數呼叫功能。SDK 處理協定訊息、參數驗證及結果格式。

Python MCP Client→Streamable HTTP／TCP→MCP Server→echo_message 工具

官方 Python SDK 目前文件以 v2 為主，要求 Python 3.11 以上，提供 stdio、Streamable HTTP 等傳輸方式。本篇選擇 Streamable HTTP，因為 server 會監聽一個本機埠號，client 可以用 URL 連線。官方 SDK 文件｜伺服器執行方式｜Client 傳輸方式

二、安裝環境

建立資料夾，放入 mcp_server.py、mcp_client.py 與 requirements.txt。建議使用 Python 3.11 以上版本。

python -m venv .venv

# Windows PowerShell
.venv\\Scripts\\Activate.ps1

# macOS／Linux
source .venv/bin/activate

python -m pip install -r requirements.txt

requirements.txt 的內容：

mcp[cli]>=2.0.0,<3.0.0

其中 mcp 是官方 Python SDK；[cli] 會一併安裝開發用命令列工具。若使用新版 Python 環境管理工具，也可依官方文件以 uv add "mcp[cli]" 安裝。

三、撰寫 MCP Server

from mcp.server import MCPServer

HOST = "127.0.0.1"
PORT = 8000

mcp = MCPServer(
    "Python MCP Echo Server",
    instructions="Provides an echo tool and basic server information.",
)

@mcp.tool()
def echo_message(message: str) -> str:
    """Return the message supplied by the client and log it on the server."""
    print(f"Client message: {message}")
    return f"Server received: {message}"

@mcp.tool()
def get_server_info() -> dict[str, str | int]:
    """Return the name and listening address of this MCP server."""
    return {
        "name": "Python MCP Echo Server",
        "host": HOST,
        "port": PORT,
        "transport": "Streamable HTTP",
    }

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

Server 程式逐段解說

1. from mcp.server import MCPServer 匯入 SDK 的高階 server 類別。它負責處理 MCP 的初始化、工具清單、呼叫與結果傳輸。
2. MCPServer(...) 建立 server。第一個參數是供 client 辨識的名稱；instructions 提供 server 的用途說明。
3. @mcp.tool() 將下一個 Python 函式登記成 MCP 工具。函式名稱成為工具名稱，型別提示用來建立參數規格，docstring 則是工具用途說明。
4. echo_message(message: str) 接收文字、在 server 終端機印出紀錄，再回傳回應。SDK 會把回傳值包裝成 MCP 工具結果。
5. get_server_info() 示範回傳結構化資料。client 可以從工具結果取得名稱、位址、埠號和傳輸方式。
6. mcp.run(transport="streamable-http", ...) 啟動 HTTP 傳輸。預設 MCP endpoint 是 /mcp，所以本機網址為 http://127.0.0.1:8000/mcp。

stdio 與 HTTP：官方 SDK 預設可用 stdio，適合 MCP host 啟動本機子程序；本範例明確指定 Streamable HTTP，因為我們要分別啟動 server 和 client，透過 URL 連線。官方文件將 Streamable HTTP 列為部署型傳輸方式。

四、撰寫 MCP Client

import anyio
from mcp import Client
from mcp.types import TextContent

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

def show_result(result) -> None:
    if result.is_error:
        print("MCP tool returned an error:")
    for block in result.content:
        if isinstance(block, TextContent):
            print(block.text)
    if result.structured_content is not None:
        print(result.structured_content)

async def main() -> None:
    async with Client(SERVER_URL) as client:
        print(f"Connected to: {SERVER_URL}")

        tools = await client.list_tools()
        print("Available tools:", ", ".join(tool.name for tool in tools.tools))

        info = await client.call_tool("get_server_info", {})
        show_result(info)

        print("Enter a message to call echo_message. Type /quit to exit.")
        while True:
            message = await anyio.to_thread.run_sync(input, "> ")
            if message.strip() == "/quit":
                break
            result = await client.call_tool(
                "echo_message",
                {"message": message},
            )
            show_result(result)

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

Client 程式逐段解說

1. Client(SERVER_URL) 依 URL 選用 Streamable HTTP 傳輸。真正建立連線是在進入 async with 時；離開區塊時 SDK 會關閉連線。
2. await client.list_tools() 向 server 查詢已註冊工具。這是 MCP 相較自訂 echo 協定的關鍵：client 能探索工具名稱與輸入規格。
3. await client.call_tool("get_server_info", {}) 呼叫不需參數的工具；第二個參數是工具輸入字典。
4. await client.call_tool("echo_message", {"message": message}) 以工具名稱與參數送出請求。SDK 負責將呼叫轉成 MCP 訊息並交給 HTTP 傳輸。
5. result.content 是內容區塊清單。範例只列印文字區塊，因此先用 isinstance(block, TextContent) 確認型別。
6. result.is_error 可判斷工具是否回報錯誤；result.structured_content 則提供適合程式讀取的結構化結果。
7. input() 是同步阻塞函式。範例以 anyio.to_thread.run_sync() 在工作執行緒讀取鍵盤，避免互動輸入直接卡住非同步流程。

五、啟動與操作

1. 開啟第一個終端機，啟用虛擬環境後執行 python mcp_server.py。server 會在 127.0.0.1:8000 提供 MCP endpoint。
2. 另開第二個終端機，使用同一個環境執行 python mcp_client.py。
3. client 會列出工具、取得 server 資訊，接著等待輸入訊息。
4. 輸入文字後按 Enter，client 呼叫 echo_message，server 終端機會顯示收到的文字，client 也會看到回傳內容。
5. 輸入 /quit 結束 client；在 server 終端機按 Ctrl+C 停止 server。

終端機 1：python mcp_server.py
終端機 2：python mcp_client.py
> Hello MCP
Server received: Hello MCP
> /quit

六、MCP 和一般 TCP／asyncio 的比較

面向	一般 TCP socket	asyncio TCP	MCP
主要工作	自行定義資料格式並收發位元組。	用協程管理非同步 TCP I/O，資料格式仍要自行定義。	以標準化方式描述工具、參數與結果。
連線傳輸	TCP socket。	TCP socket。	可使用 stdio、Streamable HTTP 等標準傳輸；本範例使用 HTTP/TCP。
工具探索	通常自行設計命令或文件。	通常自行設計命令或文件。	client 可列出 server 提供的工具及輸入 schema。
適合情境	學習 TCP、簡單客製化協定、低階網路控制。	大量 I/O 連線、希望以 async 管理 socket。	讓 AI host 或其他 MCP client 以共同介面使用工具、資源與提示。
需留意	要處理封包邊界、編碼、錯誤與多連線。	仍要處理訊息格式，且要避免阻塞事件迴圈。	SDK 會處理協定，但要選對傳輸、安裝相容 SDK，並設計工具權限與輸入驗證。

因此，MCP 不是另一種 TCP，也不是把 socket API 換個名稱。它在 TCP 或 stdio 之上定義可互通的應用層協定。若目標是理解 socket 和網路封包，先學一般 TCP；若目標是把 Python 功能提供給支援 MCP 的 host 或 agent，MCP 會提供更一致的工具介面。

七、常見問題與安全注意事項

• client 連不上：確認 server 已啟動，URL 使用 /mcp 路徑，且兩端埠號一致。
• 找不到 mcp 套件：確認安裝套件的 Python 環境與執行程式的環境相同。
• 為什麼不用自訂 JSON：MCP SDK 已處理協定初始化、訊息格式和工具呼叫；自行用 socket 傳 JSON 不會自動成為 MCP server。
• 連區網：可將 server host 設為 0.0.0.0，client 改連 server 電腦的區網 IP，並設定防火牆。對外提供服務時應採用 HTTPS、驗證與授權。
• 工具安全：正式工具要驗證輸入、限制可執行的操作與存取範圍。不要把未驗證且有敏感副作用的工具直接開放給不受信任的 client。

結語

使用 MCP 後，server 不再只接收一串自訂文字，而是公開具名稱、描述和參數規格的工具；client 能先探索工具，再呼叫指定功能。Streamable HTTP 讓兩個獨立程式可透過 URL 通訊，而 MCP SDK 負責協定細節。這讓範例從「傳送文字」進一步成為「用標準介面呼叫服務功能」。

官方參考：MCP Python SDK、Running your server、The Client、Client transports。本文範例以官方文件所載 MCP Python SDK v2 API 編寫。

Python 網路程式設計｜入門到實作

Python asyncio 與一般 TCP 程式有何不同？

兩者都使用 TCP 傳送資料，差別在程式等待網路回應時，如何安排其他工作。本文用同步 socket 與 asyncio Stream API 建立一組 server/client，逐段說明連線、訊息收發、多用戶端處理，以及各自的優缺點。

一、先釐清：TCP 是通訊協定，asyncio 是程式設計方式

TCP 負責在兩台電腦之間建立可靠的位元組串流。它會處理資料順序、重傳等傳輸工作。無論使用一般的 socket，或使用 asyncio，底層都可以是 TCP。

一般 TCP 程式常使用阻塞式 socket：呼叫 accept()、recv() 或 sendall() 時，程式可能停下來等待作業系統完成工作。asyncio 則用事件迴圈和協程管理等待中的 I/O；當某個連線等待資料時，事件迴圈能轉去處理其他連線。

容易混淆的地方：「一般 TCP 程式」不一定只支援一個用戶端。它可以透過執行緒或多進程服務多個連線。本文先以單一連線的阻塞式版本作為基準，再說明如何擴充；asyncio 範例則直接示範多連線協程處理。

client 傳送位元組→TCP 傳輸→server 收到位元組→server 回覆

二、阻塞式與非同步的執行差異

項目	阻塞式 socket	asyncio
等待資料	recv() 等待時，該執行緒暫停。	await reader.readline() 等待時，協程讓出控制權；事件迴圈可執行其他工作。
並行處理連線	常見作法是每個 client 建立一個執行緒，或使用 selector 等機制。	通常以一個事件迴圈管理許多 I/O 協程，不必替每個連線各開一個執行緒。
程式閱讀	基本 socket 呼叫直觀，適合認識 TCP API。	需要理解 async、await、事件迴圈及協程取消。
適用工作	少量連線、簡單工具、短小服務。	大量網路等待、聊天服務、I/O 密集型應用。

asyncio 的「非同步」不表示 Python 程式會自動同時使用多個 CPU 核心。一般協程在同一個事件迴圈中輪流執行；當協程遇到可等待的 I/O 才讓出控制權。CPU 密集工作若長時間不讓出控制權，仍可能卡住整個事件迴圈。

三、一般 TCP 版本：阻塞式 server/client

以下範例使用換行符號 \n 作為一則訊息的結尾。TCP 傳送的是連續位元組，不會自動替應用程式標示每則訊息的邊界，因此雙方要自行約定訊息格式。

一般 TCP server

import socket

HOST = "127.0.0.1"
PORT = 8888

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen()
    print(f"Server listening on {HOST}:{PORT}")

    conn, address = server.accept()  # 阻塞：等第一個 client 連線
    with conn:
        print("Connected:", address)
        conn.sendall(b"Connected. Type /quit to leave.\n")
        buffer = b""

        while True:
            chunk = conn.recv(1024)  # 阻塞：等資料抵達
            if not chunk:
                break
            buffer += chunk

            # TCP 是位元組串流；可能一次收到半行或多行
            while b"\n" in buffer:
                line, buffer = buffer.split(b"\n", 1)
                message = line.decode("utf-8", errors="replace").rstrip("\r")
                if message == "/quit":
                    break
                print(f"{address}: {message}")
                conn.sendall(f"Server received: {message}\n".encode("utf-8"))
            else:
                continue
            break

print("Server stopped.")

逐段理解一般 TCP server

1. socket.AF_INET 指 IPv4；socket.SOCK_STREAM 表示串流 socket，也就是常見的 TCP。
2. bind((HOST, PORT)) 把 server 綁定到 IP 位址與埠號。使用 127.0.0.1 只接受本機連線。
3. listen() 開始等待連線，accept() 取出一個已建立的連線。此簡化版一次只處理一個 client。
4. recv(1024) 最多讀取 1024 位元組。一次 recv() 不保證剛好取得一整則訊息，所以程式用 buffer 暫存，直到找到換行符號。
5. sendall() 持續送出資料，直到資料都交給作業系統傳送；這比只呼叫一次 send() 更適合初學範例。

多 client 的限制：這個 server 在處理目前 client 時，不會回到 accept() 接下一位。常見改法是在每次 accept() 後開一個執行緒處理該連線，或改用非阻塞 I/O。

一般 TCP client

import socket

HOST = "127.0.0.1"
PORT = 8888

with socket.create_connection((HOST, PORT)) as client:
    print(client.recv(1024).decode("utf-8").rstrip())

    while True:
        message = input("> ")
        client.sendall((message + "\n").encode("utf-8"))
        if message == "/quit":
            break

        # 此簡化範例假設每次回覆都會在一次 recv 中取得
        response = client.recv(1024)
        if not response:
            print("Server closed the connection.")
            break
        print(response.decode("utf-8").rstrip())

client 的 create_connection() 建立 TCP 連線，input() 讀取鍵盤輸入，再將字串編碼成 UTF-8 位元組送出。這個版本採取「送一則、等一則回覆」的流程，因此簡單易懂；若要同時接收 server 主動推送的資料，就要再加入執行緒或其他 I/O 管理方式。

四、asyncio 版本：事件迴圈安排多個連線

asyncio server

import asyncio

HOST = "127.0.0.1"
PORT = 8888

async def handle_client(reader, writer):
    peer = writer.get_extra_info("peername")
    print(f"Client connected: {peer}")
    writer.write(b"Connected. Type /quit to leave.\n")
    await writer.drain()

    try:
        while True:
            data = await reader.readline()
            if not data:                 # client 正常關閉連線
                break
            message = data.decode("utf-8", errors="replace").rstrip("\r\n")
            if message == "/quit":
                break
            print(f"{peer}: {message}")
            writer.write(f"Server received: {message}\n".encode("utf-8"))
            await writer.drain()
    finally:
        writer.close()
        await writer.wait_closed()
        print(f"Client disconnected: {peer}")

async def main():
    server = await asyncio.start_server(handle_client, HOST, PORT)
    print(f"Server listening on {HOST}:{PORT}")
    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Server stopped.")

asyncio server 的重點

1. async def handle_client(...) 定義協程。每當有 client 連入，start_server() 會呼叫這個處理函式。
2. reader 與 writer 是 asyncio 提供的串流讀寫介面。reader.readline() 讀到換行才回傳一則訊息。
3. await 表示這裡可能需要等待。等待資料或等待寫入完成時，事件迴圈可以處理其他 client。
4. writer.write() 把資料放入輸出緩衝區；await writer.drain() 等待緩衝區適度排空，避免大量資料不受控制地累積。
5. asyncio.start_server() 啟動 TCP server；serve_forever() 持續接受連線。每個 client 都有自己的 handle_client 協程，因此連線可交錯處理。
6. asyncio.run(main()) 建立並執行事件迴圈，程式結束時收尾關閉。

asyncio client：邊輸入、邊接收

import asyncio

HOST = "127.0.0.1"
PORT = 8888

async def receive_messages(reader):
    while True:
        data = await reader.readline()
        if not data:
            print("\nServer closed the connection.")
            return
        print(data.decode("utf-8", errors="replace").rstrip())

async def send_messages(writer):
    while True:
        # input() 是同步阻塞函式，移到工作執行緒以免卡住事件迴圈
        message = await asyncio.to_thread(input, "> ")
        writer.write(f"{message}\n".encode("utf-8"))
        await writer.drain()
        if message == "/quit":
            return

async def main():
    reader, writer = await asyncio.open_connection(HOST, PORT)
    receiver = asyncio.create_task(receive_messages(reader))
    try:
        await send_messages(writer)
    finally:
        writer.close()
        await writer.wait_closed()
        receiver.cancel()
        await asyncio.gather(receiver, return_exceptions=True)

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

client 把接收與傳送拆成兩個協程。asyncio.create_task() 讓接收工作先在背景執行；主流程等待使用者輸入並傳送訊息。由於內建的 input() 是阻塞函式，範例用 asyncio.to_thread() 把它放到工作執行緒，避免鍵盤輸入期間事件迴圈完全停住。

為什麼要用換行分隔？TCP 不保留應用程式的「訊息包」邊界。reader.readline() 需要一個可辨識的結束標記；本範例採用換行。實際系統也常用固定長度標頭、長度前綴或 JSON Lines。

五、asyncio 的優點與限制

面向	優點	限制與注意事項
連線數	適合管理大量同時等待 I/O 的連線，協程通常比每連線一個執行緒更輕量。	仍受作業系統、網路、記憶體、檔案描述符與程式設計品質限制。
資源使用	可減少大量執行緒的建立與切換成本。	不是零成本；每個連線仍需要狀態與緩衝記憶體。
流程控制	透過 await 清楚標出可能等待 I/O 的位置。	忘記 await、阻塞事件迴圈或處理取消不當，會造成難查的錯誤。
CPU 運算	適合網路、資料庫、檔案等 I/O 密集工作。	不會自動加速長時間 CPU 計算；需考慮 process pool 或其他架構。
函式庫相容性	許多現代網路函式庫有 async 介面。	同步函式庫若直接在協程中執行，可能阻塞整個事件迴圈。

六、如何執行兩組程式

1. 確認電腦已安裝 Python 3.9 以上版本。這些範例使用 Python 內建模組，不需額外安裝套件。
2. 將 server 與 client 存成不同檔案，例如 tcp_server.py、tcp_client.py。
3. 開啟第一個終端機，在程式所在資料夾執行 server：python tcp_server.py。asyncio 版本則執行對應的 asyncio_server.py。
4. 再開第二個終端機執行 client：python tcp_client.py，或執行 python asyncio_client.py。
5. 輸入訊息後按 Enter。輸入 /quit 結束 client；在 server 終端機按 Ctrl+C 停止 server。

要測試多個連線，可開啟多個 client 終端機。前述簡化的阻塞式 server 一次處理一個 client；asyncio server 可以在事件迴圈中交錯服務多個連線。

區網連線：若要讓同一區域網路的其他電腦連入，server 通常要綁定 0.0.0.0，client 的 HOST 則改為 server 電腦的區網 IP；也要確認作業系統防火牆允許該埠號。不要把未加密、未驗證的教學 server 直接暴露到公網。

七、什麼時候選哪一種？

• 教學初期想認識 IP、埠號、bind、listen、accept、send、recv：先寫一般 socket 版本。
• 服務需要同時等待許多 client、網路 API 或資料庫回覆：可考慮 asyncio，並搭配支援 async 的函式庫。
• 工作主要是影像處理、模型推論或大型計算：先分析 CPU 負載；asyncio 本身不會讓計算平行化。
• 系統很小、只有少數連線：選團隊較容易維護的寫法，通常比追求某種模式更重要。

結語

一般阻塞式 TCP 程式會在呼叫讀取或接受連線時等待，適合入門與簡單服務。asyncio 把等待中的工作交給事件迴圈管理，讓程式可在等待某個 client 時處理其他 client，適合 I/O 密集且連線較多的服務。理解差異的關鍵，是知道 await 讓出控制權的時機、TCP 位元組串流的特性，以及阻塞函式可能對整個事件迴圈造成的影響。

教學提醒：範例為了說明核心流程而刻意保持精簡。正式服務還需設計逾時、錯誤回覆、身分驗證、加密、訊息大小限制、記錄與資源管理。