從零開始教你打造一個MCP客戶端

↓推薦關注↓

阿里妹導讀

Anthropic開源了一套MCP協議，它為連線AI系統與資料來源提供了一個通用的、開放的標準，用單一協議取代了碎片化的整合方式。本文教你從零打造一個MCP客戶端。

一、背景

如何讓大語言模型與外部系統互動，一直是AI系統需要解決的問題：

Plugins：OpenAI推出ChatGPT Plugins，首次允許模型透過外掛與外部應用互動。外掛功能包括即時資訊檢索（如瀏覽器訪問）、程式碼直譯器（Code Interpreter）執行計算、第三方服務呼叫（如酒店預訂、外賣服務等）

Function Calling：Function Calling技術逐步成熟，成為大模型與外部系統互動的核心方案。

Agent框架 Tools: 模型作為代理（Agent），動態選擇工具完成任務，比如langchain的Tool。

一個企業，面對不同的框架或系統，可能都需要參考他們的協議，去開發對應Tool，這其實是一個非常重複的工作。

面對這種問題，Anthropic開源了一套MCP協議（Model Context Protocol），

https://www.anthropic.com/news/model-context-protocol

https://modelcontextprotocol.io/introduction

它為連線AI系統與資料來源提供了一個通用的、開放的標準，用單一協議取代了碎片化的整合方式。其結果是，能以更簡單、更可靠的方式讓人工智慧系統獲取所需資料。

二、架構

MCP Hosts：像 Claude Desktop、Cursor這樣的程式，它們透過MCP訪問資料。
MCP Clients：與伺服器保持 1:1 連線的協議客戶端。
MCP Servers：輕量級程式，每個程式都透過標準化的模型上下文協議公開特定功能。

結合AI模型，以一個Java應用為例，架構是這樣：

可以看到傳輸層有兩類：

StdioTransport
HTTP SSE

三、實現MCP Server

首先看一個最簡單的MCP Server例子：

import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";import { z } from "zod";// Create an MCP serverconst server = new McpServer({ name: "Demo", version: "1.0.0"});// Add an addition toolserver.tool("add",'Add two numbers', { a: z.number(), b: z.number() }, async ({ a, b }) => ({ content: [{ type: "text", text: String(a + b) }] }));async function main(){// Start receiving messages on stdin and sending messages on stdoutconst transport = new StdioServerTransport(); await server.connect(transport);}main()

程式碼頭部和底部都是一些樣板程式碼，主要變化的是在tool這塊，這個聲明瞭一個做加法的工具。這就是一個最簡單的可執行的Server了。

同時也可以使用官方的腳手架，來建立一個完整複雜的Server：

npx @modelcontextprotocol/create-server my-server

3.1 使用SDK

從上面程式碼可以看到很多模組都是從@modelcontextprotocol/sdk 這個SDK裡匯出的。

SDK封裝好了協議內部細節（JSON-RPC 2.0），包括架構分層，開發者直接寫一些業務程式碼就可以了。

https://github.com/modelcontextprotocol/typescript-sdk

MCP伺服器可以提供三種主要功能型別：

Resources:可以由客戶端讀取的類似檔案的資料（例如API響應或檔案內容）
Tools:LLM可以呼叫的功能（在使用者批准下）
Prompts:可幫助使用者完成特定任務的預先編寫的模板

Resources和Prompts可以讓客戶端喚起，供使用者選擇，比如使用者所有的筆記，或者最近訂單。

重點在Tools，其他很多客戶端都不支援。

3.2 除錯

如果寫好了程式碼，怎麼除錯這個Server呢？官方提供了一個偵錯程式：

npx @modelcontextprotocol/inspector

1.連線Server

2.獲取工具

3.執行除錯

3.3 在客戶端使用

如果執行結果沒錯，就可以上架到支援MCP協議的客戶端使用了，比如Claude、Cursor，這裡以Cursor為例：

在Cursor Composer中對話，會自動識別這個Tool，並尋求使用者是否呼叫

點選執行，就可以呼叫執行：

3.4 HTTP SSE型別Server

import express from "express";import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";import { z } from "zod";const server = new McpServer({ name: "demo-sse", version: "1.0.0"});server.tool("exchange",'人民幣匯率換算', { rmb: z.number() }, async ({ rmb }) => {// 使用固定匯率進行演示，實際應該調用匯率APIconst usdRate = 0.14; // 1人民幣約等於0.14美元const hkdRate = 1.09; // 1人民幣約等於1.09港幣const usd = (rmb * usdRate).toFixed(2);const hkd = (rmb * hkdRate).toFixed(2);return { content: [{ type: "text", text: `${rmb}人民幣等於:\n${usd}美元\n${hkd}港幣` }] } },);const app = express();const sessions: Record<string, { transport: SSEServerTransport; response: express.Response }> = {}app.get("/sse", async (req, res) => { console.log(`New SSE connection from ${req.ip}`);const sseTransport = new SSEServerTransport("/messages", res);const sessionId = sseTransport.sessionId;if (sessionId) { sessions[sessionId] = { transport: sseTransport, response: res } } await server.connect(sseTransport);});app.post("/messages", async (req, res) => {const sessionId = req.query.sessionId as string;const session = sessions[sessionId];if (!session) { res.status(404).send("Session not found");return; } await session.transport.handlePostMessage(req, res);});app.listen(3001);

核心的差別在於需要提供一個sse服務，對於Tool基本一樣，但是sse型別就可以部署在服務端了。上架也和command型別相似：

3.5 一個複雜一點的例子

操作瀏覽器執行自動化流程。

可以操作瀏覽器，Cursor秒變Devin。想象一下，寫完程式碼，編輯器自動開啟瀏覽器預覽效果，然後截圖給視覺模型，發現樣式不對，自動修改。

如果對接好內部系統，貼一個需求地址，自動連線瀏覽器，開啟網頁，分析需求，分析視覺稿，然後自己寫程式碼，對比視覺稿，你就喝杯咖啡，靜靜的看著它工作。

3.6 MCP Server資源

有很多寫好的Server，可以直接複用。

https://github.com/modelcontextprotocol/servers
https://github.com/punkpeye/awesome-mcp-servers/blob/main/README-zh.md

四、實現MCP Client

一般MCP Host以一個Chat box為入口，對話形式去呼叫。

那我們怎麼在自己的應用裡支援MCP協議呢？這裡需要實現MCP Client。

4.1 配置檔案

使用配置檔案來標明有哪些MCP Server，以及型別。

const config = [ { name: 'demo-stdio', type: 'command', command: 'node ~/code-open/cursor-toolkits/mcp/build/demo-stdio.js', isOpen: true }, { name: 'weather-stdio', type: 'command', command: 'node ~/code-open/cursor-toolkits/mcp/build/weather-stdio.js', isOpen: true }, { name: 'demo-sse', type: 'sse', url: 'http://localhost:3001/sse', isOpen: false }];exportdefault config;

4.2 確認互動形態

MCP Client主要還是基於LLM，識別到需要呼叫外部系統，呼叫MCP Server提供的Tool，所以還是以對話為入口，可以方便一點，直接在terminal裡對話，使用readline來讀取使用者輸入。大模型可以直接使用openai，Tool的路由直接使用function calling。

4.3 編寫Client

大致的邏輯：

1.讀取配置檔案，執行所有Server，獲取可用的Tools

2.使用者與LLM對話（附帶所有Tools名稱描述，引數定義）

3.LLM識別到要執行某個Tool，返回名稱和引數

4.找到對應Server的Tool，呼叫執行，返回結果

5.把工具執行結果提交給LLM

6.LLM返回分析結果給使用者

使用SDK編寫Client程式碼

import { Client } from "@modelcontextprotocol/sdk/client/index.js";import { StdioClientTransport, StdioServerParameters } from "@modelcontextprotocol/sdk/client/stdio.js";import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";import OpenAI from "openai";import { Tool } from "@modelcontextprotocol/sdk/types.js";import { ChatCompletionMessageParam } from "openai/resources/chat/completions.js";import { createInterface } from "readline";import { homedir } from 'os';import config from "./mcp-server-config.js";// 初始化環境變數const OPENAI_API_KEY = process.env.OPENAI_API_KEY;if (!OPENAI_API_KEY) {thrownew Error("OPENAI_API_KEY environment variable is required");}interface MCPToolResult { content: string;}interface ServerConfig { name: string; type: 'command' | 'sse'; command?: string; url?: string; isOpen?: boolean;}classMCPClient {staticgetOpenServers(): string[] {return config.filter(cfg => cfg.isOpen).map(cfg => cfg.name); }private sessions: Map<string, Client> = new Map();private transports: Map<string, StdioClientTransport | SSEClientTransport> = new Map();private openai: OpenAI; constructor() {this.openai = new OpenAI({ apiKey: OPENAI_API_KEY }); }async connectToServer(serverName: string): Promise<void> {const serverConfig = config.find(cfg => cfg.name === serverName) as ServerConfig;if (!serverConfig) {thrownew Error(`Server configuration not found for: ${serverName}`); } let transport: StdioClientTransport | SSEClientTransport;if (serverConfig.type === 'command' && serverConfig.command) { transport = await this.createCommandTransport(serverConfig.command); } elseif (serverConfig.type === 'sse' && serverConfig.url) { transport = await this.createSSETransport(serverConfig.url); } else {thrownew Error(`Invalid server configuration for: ${serverName}`); }const client = new Client( { name: "mcp-client", version: "1.0.0" }, { capabilities: { prompts: {}, resources: {}, tools: {} } } ); await client.connect(transport);this.sessions.set(serverName, client);this.transports.set(serverName, transport);// 列出可用工具const response = await client.listTools();

        console.log(`\nConnected to server '${serverName}' with tools:`, response.tools.map((tool: Tool) => tool.name));

}private async createCommandTransport(shell: string): Promise<StdioClientTransport> {const [command, ...shellArgs] = shell.split(' ');if (!command) {thrownew Error("Invalid shell command"); }// 處理引數中的波浪號路徑const args = shellArgs.map(arg => {if (arg.startsWith('~/')) {return arg.replace('~', homedir()); }return arg; });const serverParams: StdioServerParameters = { command, args, env: Object.fromEntries( Object.entries(process.env).filter(([_, v]) => v !== undefined) ) as Record<string, string> };returnnew StdioClientTransport(serverParams); }private async createSSETransport(url: string): Promise<SSEClientTransport> {returnnew SSEClientTransport(new URL(url)); }async processQuery(query: string): Promise<string> {if (this.sessions.size === 0) {thrownew Error("Not connected to any server"); }const messages: ChatCompletionMessageParam[] = [ { role: "user", content: query } ];// 獲取所有伺服器的工具列表const availableTools: any[] = [];for (const [serverName, session] of this.sessions) {const response = await session.listTools();const tools = response.tools.map((tool: Tool) => ({ type: "function" as const, function: { name: `${serverName}__${tool.name}`, description: `[${serverName}] ${tool.description}`, parameters: tool.inputSchema } })); availableTools.push(...tools); }// 呼叫OpenAI APIconst completion = await this.openai.chat.completions.create({ model: "gpt-4-turbo-preview", messages, tools: availableTools, tool_choice: "auto" });const finalText: string[] = [];// 處理OpenAI的響應for (const choice of completion.choices) {const message = choice.message;if (message.content) { finalText.push(message.content); }if (message.tool_calls) {for (const toolCall of message.tool_calls) {const [serverName, toolName] = toolCall.function.name.split('__');const session = this.sessions.get(serverName);if (!session) { finalText.push(`[Error: Server ${serverName} not found]`);continue; }const toolArgs = JSON.parse(toolCall.function.arguments);// 執行工具呼叫const result = await session.callTool({ name: toolName, arguments: toolArgs });const toolResult = result as unknown as MCPToolResult;

                    finalText.push(`[Calling tool ${toolName} on server ${serverName} with args ${JSON.stringify(toolArgs)}]`);

console.log(toolResult.content); finalText.push(toolResult.content);// 繼續與工具結果的對話 messages.push({ role: "assistant", content: "", tool_calls: [toolCall] }); messages.push({ role: "tool", tool_call_id: toolCall.id, content: toolResult.content });// 獲取下一個響應const nextCompletion = await this.openai.chat.completions.create({ model: "gpt-4-turbo-preview", messages, tools: availableTools, tool_choice: "auto" });if (nextCompletion.choices[0].message.content) { finalText.push(nextCompletion.choices[0].message.content); } } } }return finalText.join("\n"); }async chatLoop(): Promise<void> { console.log("\nMCP Client Started!"); console.log("Type your queries or 'quit' to exit.");const readline = createInterface({ input: process.stdin, output: process.stdout });const askQuestion = () => {returnnew Promise<string>((resolve) => { readline.question("\nQuery: ", resolve); }); };try {while (true) {const query = (await askQuestion()).trim();if (query.toLowerCase() === 'quit') {break; }try {const response = await this.processQuery(query); console.log("\n" + response); } catch (error) { console.error("\nError:", error); } } } finally { readline.close(); } }async cleanup(): Promise<void> {for (const transport of this.transports.values()) { await transport.close(); }this.transports.clear();this.sessions.clear(); } hasActiveSessions(): boolean {returnthis.sessions.size > 0; }}// 主函式async function main(){const openServers = MCPClient.getOpenServers(); console.log("Connecting to servers:", openServers.join(", "));const client = new MCPClient();try {// 連線所有開啟的伺服器for (const serverName of openServers) {try { await client.connectToServer(serverName); } catch (error) { console.error(`Failed to connect to server '${serverName}':`, error); } }if (!client.hasActiveSessions()) {thrownew Error("Failed to connect to any server"); } await client.chatLoop(); } finally { await client.cleanup(); }}// 執行主函式main().catch(console.error);

4.4 執行效果

NODE_TLS_REJECT_UNAUTHORIZED=0 node build/client.js

NODE_TLS_REJECT_UNAUTHORIZED=0 可以忽略（不校驗證書）

4.5 時序圖