TypeScript跑通第一個MCP，使用Deepseek查詢sqlite資料庫

最近最火的 AI 技術應該就是 MCP 了，還不了解 MCP 概念的建議查看Model Context Protocol 文檔，其中詳細介紹了 MCP 的概念和規範。本文使用官方的 TypeScript SDK，請求到 DeepSeek，從服務端到客戶端簡單實現一個 Sqlite MCP，順利開始 MCP 開發。

初始化項目#

本項目所有資源都在這個 git 倉庫TypeScript-MCP-Sqlite-Quickstart，包括了案例使用到的 sqlite 文件，你可以直接拉取，或者按照步驟初始化，自己創建一個數據庫。

創建項目

創建文件夾
mkdir TypeScript-MCP-Sqlite-Quickstart
cd TypeScript-MCP-Sqlite-Quickstart

初始化npm
npm init -y

安裝相關依賴
npm install @modelcontextprotocol/sdk zod sqlite3 express @types/express openai
npm install -D @types/node typescript

touch index.ts
touch server/sqlite_stdio.ts
touch server/sqlite_sse.ts

配置 package.json 和 tsconfig.json package.json

{
  "name": "typescript-mcp-sqlite-quickstart",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "build": "tsc && chmod 755 build/index.js"
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "type": "module",
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.8.0",
    "@types/express": "^5.0.1",
    "express": "^5.1.0",
    "openai": "^4.91.1",
    "sqlite3": "^5.1.7",
    "zod": "^3.24.2"
  },
  "devDependencies": {
    "@types/node": "^22.13.17",
    "typescript": "^5.8.2"
  }
}

tsconfig.json

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./build",
    "rootDir": "./",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "include": ["index.ts","server/**/*"],
  "exclude": ["node_modules"]
}

編寫 MCP Server#

MCP Server 有兩種啟動方式，一種是 Stdio，一種是 SSE，在本節都會進行介紹。首先介紹 Stdio 方式。

Stdio Transport#

在./server/sqlite_stdio.ts 文件中添加

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import sqlite3 from "sqlite3";
import { promisify } from "util";
import { z } from "zod";

//創建一個MCPServer實例
export const sqliteServer = new McpServer({
  name: "SQLite Explorer",
  version: "1.0.0"
});

//初始化Sqlite
const getDb = () => {
  const db = new sqlite3.Database("database.db");
  return {
    all: promisify<string, any[]>(db.all.bind(db)),
    close: promisify(db.close.bind(db))
  };
};

//定義一個server
//第一個參數是tools名
//第二個參數是描述，向大模型介紹該工具的用途
//第三個參數是輸入參數
//第四個參數是調用的方法
sqliteServer.tool(
  "query",
  "這是用於執行 SQLite 查詢的工具,你可以使用它來執行任何有效的 SQL 查詢,例如SELECT sql FROM sqlite_master WHERE type='table'，或者SELECT * FROM table_name",
  { sql: z.string() },
  async ({ sql }) => {
    const db = getDb();
    try {
      const results = await db.all(sql);
      return {
        content: [{
          type: "text",
          text: JSON.stringify(results, null, 2)
        }]
      };
    } catch (err: unknown) {
      const error = err as Error;
      return {
        content: [{
          type: "text",
          text: `Error: ${error.message}`
        }],
        isError: true
      };
    } finally {
      await db.close();
    }
  }
);

//使用StdioTransport連接Server
const transport = new StdioServerTransport();
await sqliteServer.connect(transport);

使用 build 編譯 ts 文件，inspector是官方提供的一個 MCP server 檢查工具，使用 inspector 檢查 server 服務是否正常。

//編譯ts
npm run build


//運行檢查工具
npx @modelcontextprotocol/inspector

Stdio 運行結果

SSE Transport#

在./server/sqlite_sse.ts 中添加

import express, { Request, Response } from "express";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import sqlite3 from "sqlite3";
import { promisify } from "util";
import { z } from "zod";

//初始化一個MCPServer實例
const sqliteServer = new McpServer({
  name: "SQLite Explorer",
  version: "1.0.0"
});

//初始化Sqlite
const getDb = () => {
  const db = new sqlite3.Database("database.db");
  return {
    all: promisify<string, any[]>(db.all.bind(db)),
    close: promisify(db.close.bind(db))
  };
};

const app = express();

//定義一個server，該部分與stdio相同
sqliteServer.tool(
  "query",
  "這是用於執行 SQLite 查詢的工具,你可以使用它來執行任何有效的 SQL 查詢,例如SELECT sql FROM sqlite_master WHERE type='table'，或者SELECT * FROM table_name",
  { sql: z.string() },
  async ({ sql }) => {
    const db = getDb();
    try {
      const results = await db.all(sql);
      return {
        content: [{
          type: "text",
          text: JSON.stringify(results, null, 2)
        }]
      };
    } catch (err: unknown) {
      const error = err as Error;
      return {
        content: [{
          type: "text",
          text: `Error: ${error.message}`
        }],
        isError: true
      };
    } finally {
      await db.close();
    }
  }
);

//使用SSETransport連接Server
const transports: {[sessionId: string]: SSEServerTransport} = {};

//用於處理SSE連接的路由
app.get("/sse", async (_: Request, res: Response) => {
  const transport = new SSEServerTransport('/messages', res);
  transports[transport.sessionId] = transport;
  res.on("close", () => {
    delete transports[transport.sessionId];
  });
  await sqliteServer.connect(transport);
});

//用於處理多個連接的路由
app.post("/messages", async (req: Request, res: Response) => {
  const sessionId = req.query.sessionId as string;
  const transport = transports[sessionId];
  if (transport) {
    await transport.handlePostMessage(req, res);
  } else {
    res.status(400).send('No transport found for sessionId');
  }
});

app.listen(3001);

完成 server 編寫後，同樣要進行編譯，還需要運行 express 服務，然後使用 Inspector 連接測試功能是否正常。

//編譯ts
npm run build

//運行express
node ./build/server/sqlite_sse.js

打開 Inspector，選擇 transport type 為 sse，設置 url 為http://localhost:3001/sse，點擊連接，測試工具。

MCP 客戶端編寫#

MCP 客戶端同樣支持兩種 transport，一種是 stdio，一種是 sse，與服務端相對應。在 index.ts 中添加

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

//使用StdioTransport連接Server
//這裡填寫的參數與Inspector測試中填寫的參數一致
const stdioTransport = new StdioClientTransport({
  command: "node",
  args: ["./build/server/sqlite_stdio.js"]
});

//使用SSETransport連接Server
//這裡填寫的參數與Inspector測試中填寫的參數一致
//const sseTransport = new SSEClientTransport(new URL("http://localhost:3001/sse"));

const client = new Client(
  {
    name: "example-client",
    version: "1.0.0"
  },
  {
    capabilities: {
      prompts: {},
      resources: {},
      tools: {}
    }
  }
);

//連接StdioTransport和SSETransport,不過貌似只能同時連接一個，
//按需選擇連接方式
await client.connect(stdioTransport);
// await client.connect(sseTransport);

//獲取工具列表
const tools = await client.listTools();
console.log("Tools:", tools);

//獲取資源列表
// const resources = await client.listResources();

//獲取提示列表
// const prompts = await client.listPrompts();

編寫完 MCP 客戶端後，編譯 ts，運行 MCP 客戶端，如果列出可用工具列表，則說明客戶端運行正常。

//編譯ts
npm run build

//運行客戶端
node build/index.js

向 Deepseek 發送請求調用 MCP#

deepseek 可以使用 openai sdk 發送請求，在發送請求的時候在參數中帶上 tools 即可。由於 openai 的 Tools 參數定義和 anthropic 的 tools 參數定義不同，在獲取到 tools 後按需進行轉換。 轉換函數

const toolsResult = await client.listTools();

//anthropic格式的tools
const anthropicTools = toolsResult.tools.map((tool) => {
        return {
          name: tool.name,
          description: tool.description,
          input_schema: tool.inputSchema,
        };
});

//openai格式的tools
const openaiTools = toolsResult.tools.map((tool) => {
        return {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema,
            }
        }
})

至此，已經成功獲取 tools 信息，只需要在請求 deepseek 的時候帶上 tools 參數，然後監聽 response 中的 finish_reson，如果為 tools_calls，則獲取 response 中的 toll_calls 中的 name 和 args。然後調用 MCP 客戶端的 callTool 方法，將獲取到的結果重新發送給 Deepseek，Deepseek 就成功獲取到了數據庫中的信息。

const openai = new OpenAI({
  apiKey: 'apikey',  // 使用你的apikey，建議從環境變量讀取
  baseURL: 'https://api.deepseek.com/v1',
});

//替換成你的問題
const messages: ChatCompletionMessageParam[] = [{ role: "user", content: "我有一個sqlite數據庫，數據庫中有expenses表和incomes表，告訴我我的收入和支出分別是多少？" }];

// 发送第一次请求，带上tools参数
const response = await openai.chat.completions.create({
        model: "deepseek-chat",
        messages: messages,
        max_tokens: 1000,
        tools: openaiTools,
  }
);

const content = response.choices[0];
//監聽finish_reason，如果是tool_calls，說明大模型調用了工具
console.log("finish_resaon",content.finish_reason);
if (content.finish_reason === "tool_calls" && content.message.tool_calls && content.message.tool_calls.length > 0) {
    //獲取大模型返回工具調用的參數    
      const tool_call = content.message.tool_calls[0];
      const toolName = tool_call.function.name;
      const toolArgs = JSON.parse(tool_call.function.arguments) as { [x: string]: unknown } | undefined;


      const result = await client.callTool({
        name: toolName,
        arguments: toolArgs,
      });

      console.log(`[大模型調用工具 ${toolName}  參數為 ${JSON.stringify(toolArgs)}]`)

      //將獲取到的結果添加到messages中，準備發送給大模型 
      messages.push({
        role: "user",
        content: result.content as string,
      });

      //將獲取到的結果添加到messages中，再發送一次請求，可以再在這個請求中帶上tools，實現多輪調用，但是也需要更好的流程處理邏輯
      const response = await openai.chat.completions.create({
        model: "deepseek-chat",
        max_tokens: 1000,
        messages,
      });

    console.log(response.choices[0].message.content);
}

至此已經完成了 Deepseek 通過 MCP 獲取數據庫信息，使用以下指令查看調用情況。

//編譯ts
npm run build

//運行index.js
node  ./build/index.js

運行結果

下一步#

至此，一個最簡單的 MCP 流程已經完成，有非常多的地方可以完善。

這個流程提供了最簡單的 sql 查詢功能。server 描述做的也比較粗糙，并不是每一次大模型都能成功調用 tool
目前的請求方式是非流式的，可以改進為流式
目前只支持一輪調用工具，無法實現多輪調用，允許 MCP 多輪調用能夠更好發揮 MCP 的能力。
目前一個客戶端好像只支持連接一個服務端，在有多個服務的情況下，如何高效的管理客戶端也是一個問題。