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 的能力。
目前一个客户端好像只支持连接一个服务端，在有多个服务的情况下，如何高效的管理客户端也是一个问题。