【LLM】MCP(Python)：實現 stdio 通信的Client與Server

本文將詳細介紹如何使用 Model Context Protocol (MCP) 在 Python 中實現基于 STDIO 通信的 Client 與 Server。MCP 是一個開放協議，它使 LLM 應用與外部數據源和工具之間的無縫集成成為可能。無論你是構建 AI 驅動的 IDE、改善 chat 交互，還是構建自定義的 AI 工作流，MCP 提供了一種標準化的方式，將 LLM 與它們所需的上下文連接起來。部分靈感來源：Se7en。

提示
在開始前，請確保你已經安裝了必要的依賴包：
pip install openai mcp

本文中，我們將介紹如何配置環境、編寫 MCP Server 以及實現 MCP Client。

環境配置

在使用 MCP 之前，需要先配置相關環境變量，以便 Client 與 Server 都能正確加載所需的參數。你可以在項目根目錄下創建一個 .env 文件，并寫入以下內容：

此外，創建一個 .env 文件來存儲您的配置：

MODEL_NAME=deepseek-chat
BASE_URL=https://api.deepseek.com/v1
API_KEY=your_api_key_here

上述配置中，MODEL_NAME 表示使用的 OpenAI 模型名稱（例如 “deepseek-chat”），BASE_URL 指向 OpenAI API 的基礎地址，而 API_KEY 則為訪問 API 所需的密鑰。

書寫 Server 的規范

構建 MCP Server（特別是基于 stdio 通信）時，推薦遵循統一規范，提升可讀性、可維護性與復用性。

服務命名統一
使用 MCP_SERVER_NAME 作為唯一名稱，貫穿日志、初始化等環節。
日志配置清晰
統一使用 logging 模塊，推薦 INFO 級別，便于調試和追蹤。
工具注冊規范
通過 @mcp.tool() 裝飾器注冊工具函數，要求：
- 命名清晰
- 參數有類型注解
- 注釋說明參數與返回值（推薦中文）
- 加入邊界檢查或異常處理
使用標準 stdio 啟動方式
通過 async with stdio_server() 獲取輸入輸出流，統一調用 _mcp_server.run(...) 啟動服務。
初始化選項規范
使用 InitializationOptions 設置服務名、版本及能力聲明（通常由 FastMCP 提供）。

通用模板

import asyncio
import logging
from mcp.server.fastmcp import FastMCP
from mcp.server import InitializationOptions, NotificationOptions
from mcp.server.stdio import stdio_server  # STDIO 通信方式# 定義唯一服務名稱
MCP_SERVER_NAME = "your-stdio-server-name"# 配置日志輸出
logging.basicConfig(level=logging.INFO,format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(MCP_SERVER_NAME)# 創建 FastMCP 實例
mcp = FastMCP(MCP_SERVER_NAME)# 定義工具
@mcp.tool()
def your_tool_name(param1: type, param2: type) -> return_type:"""工具描述。參數：- param1 (type): 描述- param2 (type): 描述返回：- return_type: 描述"""# 工具實現pass# 啟動 MCP Server 主函數
async def main():# 創建 stdio 通信通道async with stdio_server() as (read_stream, write_stream):# 構建初始化選項init_options = InitializationOptions(server_name=MCP_SERVER_NAME,server_version="1.0.0",capabilities=mcp._mcp_server.get_capabilities(notification_options=NotificationOptions(),experimental_capabilities={}))logger.info("MCP Server 以 STDIO 模式啟動中...")# 啟動 Serverawait mcp._mcp_server.run(read_stream, write_stream, init_options)# 主程序入口
if __name__ == "__main__":asyncio.run(main())

編寫 MCP Server

MCP Server 的實現主要基于標準輸入輸出（STDIO）進行通信。服務端通過注冊工具，向外界提供如加法、減法、乘法以及除法等計算功能。下面簡述服務端的主要實現步驟：

初始化 FastMCP 實例
服務端首先創建一個 FastMCP 實例，并為其命名（例如 “math-stdio-server”）。
工具注冊
使用裝飾器的方式注冊加法、減法、乘法、除法等工具，每個工具均包含詳細的參數說明和返回值說明。
日志配置
通過 Python 標準日志模塊對服務端進行日志配置，以便記錄服務運行狀態和錯誤信息。
建立 STDIO 通信
使用 stdio_server() 函數建立基于 STDIO 的通信，并構造初始化選項，包含服務器名稱、版本以及能力說明。隨后，調用 MCP 內部的服務啟動函數開始監聽和處理來自 Client 的請求。

import asyncio
import logging
from mcp.server.fastmcp import FastMCP
from mcp.server import InitializationOptions, NotificationOptions
from mcp.server.stdio import stdio_server  # 直接導入 stdio_server 函數# 定義服務器名稱
MCP_SERVER_NAME = "math-stdio-server"# 配置日志
logging.basicConfig(level=logging.INFO,format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(MCP_SERVER_NAME)# 初始化 FastMCP 實例
mcp = FastMCP(MCP_SERVER_NAME)# 注冊加法工具
@mcp.tool()
def add(a: float, b: float) -> float:"""加法工具參數：- a (float): 第一個數字（必填）- b (float): 第二個數字（必填）返回：- float: a 與 b 的和"""return a + b# 注冊減法工具
@mcp.tool()
def subtract(a: float, b: float) -> float:"""減法工具參數：- a (float): 被減數（必填）- b (float): 減數（必填）返回：- float: a 與 b 的差"""return a - b# 注冊乘法工具
@mcp.tool()
def multiply(a: float, b: float) -> float:"""乘法工具參數：- a (float): 第一個數字（必填）- b (float): 第二個數字（必填）返回：- float: a 與 b 的積"""return a * b# 注冊除法工具
@mcp.tool()
def divide(a: float, b: float) -> float:"""除法工具參數：- a (float): 分子（必填）- b (float): 分母（必填，且不能為零）返回：- float: a 與 b 的商"""if b == 0:raise ValueError("除數不能為零")return a / basync def main():# 使用 stdio_server 建立 STDIO 通信async with stdio_server() as (read_stream, write_stream):# 構造初始化選項init_options = InitializationOptions(server_name=MCP_SERVER_NAME,server_version="1.0.0",capabilities=mcp._mcp_server.get_capabilities(notification_options=NotificationOptions(),experimental_capabilities={}))logger.info("通過 STDIO 模式啟動 MCP Server ...")# 使用內部的 _mcp_server 運行服務await mcp._mcp_server.run(read_stream, write_stream, init_options)if __name__ == "__main__":asyncio.run(main())

編寫 MCP Client

MCP Client 主要實現與多個基于 STDIO 的服務器建立連接，并通過 OpenAI API 對用戶的自然語言查詢進行處理，調用相應工具獲得最終結果。客戶端的主要邏輯可以分為以下部分：

初始化客戶端
在 MCPClient 類的構造函數中，傳入所需的模型名稱、OpenAI API 基礎地址、API 密鑰以及包含服務端腳本路徑的列表。客戶端將使用這些參數初始化 OpenAI 異步客戶端，同時準備一個 AsyncExitStack 來管理所有異步上下文。
建立多個 STDIO 連接
通過遍歷服務器腳本列表，為每個腳本生成唯一標識符（如 server0、server1 等），然后依次調用 stdio_client 函數建立連接，并通過 ClientSession 完成初始化。在連接成功后，從每個服務器獲取可用工具列表，并將工具名稱加上前綴（例如 server0_add）保存到映射表中，避免工具名稱沖突。
處理用戶查詢
在 process_query 方法中，客戶端首先根據用戶的輸入構造消息，然后匯總所有連接服務器提供的工具，傳遞給 OpenAI API 進行處理。當 API 返回調用工具的請求時，客戶端根據工具名稱找到對應服務器會話，并執行相應的工具調用，收集返回結果后再交由 API 生成后續回復，直至所有工具調用處理完成。
交互式對話循環
客戶端提供一個簡單的命令行交互循環，用戶輸入查詢后，調用 process_query 方法獲取最終回復，并打印在終端上。如果用戶輸入 quit 或使用 Ctrl+C 中斷，則客戶端將平滑退出并釋放所有資源。
資源清理
最后，在退出前，通過 AsyncExitStack 統一關閉所有連接，確保資源不會泄露。

import asyncio
import json
import os
import sys
from typing import List
from contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import AsyncOpenAIclass MCPClient:def __init__(self, model_name: str, base_url: str, api_key: str, server_scripts: List[str]):"""初始化 MCP 客戶端，支持多個 stdio 服務器。:param model_name: OpenAI 模型名稱，例如 "deepseek-chat"。:param base_url: OpenAI API 基礎地址，例如 "https://api.deepseek.com/v1"。:param api_key: OpenAI API 密鑰。:param server_scripts: stdio 服務腳本路徑列表。"""self.model_name = model_nameself.base_url = base_urlself.api_key = api_keyself.server_scripts = server_scriptsself.sessions = {}         # server_id -> (session, session_ctx, stdio_ctx)self.tool_mapping = {}     # 帶前綴的工具名 -> (session, 原始工具名)self.client = AsyncOpenAI(base_url=base_url, api_key=api_key)self.exit_stack = AsyncExitStack()async def initialize_sessions(self):"""初始化所有 stdio 服務器連接，并收集工具映射。"""for i, script in enumerate(self.server_scripts):if not (os.path.exists(script) and script.endswith(".py")):print(f"腳本 {script} 不存在或不是 .py 文件，跳過。")continueserver_id = f"server{i}"params = StdioServerParameters(command="python", args=[script], env=None)try:stdio_ctx = stdio_client(params)stdio = await self.exit_stack.enter_async_context(stdio_ctx)session_ctx = ClientSession(*stdio)session = await self.exit_stack.enter_async_context(session_ctx)await session.initialize()self.sessions[server_id] = (session, session_ctx, stdio_ctx)response = await session.list_tools()for tool in response.tools:self.tool_mapping[f"{server_id}_{tool.name}"] = (session, tool.name)print(f"已連接到 {script}，工具：{[tool.name for tool in response.tools]}")except Exception as e:print(f"連接 {script} 失敗：{e}")async def cleanup(self):"""釋放所有資源。"""try:await self.exit_stack.aclose()print("所有連接資源已釋放")except asyncio.CancelledError:passexcept Exception as e:print(f"清理資源時異常：{e}")async def _gather_available_tools(self):"""匯總所有服務器的工具列表。"""tools = []for server_id, (session, _, _) in self.sessions.items():response = await session.list_tools()for tool in response.tools:tools.append({"type": "function","function": {"name": f"{server_id}_{tool.name}","description": tool.description,"parameters": tool.inputSchema,}})return toolsasync def process_query(self, query: str) -> str:"""處理查詢，調用 OpenAI API 和相應工具后返回結果。"""messages = [{"role": "user", "content": query}]available_tools = await self._gather_available_tools()try:response = await self.client.chat.completions.create(model=self.model_name, messages=messages, tools=available_tools)except Exception as e:return f"調用 OpenAI API 失敗：{e}"final_text = [response.choices[0].message.content or ""]message = response.choices[0].message# 當有工具調用時循環處理while message.tool_calls:for call in message.tool_calls:tool_name = call.function.nameif tool_name not in self.tool_mapping:final_text.append(f"未找到工具：{tool_name}")continuesession, original_tool = self.tool_mapping[tool_name]tool_args = json.loads(call.function.arguments)try:result = await session.call_tool(original_tool, tool_args)final_text.append(f"[調用 {tool_name} 參數: {tool_args}]")final_text.append(f"工具結果: {result.content}")except Exception as e:final_text.append(f"調用 {tool_name} 出錯：{e}")continuemessages += [{"role": "assistant", "tool_calls": [{"id": call.id,"type": "function","function": {"name": tool_name, "arguments": json.dumps(tool_args)}}]},{"role": "tool", "tool_call_id": call.id, "content": str(result.content)}]try:response = await self.client.chat.completions.create(model=self.model_name, messages=messages, tools=available_tools)except Exception as e:final_text.append(f"調用 OpenAI API 失敗：{e}")breakmessage = response.choices[0].messageif message.content:final_text.append(message.content)return "\n".join(final_text)async def chat_loop(self):"""交互式對話循環，捕獲中斷平滑退出。"""print("MCP 客戶端已啟動，輸入問題，輸入 'quit' 退出。")while True:try:query = input("問題: ").strip()if query.lower() == "quit":breakresult = await self.process_query(query)print("\n" + result)except KeyboardInterrupt:print("\n檢測到中斷信號，退出。")breakexcept Exception as e:print(f"發生錯誤：{e}")async def main():model_name = os.getenv("MODEL_NAME", "deepseek-chat")base_url = os.getenv("BASE_URL", "https://api.deepseek.com/v1")api_key = os.getenv("API_KEY")if not api_key:print("未設置 API_KEY 環境變量")sys.exit(1)# 示例：使用兩個 stdio 腳本server_scripts = ["server.py"]client = MCPClient(model_name, base_url, api_key, server_scripts)try:await client.initialize_sessions()await client.chat_loop()except KeyboardInterrupt:print("\n收到中斷信號")finally:await client.cleanup()if __name__ == "__main__":try:asyncio.run(main())except KeyboardInterrupt:print("程序已終止。")