2026年10月25日 星期日

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

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 socketasyncio TCPMCP
主要工作自行定義資料格式並收發位元組。用協程管理非同步 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 負責協定細節。這讓範例從「傳送文字」進一步成為「用標準介面呼叫服務功能」。

沒有留言:

張貼留言