用 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。
官方 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 程式逐段解說
from mcp.server import MCPServer匯入 SDK 的高階 server 類別。它負責處理 MCP 的初始化、工具清單、呼叫與結果傳輸。MCPServer(...)建立 server。第一個參數是供 client 辨識的名稱;instructions提供 server 的用途說明。@mcp.tool()將下一個 Python 函式登記成 MCP 工具。函式名稱成為工具名稱,型別提示用來建立參數規格,docstring 則是工具用途說明。echo_message(message: str)接收文字、在 server 終端機印出紀錄,再回傳回應。SDK 會把回傳值包裝成 MCP 工具結果。get_server_info()示範回傳結構化資料。client 可以從工具結果取得名稱、位址、埠號和傳輸方式。mcp.run(transport="streamable-http", ...)啟動 HTTP 傳輸。預設 MCP endpoint 是/mcp,所以本機網址為http://127.0.0.1:8000/mcp。
四、撰寫 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 程式逐段解說
Client(SERVER_URL)依 URL 選用 Streamable HTTP 傳輸。真正建立連線是在進入async with時;離開區塊時 SDK 會關閉連線。await client.list_tools()向 server 查詢已註冊工具。這是 MCP 相較自訂 echo 協定的關鍵:client 能探索工具名稱與輸入規格。await client.call_tool("get_server_info", {})呼叫不需參數的工具;第二個參數是工具輸入字典。await client.call_tool("echo_message", {"message": message})以工具名稱與參數送出請求。SDK 負責將呼叫轉成 MCP 訊息並交給 HTTP 傳輸。result.content是內容區塊清單。範例只列印文字區塊,因此先用isinstance(block, TextContent)確認型別。result.is_error可判斷工具是否回報錯誤;result.structured_content則提供適合程式讀取的結構化結果。input()是同步阻塞函式。範例以anyio.to_thread.run_sync()在工作執行緒讀取鍵盤,避免互動輸入直接卡住非同步流程。
五、啟動與操作
- 開啟第一個終端機,啟用虛擬環境後執行
python mcp_server.py。server 會在127.0.0.1:8000提供 MCP endpoint。 - 另開第二個終端機,使用同一個環境執行
python mcp_client.py。 - client 會列出工具、取得 server 資訊,接著等待輸入訊息。
- 輸入文字後按 Enter,client 呼叫
echo_message,server 終端機會顯示收到的文字,client 也會看到回傳內容。 - 輸入
/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 負責協定細節。這讓範例從「傳送文字」進一步成為「用標準介面呼叫服務功能」。
沒有留言:
張貼留言