手搓Manus？MCP原理解析與MCPClient實踐

阿里妹導讀

本文講述了MCP 原理解析和作者的MCP Client實踐，希望能實現一個自己的 agent，讓 AI 不僅能與人交流，還能協助工作。文末附原始碼！

MCP（Model Context Protocol）是由Anthropic於2024年底提出並開源的一種協議，旨在為AI系統（如AI程式設計助手、Agent等）提供安全、標準化的資料訪問方式。它採用客戶端-伺服器架構，使AI工具（如Claude Desktop、IDE外掛等）能夠透過MCP客戶端與MCP服務端互動，訪問本地或遠端資料來源。

官方文件：MCP Quickstart

https://modelcontextprotocol.io/quickstart/server

基礎概念

基礎概念講解總結自官方文件

MCP 是客戶端-服務端架構，一個 Host 可以連線多個 MCP Server。

MCP Hosts（宿主程式）：如Claude Desktop、IDE等，透過MCP訪問資料。
MCP Clients（客戶端）：與伺服器建立1:1連線，處理通訊。
MCP Servers（服務端）：輕量級程式，提供標準化的資料或工具訪問能力。
Local Data Sources（本地資料來源）：如檔案、資料庫等，由MCP服務端安全訪問。
Remote Services（遠端服務）：如API、雲服務等，MCP服務端可代理訪問。

協議層與傳輸層

協議層（Protocol Layer）

負責訊息封裝（framing）、請求/響應關聯、高階通訊模式管理。

傳輸層（Transport Layer）

支援兩種通訊方式：

1.Stdio傳輸（標準輸入/輸出）

適用於本地程序間通訊。

2.HTTP + SSE傳輸

服務端→客戶端：Server-Sent Events（SSE）
客戶端→服務端：HTTP POST
適用於遠端網路通訊。

所有傳輸均採用JSON-RPC 2.0進行訊息交換。

訊息型別

MCP 擁有多種型別的訊息來處理不同的場景

請求（Request）（期望獲得響應）

interfaceRequest { method: string;params?: { ... };}

成功響應（Result）

interfaceResult { [key: string]: unknown;}

錯誤響應（Error）

interfaceError {code: number;message: string; data?: unknown;}

通知（Notification）（單向，無需響應）

interfaceNotification { method: string;params?: { ... };}

生命週期

類似於三次握手，MCP客戶端與MCP服務端初始化建立連線會進行以下步驟：

初始化（Initialization）

1.客戶端傳送initialize請求（含協議版本、能力集）。

2.服務端返回版本及能力資訊。

3.客戶端傳送initialized通知確認。

4.進入正常通訊階段。

訊息交換（Message Exchange）

當初始化完畢，就可以進行通訊了，目前支援：

請求-響應模式（Request-Response）：雙向通訊。

通知模式（Notification）：單向訊息。

終止（Termination）

有以下幾種方式會關閉連線

主動關閉（close()）。
傳輸層斷開。
錯誤觸發終止。

實踐

基礎概念介紹完畢，接下來進行實踐，我希望能實現一個自己的 agent，讓 AI 不僅能和我交流，還能幫我幹活。換一句話就是

myAgent is a general AI agent that turns your thoughts into actions. It excels at various tasks in work and life, getting everything done while you rest.

[手動doge][手動doge]

先畫一個圖：

如圖，要實現一個這樣的效果，實現一個 myAgent，啟動時，MCP Client建立與 MCP 服務端的連線，此時 MCP Server 上的能力或者可呼叫的工具會註冊進來，讓 Client 感知到這個MCP服務能夠幹啥。

當用戶與 Agent 進行互動時，Agent 會讓 MCP Client 將使用者的輸入傳送給 AI，讓 AI 解析使用者意圖，一併傳送的還有註冊在 Client 上的能力集合。

我寫了一個搜尋助手的 MCP Server ，能力集的資料是這樣的，可以看到目前只有一個 function，get_offers，可以看到工具裡有它的名字，能力描述，需要的欄位（包含欄位型別，欄位描述）。

available_tools is: [{'type': 'function', 'function': {'name': 'get_offers', 'description': 'Get product offers from API', 'parameters': {'type': 'object', 'properties': {'keywords': {'type': 'string', 'description': 'Keywords to search for products', 'default': ''}, 'pageSize': {'type': 'number', 'description': 'Number of items per page', 'minimum': 1, 'maximum': 100, 'default': 10}}}}}]

AI 會理解使用者意圖，決定是否用自然語言回答使用者，或者選擇合適的工具，告訴 client，幫我呼叫它。

當輸入 你好 時，傳給 AI 的 message 是這樣的，這裡系統預設了 AI 的一個身份，用於更好的完成特定領域的任務。

[{"role":"user","content":"你好"},{"role":"system","content":"You're a helpful digital assistant who can answer questions and support users in completing tasks."}]

此時 client 接收到 AI 的訊息後，會解析資料，當沒有工具要呼叫時

AI返回的是這樣的：

ChatCompletionMessage(content='你好！有什麼可以幫助你的嗎？', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=None)

可以看到，AI 返回是帶有角色資訊的，然後本次並沒有識別到需要呼叫工具的地方，因此直接返回給使用者就好，當然，在工程應用時，可以進行額外的邏輯處理。

讓 AI 長出手，AI呼叫 MCP Server流程揭秘

當輸入 幫我找一些手錶 時，輸入是：

[{'role': 'user', 'content': '幫我找一些手錶'}, {'role': 'system', 'content': 'You are a helpful assistant that can answer questions and help with tasks.'}]

第一次AI 互動

AI返回的是

AI response isChatCompletionMessage(content='', refusal=None, role='assistant', annotations=None, audio=None, function_call=None,

tool_calls=[ChatCompletionMessageToolCall(id='0195c8c06aaf3ea050e6d8eed17380ec', function=Function(arguments='{"keywords": "手錶", "pageSize": 10}', name='get_offers'), type='function')])

可以看到，AI 識別到了使用者的意圖，是要尋找一些手錶，並自動的選擇了一個工具進行呼叫，根據工具使用說明，決定了選擇的工具應該輸入什麼入參。(這裡和模型很相關，是一個重要的節點，識別使用者意圖並決定要呼叫工具，有時識別的並不準確，或者返回的結構不是標準可解析的，這時就觸發不了工具的呼叫，還會引入一些辣雞資訊，可能的解決方案是換效果更好的模型，或者用提示詞來約束模型返回，或者系統自己增加魯棒性，提升成功率）

下面舉一個效果不好的例子，大家如果知道有其他解決方法歡迎留言。

AI response is ChatCompletionMessage(content='leton\n{{"name": "get_offers", "arguments": {"keywords": "手錶", "pageSize": 10}}}\n</tool_call>', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=None)

Client 呼叫 MCP Server

client 接收到AI 的訊息後，發現要呼叫工具，並且也有了工具所需的引數，就會與透過協議與 MCP Server 進行通訊，告訴 MCP Server 該使用 get_offers 能力了，並且期待 MCP Server 將結果返回回來：

result = await self.session.call_tool(tool_name, tool_args)

獲取 MCP Server 資料

MCP Server 不負眾望，將結果返回了，可以看看返回的格式是什麼樣的：

meta=None content=[TextContent(type='text', text='some...product...info', annotations=None)] isError=False

MCP Client 拿到資料後，再將資料傳送給 AI，輸入是這樣的：

{"messages": [ {"role": "user","content": "幫我找一些手錶" }, {"role": "system","content": "You are a helpful assistant that can answer questions and help with tasks." }, {"role": "assistant","content": "","tool_calls": [ {"id": "0195c8c06aaf3ea050e6d8eed17380ec","type": "function","function": {"name": "get_offers","arguments": "{\"keywords\": \"手錶\", \"pageSize\": 10}" } } ] }, {"role": "tool","tool_call_id": "0195c8c06aaf3ea050e6d8eed17380ec","content": {"type": "text","text": "some...product...info" } } ]}

第二次 AI 互動

最後，AI 將 MCPServer 的結果，進行總結潤色，結構化返回：

🤖AI: [📞呼叫工具 get_offers 🔑引數是 {'keywords': '手錶', 'pageSize': 10}]根據您的搜尋，這裡有幾款手錶供您參考：1. 款式 ID: ididid2 價格: $0.24 供應商: GuangzhouHuapanCosmeticsCo., Ltd. 評分: ** 收評次數: ** 供應商年限: **年 推薦指數: ★★2. 款式 ID: ididid 價格: $3.99 供應商: ShenzhenTopQualityElectronicsTechnologyCo., Ltd. 評分: ** 收評次數: ** 供應商年限: **年 推薦指數: ★★

這兩款手錶的評價和銷售情況都還不錯，您可以根據自己的需求選擇合適的款式。如果還有其他問題或需要更多資訊，請隨時告訴我。

這裡給了它 10 個品，但是隻總結了兩個品，可能適合我之前的輸入 幫我找一些手錶 有關，看來AI 也會偷懶😅。

實際效果

實踐後的總結

上面的互動過程，其實可以化簡，如果在工程應用上，呼叫的 MCP Server 是一個預期內的結構化的結果或者觸發某個任務時，可以不必進行二次 AI 呼叫。如上面的例子中，MCP Server 是一個搜尋助手，內部發起呼叫的是搜尋的介面，並進行結構化返回。此時在 AI 識別到使用者意圖並告訴 Client 該呼叫什麼工具時，與 AI 的互動就可以結束了，由系統接管決定應該返回給使用者什麼，不必再將結果給到 AI 進行潤色總結。

給到 AI 進行潤色總結的好處是可以根據使用者的輸入，再結合工具獲取的資料，更智慧友好的返回給使用者資訊，這一點可以在工程應用時，進行衡量取捨。

將MCP Server 的結果交給 AI，在需要進行多輪互動場景是有必要的，根據 MCP Server的結果，進行分析及決策，動態調整要使用的工具，可以將一個複雜的任務交給 AI , 它會自己拆解成小任務，然後自動完成。

對於該場景，也進行了一些實踐。

例如，讓Al agent拆解抽象任務，並自己主動與系統進行多輪互動，完成任務場景。

後續文章我們會詳細介紹具體實現方法，講講如何讓Al在我的電腦上玩起貪吃蛇？

附錄

MCP Client程式碼 (Python實現)

import asyncioimport jsonimport osimport tracebackfrom typing importOptionalfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom openai import OpenAIfrom dotenv import load_dotenvload_dotenv() # load environment variables from .envclassMCPClient:def__init__(self):# Initialize session and client objects self.session: Optional[ClientSession] = None self.exit_stack = AsyncExitStack() self.client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL") ) self.model = os.getenv("OPENAI_MODEL") self.messages = [ {"role": "system",

"content": "You are a versatile assistant capable of answering questions, completing tasks, and intelligently invoking specialized tools to deliver optimal results."

} ] self.available_tools = [] @staticmethoddefconvert_custom_object(obj):""" 將自定義物件轉換為字典 """ifhasattr(obj, "__dict__"): # 如果物件有 __dict__ 屬性，直接使用return obj.__dict__elifisinstance(obj, (list, tuple)): # 如果是列表或元組，遞迴處理return [MCPClient.convert_custom_object(item) for item in obj]elifisinstance(obj, dict): # 如果是字典，遞迴處理值return {key: MCPClient.convert_custom_object(value) for key, value in obj.items()}else: # 其他型別（如字串、數字等）直接返回return objasyncdefconnect_to_server(self, server_script_path: str):"""Connect to an MCP server Args: server_script_path: Path to the server script (.py or .js) """ is_python = server_script_path.endswith('.py') is_js = server_script_path.endswith('.js')ifnot (is_python or is_js):raise ValueError("Server script must be a .py or .js file") command = "python"if is_python else"node" server_params = StdioServerParameters( command=command, args=[server_script_path], env=None ) stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params)) self.stdio, self.write = stdio_transport self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize()# List available tools response = await self.session.list_tools() tools = response.toolsprint("\nConnected to server with tools:", [tool.name for tool in tools])asyncdefprocess_query(self, query: str) -> str:"""Process a query with multi-turn tool calling support"""# Add user query to message history self.messages.append({"role": "user","content": query })# Get available tools if not already setifnot self.available_tools: response = await self.session.list_tools() self.available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"parameters": tool.inputSchema } } for tool in response.tools] current_response = self.client.chat.completions.create( model=self.model, messages=self.messages, tools=self.available_tools, stream=False )# Print initial response if existsif current_response.choices[0].message.content:print("\n🤖 AI:", current_response.choices[0].message.content)# 直到下一次互動 AI 沒有選擇呼叫工具時退出迴圈while current_response.choices[0].message.tool_calls:# AI 一次互動中可能會呼叫多個工具for tool_call in current_response.choices[0].message.tool_calls: tool_name = tool_call.function.nametry: tool_args = json.loads(tool_call.function.arguments)except json.JSONDecodeError: tool_args = {}print(f"\n🔧 呼叫工具 {tool_name}")print(f"📝 引數: {tool_args}")# Execute tool call result = await self.session.call_tool(tool_name, tool_args)print(f"\n工具結果: {result}")# Add AI message and tool result to history self.messages.append(current_response.choices[0].message) self.messages.append({"role": "tool","tool_call_id": tool_call.id,"content": json.dumps(result.content) })# Get next response current_response = self.client.chat.completions.create( model=self.model, messages=self.messages, tools=self.available_tools, stream=False )# Add final response to history self.messages.append(current_response.choices[0].message)return current_response.choices[0].message.content or""asyncdefchat_loop(self):"""Run an interactive chat loop"""print("\nMCP Client Started!")print("Type your queries or 'quit' to exit.")whileTrue:try: query = input("\nCommend: ").strip()if query.lower() == 'quit':break response = await self.process_query(query)print("\n🤖AI: " + response)except Exception as e:print(f"\nError occurs: {e}") traceback.print_exc()asyncdefcleanup(self):"""Clean up resources"""await self.exit_stack.aclose()asyncdefmain():iflen(sys.argv) < 2:print("Usage: python client.py <path_to_server_script>") sys.exit(1) client = MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":import sys asyncio.run(main())

MCP Client程式碼 (TypeScript實現)

/** * MCP客戶端實現 * * 提供與MCP伺服器的連線、工具呼叫和聊天互動功能 * * 主要功能： * 1. 連線Python或JavaScript實現的MCP伺服器 * 2. 獲取伺服器提供的工具列表 * 3. 透過OpenAI API處理使用者查詢 * 4. 自動處理工具呼叫鏈 * 5. 提供互動式命令列介面 * * 使用說明： * 1. 確保設定OPENAI_API_KEY環境變數 * 2. 透過命令列引數指定MCP伺服器指令碼路徑 * 3. 啟動後輸入查詢或'quit'退出 * * 依賴： * - @modelcontextprotocol/sdk: MCP協議SDK * - openai: OpenAI API客戶端 * - dotenv: 環境變數載入 */import { Client } from"@modelcontextprotocol/sdk/client/index.js";import { StdioClientTransport } from"@modelcontextprotocol/sdk/client/stdio.js";importOpenAIfrom"openai";importtype { ChatCompletionMessageParam } from"openai/resources/chat/completions";importtype { Tool } from"@modelcontextprotocol/sdk/types.js";import * as dotenv from"dotenv";import * as readline from'readline';// 載入環境變數配置dotenv.config();/** * MCP客戶端類，封裝與MCP伺服器的互動邏輯 */classMCPClient {privateopenai: OpenAI; // OpenAI API客戶端例項privateclient: Client; // MCP協議客戶端例項privatemessages: ChatCompletionMessageParam[] = [ {role: "system",

content: "You are a versatile assistant capable of answering questions, completing tasks, and intelligently invoking specialized tools to deliver optimal results."

}, ]; // 聊天訊息歷史記錄，用於維護對話上下文privateavailableTools: any[] = []; // 伺服器提供的可用工具列表，格式化為OpenAI工具格式/** * 建構函式，初始化OpenAI和MCP客戶端 * * @throws {Error} 如果OPENAI_API_KEY環境變數未設定 * * 初始化過程： * 1. 檢查必要的環境變數 * 2. 建立OpenAI客戶端例項 * 3. 建立MCP客戶端例項 * 4. 初始化訊息歷史記錄 */constructor() {if (!process.env.OPENAI_API_KEY) {thrownewError("OPENAI_API_KEY環境變數未設定"); }this.openai = newOpenAI({apiKey: process.env.OPENAI_API_KEY,baseURL: process.env.OPENAI_BASE_URL, });this.client = newClient( {name: "my-mcp-client",version: "1.0.0", }, ); }/** * 連線到MCP伺服器 * * @param {string} serverScriptPath - 伺服器指令碼路徑(.py或.js) * @returns {Promise<void>} 連線成功時解析 * @throws {Error} 如果伺服器指令碼不是.py或.js檔案，或連線失敗 * * 連線過程： * 1. 檢查指令碼副檔名 * 2. 根據副檔名決定使用python或node執行 * 3. 透過stdio建立連線 * 4. 獲取伺服器工具列表並轉換為OpenAI工具格式 * * 注意事項： * - 伺服器指令碼必須具有可執行許可權 * - 連線成功後會自動獲取工具列表 */asyncconnectToServer(serverScriptPath: string) {const isPython = serverScriptPath.endsWith('.py');const isJs = serverScriptPath.endsWith('.js');if (!isPython && !isJs) {thrownewError("Server script must be a .py or .js file"); }const command = isPython ? "python" : "node";const transport = newStdioClientTransport({ command,args: [serverScriptPath], });awaitthis.client.connect(transport);// 獲取並轉換可用工具列表const tools = (awaitthis.client.listTools()).toolsasunknownasTool[];this.availableTools = tools.map(tool => ({type: "function"asconst,function: {name: tool.nameasstring,description: tool.descriptionasstring,parameters: {type: "object",properties: tool.inputSchema.propertiesasRecord<string, unknown>,required: tool.inputSchema.requiredasstring[], }, } }));console.log("\n已連線到伺服器，可用工具:", tools.map(tool => tool.name)); }/** * 處理工具呼叫鏈 * * @param {OpenAI.Chat.Completions.ChatCompletion} response - 初始OpenAI響應，包含工具呼叫 * @param {ChatCompletionMessageParam[]} messages - 當前訊息歷史記錄 * @returns {Promise<OpenAI.Chat.Completions.ChatCompletion>} 最終OpenAI響應 * * 處理流程： * 1. 檢查響應中是否包含工具呼叫 * 2. 迴圈處理所有工具呼叫 * 3. 解析每個工具呼叫的引數 * 4. 執行工具呼叫 * 5. 將工具結果新增到訊息歷史 * 6. 獲取下一個OpenAI響應 * * 錯誤處理： * - 引數解析失敗時使用空物件繼續執行 * - 工具呼叫失敗會丟擲異常 * * 注意事項： * - 此方法會修改傳入的messages陣列 * - 可能多次呼叫OpenAI API */privateasynctoolCalls(response: OpenAI.Chat.Completions.ChatCompletion, messages: ChatCompletionMessageParam[]) {let currentResponse = response;// 直到下一次互動 AI 沒有選擇呼叫工具時退出迴圈while (currentResponse.choices[0].message.tool_calls) {if (currentResponse.choices[0].message.content) {console.log("\n🤖 AI: tool_calls", JSON.stringify(currentResponse.choices[0].message)); }// AI 一次互動中可能會呼叫多個工具for (const toolCall of currentResponse.choices[0].message.tool_calls) {const toolName = toolCall.function.name;const rawArgs = toolCall.function.arguments;let toolArgs;try {console.log(`rawArgs is ===== ${rawArgs}`) toolArgs = "{}" == JSON.parse(rawArgs) ? {} : JSON.parse(rawArgs);if (typeof toolArgs === "string") { toolArgs = JSON.parse(toolArgs); } } catch (error) {console.error('⚠️ 引數解析失敗，使用空物件替代'); toolArgs = {}; }console.log(`\n🔧 呼叫工具 ${toolName}`);console.log(`📝 引數:`, toolArgs);// 呼叫工具獲取結果const result = awaitthis.client.callTool({name: toolName,arguments: toolArgs });console.log(`\n result is ${JSON.stringify(result)}`);// 新增 AI 的響應和工具呼叫結果到訊息歷史// console.log(`📝 currentResponse.choices[0].message:`, currentResponse.choices[0].message); messages.push(currentResponse.choices[0].message); messages.push({role: "tool",tool_call_id: toolCall.id,content: JSON.stringify(result.content), } asChatCompletionMessageParam); }// console.log(`📝 messages: `, messages);// 獲取下一個響應 currentResponse = awaitthis.openai.chat.completions.create({model: process.env.OPENAI_MODELasstring,messages: messages,tools: this.availableTools, }); }return currentResponse; }/** * 處理使用者查詢 * * @param {string} query - 使用者輸入的查詢字串 * @returns {Promise<string>} AI生成的響應內容 * * 處理流程： * 1. 將使用者查詢新增到訊息歷史 * 2. 呼叫OpenAI API獲取初始響應 * 3. 如果有工具呼叫，處理工具呼叫鏈 * 4. 返回最終響應內容 * * 錯誤處理： * - OpenAI API呼叫失敗會丟擲異常 * - 工具呼叫鏈中的錯誤會被捕獲並記錄 * * 注意事項： * - 此方法會更新內部訊息歷史 * - 可能觸發多個工具呼叫 */asyncprocessQuery(query: string): Promise<string> {// 新增使用者查詢到訊息歷史this.messages.push({role: "user",content: query, });// 初始OpenAI API呼叫let response = awaitthis.openai.chat.completions.create({model: process.env.OPENAI_MODELasstring,messages: this.messages,tools: this.availableTools, });// 列印初始響應if (response.choices[0].message.content) {console.log("\n🤖 AI:", response.choices[0].message); }// 處理工具呼叫鏈if (response.choices[0].message.tool_calls) { response = awaitthis.toolCalls(response, this.messages); }// 更新訊息歷史this.messages.push(response.choices[0].message);return response.choices[0].message.content || ""; }/** * 啟動互動式聊天迴圈 * * @returns {Promise<void>} 當用戶退出時解析 * * 功能： * 1. 持續接收使用者輸入 * 2. 處理使用者查詢 * 3. 顯示AI響應 * 4. 輸入'quit'退出 * * 實現細節： * - 使用readline模組實現互動式輸入輸出 * - 迴圈處理直到使用者輸入退出命令 * - 捕獲並顯示處理過程中的錯誤 * * 注意事項： * - 此方法是阻塞呼叫，會一直執行直到使用者退出 * - 確保在呼叫前已連線伺服器 */asyncchatLoop() {console.log("\nMCP Client Started!");console.log("Type your queries or 'quit' to exit.");const rl = readline.createInterface({input: process.stdin,output: process.stdout, });while (true) {const query = awaitnewPromise<string>((resolve) => { rl.question("\nQuery: ", resolve); });if (query.toLowerCase() === 'quit') {break; }try {const response = awaitthis.processQuery(query);console.log("\n" + response); } catch (e) {console.error("\nError:", e instanceofError ? e.message : String(e)); } } rl.close(); }/** * 清理資源 * * @returns {Promise<void>} 資源清理完成後解析 * * 關閉以下資源： * 1. MCP客戶端連線 * 2. 任何開啟的控制代碼 * * 最佳實踐： * - 應在程式退出前呼叫 * - 建議在finally塊中呼叫以確保執行 * * 注意事項： * - 多次呼叫是安全的 * - 清理後例項不可再用 */asynccleanup() {if (this.client) {awaitthis.client.close(); } }}/** * 主函式 * * 程式入口點，執行流程： * 1. 檢查命令列引數 * 2. 建立MCP客戶端例項 * 3. 連線到指定伺服器指令碼 * 4. 啟動互動式聊天迴圈 * 5. 退出時清理資源 * * @throws {Error} 如果缺少命令列引數或連線失敗 * * 使用示例： * ```bash * node index.js /path/to/server.js * ``` * * 退出碼： * - 0: 正常退出 * - 1: 引數錯誤或執行時錯誤 */asyncfunctionmain() {if (process.argv.length < 3) {console.log("Usage: node dist/index.js <path_to_server_script>"); process.exit(1); }const client = newMCPClient();try {await client.connectToServer(process.argv[2]);await client.chatLoop(); } finally {await client.cleanup(); }}main().catch((error) => {console.error("Error:", error); process.exit(1);});

MCP Server 程式碼 (TypeScript實現)

#!/usr/bin/env nodeimport { Server } from'@modelcontextprotocol/sdk/server/index.js';import { StdioServerTransport } from'@modelcontextprotocol/sdk/server/stdio.js';import {CallToolRequestSchema,ErrorCode,ListToolsRequestSchema,McpError,} from'@modelcontextprotocol/sdk/types.js';import axios from'axios';interfaceProduct {id: string;name: string;supplier: string; 省略...}interfaceOffer {productId: string;title: string;companyName: string; 省略...}interfaceApiResponse {ret: string[];encode: string;code: number;traceId: string;msg: string;time: number;data: {offers: Offer[];resultCount: string;totalCount: number; };}classProductOffersServer {privateserver: Server;private baseUrl = '換成你要呼叫的 url';constructor() {this.server = newServer( {name: 'search-assistant-server',version: '0.1.0', }, {capabilities: {tools: {}, }, } );this.setupToolHandlers();// Error handlingthis.server.onerror = (error) =>console.error('[MCP Error]', error); process.on('SIGINT', async () => {awaitthis.server.close(); process.exit(0); }); }privateasyncfetchOffers(keywords?: string, pageSize?: number): Promise<Product[]> {try {constparams: Record<string, any> = {};if (keywords) params.keywords = keywords;if (pageSize) params.pageSize = pageSize;const response = await axios.get<ApiResponse>(this.baseUrl, { params });return response.data.data.offers.map(offer => ({id: offer.productId,name: offer.title,supplier: offer.companyName 省略... })); } catch (error) {if (axios.isAxiosError(error)) {thrownewMcpError(ErrorCode.InternalError,`Failed to fetch offers: ${error.message}` ); }throw error; } }privatesetupToolHandlers() {this.server.setRequestHandler(ListToolsRequestSchema, async () => ({tools: [ {name: 'get_offers',description: 'Get product offers from API',inputSchema: {type: 'object',properties: {keywords: {type: 'string',description: 'Keywords to search for products',default: '' },pageSize: {type: 'number',description: 'Number of items per page',minimum: 1,maximum: 100,default: 10 } } } } ] }));this.server.setRequestHandler(CallToolRequestSchema, async (request) => {if (request.params.name !== 'get_offers') {thrownewMcpError(ErrorCode.MethodNotFound,`Unknown tool: ${request.params.name}` ); }const args = request.params.argumentsas { keywords?: string; pageSize?: number };try {const products = awaitthis.fetchOffers(args.keywords, args.pageSize);return {content: [ {type: 'text',text: JSON.stringify({products: products,totalCount: products.length }, null, 2) } ] }; } catch (error) {if (error instanceofMcpError) {throw error; }thrownewMcpError(ErrorCode.InternalError,`Failed to fetch offers: ${error}` ); } }); }asyncrun() {const transport = newStdioServerTransport();awaitthis.server.connect(transport);console.error('Product Offers MCP server running on stdio'); }}const server = newProductOffersServer();server.run().catch(console.error);

SelectDB 實現日誌高效儲存與即時分析

企業級日誌資料具有資料量巨大、寫入和查詢速度快、結構多樣的特點，本方案基於阿里云云資料庫 SelectDB 版構建高效能、低成本、分析能力強大的日誌儲存與分析解決方案，覆蓋運維監控、安全審計、業務分析等場景，並透過智慧索引與分級儲存實現資料亞秒級檢索。

點選閱讀原文檢視詳情。