Posted by on | Featured

模型上下文协议正在改变人工智能应用程序与外部工具及数据之间的连接方式。虽然一些教程仅停留在“连接到GitHub”或“读取文件”这样的内容上，但模型上下文协议的真正价值在于能够以结构化、安全的方式，让人工智能助手访问你的内部数据——包括数据库、内部API、知识库以及专有系统。

目录

预备知识

这是一本面向高级读者的指南。在开始学习之前，你应当已经掌握以下内容：

  • TypeScript / Node.js

  • REST API与服务器端开发

  • 对大语言模型及工具调用的基本了解

  • 熟悉JSON-RPC等协议

什么是MCP？它为何对内部数据管理如此重要？

MCP是一种由Anthropic开发的开放协议，它规范了人工智能助手发现和调用外部工具的方式。可以将它理解为人工智能领域的“USB-C接口”——一种标准化的连接方式，使得任何人工智能模型都能连接到任何数据源。

在MCP出现之前，要将人工智能助手与企业的内部数据库连接起来，需要做到以下几点：

  • 为每个大语言模型提供商编写自定义的工具定义

  • 将数据访问逻辑直接编码到人工智能应用程序中

  • 每当更换模型或添加新的数据源时，都需要重新进行相关配置

MCP将数据层与人工智能层分离开来。MCP服务器会暴露各种工具和资源，任何兼容MCP的客户端——无论是Claude、ChatGPT，还是你自定义开发的应用程序——都可以直接使用这些资源而无需进行任何修改。

对于内部数据管理而言，这一点意义重大，因为：

  • 通过同一个协议，你的CRM系统、ERP系统、工单管理系统以及维基页面等都可以被人工智能访问

  • 访问控制机制集中在MCP服务器中，而不会分散在各个人工智能应用程序的代码中

  • 新的人工智能模型或客户端可以自动获得访问权限，无需重新配置集成方案

  • 工具定义位于数据源附近，因此更便于维护和版本管理

架构概述

下面是我们正在构建的系统结构：

MCP服务器架构：通过HTTP/SSE/stdio协议及JSON-RPC技术，将人工智能客户端与PostgreSQL数据库、内部API以及文件存储系统连接起来。

MCP服务器位于人工智能客户端与内部系统之间，它负责以下功能：

  • 工具发现：告诉人工智能哪些操作是可用的

  • 参数验证：确保人工智能发送正确的输入数据

  • 数据访问

  • 响应格式化：返回结构化的数据，以便人工智能能够对其进行处理

  • 身份验证：确认发起请求的用户身份

项目设置步骤

让我们来构建一个MCP服务器，让它能够暴露企业内部的员工目录和项目管理系统。

mkdir internal-data-mcp && cd internal-data-mcp
npm init -y
npm install @modelcontextprotocol/sdk zod express pg
npm install -D typescript @types/node @types/express @types/pg tsx

这些命令为项目的构建提供了基础框架。`npm install`命令用于下载运行时所需的依赖项：官方的MCP SDK、用于进行模式验证的Zod工具、用于搭建HTTP服务器的Express框架，以及用于连接PostgreSQL数据库的`pg`模块。`-D`选项会将TypeScript及其类型定义作为仅适用于开发环境的依赖项进行安装——这些依赖项是编译代码所必需的，但不会被部署到生产环境中。使用`tsx`命令，可以在开发过程中直接运行TypeScript代码，而无需再进行单独的编译步骤。

现在，创建你的 `tsconfig.json` 文件：

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "declaration": true
  },
  "include": ["src/**/*"]
}

这个 TypeScript 配置文件的目标是 ES2022，该版本支持诸如顶层 `await` 这样的现代 JavaScript 特性。当使用 MCP SDK 的 `.js` 导入扩展时，必须设置 `"module": "Node16"` 和 `"moduleResolution": "Node16"`。配置 `"strict": true` 可以启用 TypeScript 的所有严格检查功能，这样就能在代码进入生产环境之前及时发现其中的错误。outDir/rootDir 的设置告诉编译器从 `src/` 目录中获取源文件，并将编译后的 JavaScript 代码输出到 `dist/` 目录中。

构建 MCP 服务器

步骤 1：创建服务器框架

创建文件 `src/server.ts`：

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

const server = new McpServer(
  { name: "internal-data", version: "1.0.0" },
  { capabilities: { tools: {}, resources: {} } }
);

官方 SDK 中提供的 `McpServer` 类负责处理 JSON-RPC 协议、传输协议协商以及生命周期管理。我们配置了该服务器对 tools（AI 可以执行的操作）和 resources（AI 可以读取的数据）的支持。

步骤 2：连接内部数据源

假设你有一个包含员工信息和项目数据的 PostgreSQL 数据库，那么就需要创建一个数据访问层：

// src/db.ts
import pg from "pg";

const pool = new pg.Pool({
  connectionString: process.env.INTERNAL_DB_URL,
  max: 10,
  idleTimeoutMillis: 30000,
});

export interface Employee {
  id: string;
  name: string;
  email: string;
  department: string;
  role: string;
  manager_id: string | null;
  start_date: string;
}

export interface Project {
  id: string;
  name: string;
  status: "active" | "completed" | "on_hold";
  lead_id: string;
  department: string;
  deadline: string | null;
}

export async function searchEmployees(
  query: string,
  department?: string
): Promise<Employee[]>> {
  const conditions = ["(name ILIKE \(1 OR email ILIKE \)1 OR role ILIKE $1)"];
  const params: string[] = [`%${query}%`];

  if (department) {
    conditions.push(`department = $${params.length + 1}`);
    params.push(department);
  }

  const result = await pool.query<Employee>(
    `SELECT id, name, email, department, role, manager_id, start_date
     FROM employees
     WHERE ${conditions.join(" AND ")}
     ORDER BY name
     LIMIT 25`,
    params
  );

  return result.rows;
}

export async function getProjectsByStatus(
  status: string
): Promise<Project[]>> {
  const result = await pool.query<Project>(
    `SELECT id, name, status, lead_id, department, deadline
     FROM projects
     WHERE status = $1
     ORDER BY deadline ASC NULLS LAST`,
    [status]
  );

  return result.rows;
}

export async function getProjectMembers(
  projectId: string
): Promise<Employee[]>> {
  const result = await pool.query<Employee>(
    `SELECT e.id, e.name, e.email, e.department, e.role,
            e.manager_id, e.start_date
     FROM employees e
     JOIN project_members pm ON pm.employee_id = e.id
     WHERE pm.project_id = $1
     ORDER BY e.name`,
    [projectId]
  );

  return result.rows;
}

请注意，这里使用的是带有参数化查询的普通SQL。您的MCP服务器的数据访问层应该采用团队已经在使用的技术——无论是Prisma、Drizzle、Knex还是原始SQL。MCP并不会规定您必须使用哪种数据访问方式。

步骤3：定义工具

现在，需要通过MCP工具来暴露这些数据。在这个阶段，设计显得尤为重要。合理的工具定义会直接影响到AI对数据的利用效果。

// src/tools.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import {
  searchEmployees,
  getProjectsByStatus,
  getProjectMembers,
} from "./db.js";

export function registerTools(server: McpServer) {
  // 工具1：搜索员工信息
  server.tool(
    "search_employees",
    `通过姓名、电子邮件或职位来搜索内部员工信息。
     返回匹配的员工信息，包括他们所属的部门及汇报关系。
     当用户询问有关人员、团队或组织结构的信息时，可以使用此工具。",
    {
      query: z
        .string()
        .describe("搜索条件：员工姓名、电子邮件或职位"),
      department: z
        .string()
        .optional()
        .describe(
          "按部门名称过滤（例如：'工程部'、'市场部'）
        ),
    },
    async ({ query, department }) => {
      const employees = await searchEmployees(query, department);

      if (employees.length === 0) {
        return {
          content: [
            {
              type: "text",
              text: `未找到与 "\({query}"\){department ? ` 在 ${department}` : ""} 匹配的员工。`,
            },
          ],
        };
      }

      const formatted = employees
        .map(
          (e) =>
            `- **\({e.name}** (\){e.email})\n  职位：\({e.role} | 部门：\){e.department} | 入职时间：${e.start_date}`
        )
        .join("\n");

      return {
        content: [
          {
            type: "text",
            text: `找到了 \({employees.length} 名员工信息:\n\n\){formatted}`,
          },
        ],
      };
    }
  );

  // 工具2：按状态列出项目
  server.tool(
    "list_projects",
    `按状态列出内部项目。
     返回项目名称、负责人、所属部门及截止日期。
     当用户询问正在进行的工作、项目状态或截止日期时，可以使用此工具。",
    {
      status: z
        .enum(["active", "completed", "on_hold"])
        .describe("用于过滤的项目状态"),
    },
    async ({ status }) => {
      const projects = await getProjectsByStatus(status);

      if (projects.length === 0) {
        return {
          content: [
            {
              type: "text",
              text: `未找到状态为 ${status} 的项目。`,
            },
          ],
        };
      }

      const formatted = projects
        .map(
          (p) =>
            `- **\({p.name}** [\){p.status}]\n  负责人：\({p.lead_id} | 部门：\){p.department} | 截止日期：${p.deadline ?? "无"}`
        )
        .join("\n");

      return {
        content: [
          {
            type: "text",
            text: `\({projects.length} 个状态为 ${status} 的项目:\n\n${formatted}`,
          },
        ],
      };
    }
  );

  // 工具3：获取某个项目的团队成员信息
  server.tool(
    "get_project_team",
    `获取分配给特定项目的所有团队成员信息。
     返回每个成员的详细资料。
     当用户询问谁在参与某个项目时，可以使用此工具。",
    {
      project_id: z
        .string()
        .uuid()
        .describe("要查询的项目的UUID"),
    },
    async ({ project_id }) => {
      const members = await getProjectMembers(project_id);

      if (members.length === 0) {
        return {
          content: [
            {
              type: "text",
              text: "未找到该项目的团队成员信息。",
            },
          ],
        };
      }

      const formatted = members
        .map((m) => `- \({m.name} (\){m.role}, ${m.department})`)
        .join("\n");

      return {
        content: [
          {
            type: "text",
            text: `项目团队（共 \({members.length} 人）:\n\n\{{formatted}}`,
          },
        ],
      };
    }
  );
}

server.tool() 方法会使用四个参数来注册每个工具：工具名称、人工智能能够读取的英文描述（用于判断何时调用该工具）、定义参数结构的 Zod 数据模型，以及当工具被调用时执行的异步处理函数。处理函数会接收经过验证且类型明确的参数——在处理函数执行之前，Zod 会先拒绝所有格式错误的输入。每个处理函数都会返回一个 content 数组；其中类型为 “text” 的数据表示人工智能客户端应将该响应视为可读文本。如果需要返回空结果（即没有匹配项），也必须进行明确的处理，这样人工智能才能得到有用的信息，而不会收到一个可能被误解读的空数组。

工具设计原则

有三点因素会决定人工智能使用某个工具的效率：

1. 描述性强的名称和说明。人工智能会根据这些描述来决定调用哪个工具。因此，需要明确说明该工具在什么情况下可以使用，而不仅仅是它能做什么。以下是具体的对比示例：

// 描述模糊——人工智能不知道何时应该使用这个工具
"搜索员工"

// 描述具体——人工智能能清楚地知道在什么情况下使用这个工具
“可以通过姓名、电子邮件或职位来搜索内部员工目录。
当用户询问有关人员、团队或组织结构的信息时，可以使用这个工具。”

2>带有说明的类型明确的参数。应为每个参数使用 Zod 的 .describe() 方法。这样人工智能才能清楚了解每个字段应该包含什么类型的数据：

// 如果不使用描述，人工智能就会不知道“query”字段应该是什么格式
{ query: z.string() }

// 使用描述后，人工智能就能清楚地知道应该传递什么数据
{ query: z.string().describe("搜索词：员工姓名、电子邮件或职位名称") }

3>结构化的返回值。应以人工智能能够理解的格式来返回数据。建议使用 Markdown 表格或结构化列表，而不是原始的 JSON 数据。与深度嵌套的对象相比，人工智能处理结构化文本的能力更强。

步骤 4：暴露资源

资源是人工智能可以将其纳入自身处理流程中的只读数据。与工具不同（工具是在推理过程中被调用的），资源通常会提前被加载进来，以便为后续的处理提供背景信息。

// src/resources.ts
import {
  McpServer,
  ResourceTemplate,
} from "@modelcontextprotocol/sdk/server/mcp.js";

export function registerResources(server: McpServer) {
  // 静态资源：组织结构图概览
  server.resource(
    "org-structure",
    "internal://org-structure",
    {
      description:
        "包含各部门及领导层信息的组织结构概览",
      mimeType: "text/markdown",
    },
    async (uri) => ({
      contents: [
        {
          uri: uri.href,
          mimeType: "text/markdown",
          text: await generateOrgOverview(),
        },
      ],
    })
  );

  // 动态资源模板：部门详细信息
  server.resource(
    "department-info",
    new ResourceTemplate("internal://departments/{name}", {
      list: undefined,
    }),
    {
      description: "特定部门的详细信息",
      mimeType: "text/markdown",
    },
    async (uri, variables) => ({
      contents: [
        {
          uri: uri.href,
          mimeType: "text/markdown",
          text: await getDepartmentDetails(
            variables.name as string
          ),
        },
      ],
    })
  );
}

server.resource() 方法用于注册两种类型的资源。第一种资源使用固定的 URI 地址（internal://org-structure），这种资源是 AI 可以通过名称来请求的静态资源；第二种资源则使用 ResourceTemplate，该模板定义了一个包含 {name} 占位符的 URI 模式——AI 可以请求 internal://departments/Engineering，而在运行时，variables.name 这个参数会被填充为 “Engineering”。这两种资源都会返回一个 contents 数组，其 mimeType 属性值为 “text/markdown” —— 这告诉客户端应该如何渲染响应内容。与工具不同，资源的作用是作为背景信息被读取，而不是被用来执行特定的操作。

那些用于提供背景信息而非回答具体问题的数据，都非常适合使用资源这种形式来存储，例如公司政策、API 文档、数据库架构以及配置参考资料等。

步骤 5：传输与启动

MCP 支持多种传输方式。对于内部数据服务器来说，通常会使用以下两种方式之一：

Streamable HTTP —— 这是推荐用于远程服务器的传输方式（它取代了旧版的 SSE 传输方式）：

// src/index.ts
import express from "express";
import { randomUUID } from "node:crypto";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { registerTools } from "./tools.js";
import { registerResources } from "./resources.js";

const app = express();
app.use(express.json());

const server = new McpServer(
  { name: "internal-data", version: "1.0.0" },
  { capabilities: { tools: {}, resources: {} } }
);

registerTools(server);
registerResources(server);

// 根据会话 ID 存储传输对象
const transports = new Map();

// 在一个统一的端点处理所有 MCP 请求
app.all("/mcp", async (req, res) => {
  // 检查是否存在当前会话
  const sessionId = req.headers["mcp-session-id"] as string | undefined;

  if (sessionId && transports.has(sessionId)) {
    // 如果存在当前会话，就使用对应的传输方式处理请求
    const transport = transports.get(sessionId);
    await transport.handleRequest(req, res);
    return;
  }

  if (sessionId && !transports.has(sessionId)) {
    // 未知的会话 ID
    res.status(404).json({ error: "未找到相应的会话" });
    return;
  }

  // 如果是新会话，就创建对应的传输对象并进行连接
  const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: () => randomUUID(),
    onsessioninitialized: (id) => {
      transports.set(id, transport);
    },
  });

  transport.onclose = () => {
    if (transport.sessionId) {
      transports.delete(transport.sessionId);
    }
  };

  await server.connect(transport);
  await transport.handleRequest(req, res);
});

app.listen(3100, () => {
  console.log("MCP 服务器正在运行，地址为 http://localhost:3100/mcp");
});

这种方式设置了一个统一的 `/mcp` 端点，用于处理所有与 MCP 相关的通信请求。当有新的客户端连接时（如果该客户端没有发送 `mcp-session-id` 头部信息），系统会创建一个 `StreamableHTTPServerTransport` 对象，并将其存储在以生成的 UUID 作为键值的 `transports` 映射中。在后续的请求中，服务器会根据请求头中的会话 ID 来查找对应的传输对象，并将请求路由到该对象上——正是通过这种机制，服务器才能同时为多个客户端维护会话状态。当某个会话结束时，`transport.onclose` 方法会清除映射中的相应条目，从而避免内存泄漏。而另一种替代方案 `StdioServerTransport`（如下所示）则完全省略了这些复杂步骤：它直接从标准输入读取数据并写入标准输出，这也是 Claude Desktop 通过创建子进程来启动本地服务器的方式。

Stdio——适用于本地开发环境，或者当 MCP 客户端以子进程的形式运行服务器时：


// src/stdio.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { registerTools } from "./tools.js";
import { registerResources } from "./resources.js";

const server = new McpServer(
  { name: "internal-data", version: "1.0.0" },
  { capabilities: { tools: {}, resources: {} } }
);

registerTools(server);
registerResources(server);

const transport = new StdioServerTransport();
await server.connect(transport);

在生产环境中，对于内部数据交换来说，HTTP/SSE 几乎总是更合适的选择。而 Stdio 在开发过程中使用起来非常方便，尤其是当客户端和服务器运行在同一台机器上时。

添加身份验证机制

内部数据服务器必须具备身份验证功能——你肯定不希望网络中的任何 AI 客户端都能在未经授权的情况下访问你的员工数据库。

使用承载令牌进行身份验证

最简单的方法就是在每个请求中对令牌进行验证：


// src/auth-middleware.ts
import { Request, Response, NextFunction } from "express";

interface AuthenticatedRequest extends Request {
  userId?: string;
  orgId?: string;
}

export function authMiddleware(
  req: AuthenticatedRequest,
  res: Response,
  next: NextFunction
) {
  const authHeader = req.headersauthorization;

  if (!authHeader || !authHeader.startsWith("Bearer ")) {
    return res.status(401).json({ error: "缺少授权头部信息" });
  }

  const token = authHeader.slice(7);

  try {
    // 根据你的内部身份验证系统来验证该令牌
    const claims = validateInternalToken(token);
    req.userId = claims.sub;
    req.orgId = claims.org;
    next();
  } catch {
    return res.status(403).json({ error: "令牌无效" });
  }
}

function validateInternalToken(token: string) {
  // 请根据你的实际需求替换以下代码：
  // - 使用 JWT 技术来验证令牌
  // - 从数据库中查询对应的 API 密钥
  // - 或者通过 Redis 来验证会话令牌
  // 这里只是示例代码
  return { sub: "user-123", org: "org-456" };
}

中间件会在任何请求到达MCP处理程序之前，检查其中是否包含Authorization: Bearer 这一头部信息。validateInternalToken只是一个占位符——你需要用实际的验证逻辑来替换它：例如使用jsonwebtoken这样的库进行JWT验证，或者在数据库中查找API密钥，或者通过Redis检查会话令牌。经过验证后的信息会被添加到请求对象中（如req.userId、req.orgId），这样下游的处理程序就可以利用这些信息来控制访问权限。app.use("/mcp", authMiddleware)这一行代码确保了没有任何请求能够在未通过这项验证的情况下到达MCP端点。

将这段代码添加到你的Express应用中：

app.use("/mcp", authMiddleware);

MCP的OAuth 2.0支持

对于那些支持MCP内置OAuth流程的客户端（比如Claude Desktop），你可以实现完整的OAuth认证流程。MCP SDK提供了OAuthServerProvider接口，其中包含了以下必要的方法：

import type { OAuthServerProvider } from "@modelcontextprotocol/sdk/server/auth/provider.js";
import type {
  AuthorizationParams,
  OAuthClientInformationFull,
  OAuthRegisteredClientsStore,
  OAuthTokens,
  AuthInfo,
} from "@modelcontextprotocol/sdk/server/auth/types.js";

class InternalOAuthProvider implements OAuthServerProvider {
  // 用于存储已注册的OAuth客户端信息
  get clientsStore(): OAuthRegisteredClientsStore {
    return this._clientsStore;
  }

  private _clientsStore: OAuthRegisteredClientsStore = {
    async getClient(clientId: string) {
      // 在数据库中查找该客户端信息
      return db.getOAuthClient(clientId);
    },
    async registerClient(clientMetadata) {
      // 注册一个新的动态客户端
      return db.createOAuthClient(clientMetadata);
    },
  };

  // 将用户重定向到你的内部SSO系统进行认证
  async authorize(
    client: OAuthClientInformationFull,
    params: AuthorizationParams,
    res: Response
  ): Promise {
    const authUrl = new URL("https://sso.internal.company.com/authorize");
    authUrl.searchParams.set("client_id", client.client_id);
    authUrl.searchParams.set("redirect_uri", params.redirectUri);
    authUrl.searchParams.set("state", params.state ?? "");
    authUrl.searchParams.set("code_challenge", params.codeChallenge);
    // 直接在响应中设置重定向地址
    res Redirect(authUrl.toString());
  }

  // 根据给定的授权码生成PKCE挑战码
  async challengeForAuthorizationCode(
    _client: OAuthClientInformationFull,
    authorizationCode: string
  ): Promise {
    const session = await db.getSessionByCode(authorizationCode);
    return session.codeChallenge;
  }

  // 使用授权码换取访问令牌和刷新令牌
  async exchangeAuthorizationCode(
    client: OAuthClientInformationFull,
    authorizationCode: string,
    _codeVerifier?: string,
    _redirectUri?: string
  ): Promise {
    const response = await fetch("https://sso.internal.company.com/token",
      {
        method: "POST",
        headers: {
          "Content-Type": "application/x-www-form-urlencoded",
        },
        body: new URLSearchParams({
          grant_type: "authorization_code",
          code: authorizationCode,
          client_id: client.client_id,
        }),
      }
    );

    return response.json() as Promise;
  }

  // 刷新过期的令牌
  async exchangeRefreshToken(
    client: OAuthClientInformationFull,
    refreshToken: string
  ): Promise {
    const response = await fetch("https://sso.internal.company.com/token",
      {
        method: "POST",
        headers: {
          "Content-Type": "application/x-www-form-urlencoded",
        },
        body: new URLSearchParams({
          grant_type: "refresh_token",
          refresh_token: refreshToken,
          client_id: client.client_id,
        }),
      }
    );

    return response.json() as Promise;
  }

  // 在每个请求中验证访问令牌的有效性
  async verifyAccessToken(token: string): Promise {
    const response = await fetch("https://sso.internal.company.com/introspect",
      {
        method: "POST",
        headers: {
          "Content-Type": "application/x-www-form-urlencoded",
        },
        body: new URLSearchParams({ token }),
      }
    );

    const data = await response.json();
    if (!data.active) throw new Error("令牌已失效");

    return {
      token,
      clientId: data.client_id,
      scopes: data.scope?.split(" ") ?? [],
      expiresAt: data.exp,
    };
  }
}

InternalOAuthProvider 实现了 OAuthServerProvider 接口，MCP SDK 在 OAuth 流程的每个阶段都会调用这个接口。clientsStore 负责处理动态客户端注册工作——比如 Claude Desktop 这类 MCP 客户端在首次连接时就会进行自我注册。authorize() 会将用户重定向到您的内部 SSO 系统，并直接写入 Express 的响应结果中。challengeForAuthorizationCode() 会返回在授权会话开始时存储的 PKCE 代码挑战信息，这样就可以在不传输任何敏感数据的情况下验证令牌交换过程。exchangeAuthorizationCode() 和 exchangeRefreshToken() 会通过服务器之间的调用来与您的 SSO 系统的令牌端点进行交互，从而确保用户名称和密码不会被保存在浏览器中。verifyAccessToken() 会在每次收到来自 MCP 的请求时被调用，它会利用令牌验证接口来确认该令牌仍然有效，并提取用户的权限范围。

按用户权限限制数据访问

这是任何内部数据管理 MCP 服务器中最为重要的部分：人工智能系统只能访问那些被授权的用户才能查看的数据。

千万不要忽视这一点。如果没有根据用户权限来控制数据访问，那么您实际上就是在构建一个带有人工智能功能的数据泄露工具而已。

// src/scoped-tools.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

export function registerScopedTools(
  server: McpServer,
  getUserContext: () => { userId: string; orgId: string; role: string }
) {
  server.tool(
    "search_employees",
    "搜索员工信息。搜索结果会根据用户的权限进行筛选。",
    {
      query: z.string().describe("要搜索的名称、电子邮件或职位"),
    },
    async ({ query }) => {
      const ctx = getUserContext();

      // 确保访问权限得到限制
      let departmentFilter: string | undefined;

      if (ctx.role === "manager") {
        // 经理只能看到自己所在部门的信息
        departmentFilter = await getUserDepartment(ctx.userId);
      } else if (ctx.role === "employee") {
        // 普通员工只能看到有限的信息
        departmentFilter = await getUserDepartment(ctx.userId);
      }
      // 管理员和人力资源人员可以查看所有信息，因此不需要进行任何筛选

      const employees = await searchEmployees(query, departmentFilter);

      // 根据用户的角色来隐藏敏感信息
      const results = employees.map((e) => ({
        name: e.name,
        email: e.email,
        department: e.department,
        role: e.role,
        // 只有管理员和人力资源人员才能看到开始日期和经理信息
        ...(["admin", "hr"].includes(ctx.role)
          ? { start_date: e.start_date, manager_id: e.manager_id }
          : {},
      });

      return {
        content: [
          {
            type: "text",
            text: formatEmployeeList(results),
          },
        ],
      };
    }
  );
}

这里的处理逻辑是：…

  1. 从已认证的会话中提取用户相关信息。

  2. 在数据库层面对查询结果进行过滤处理（而不是在获取所有数据之后再进行过滤）。

  3. 隐藏用户不应看到的字段内容。

  4. 记录访问操作信息，以便后续进行审计。

连接内部API

并非所有内部数据都存储在数据库中。因此，通常需要对外部请求访问这些内部API进行封装处理：

server.tool(
  "get_ticket_details",
  `从内部工单系统中查询相应的支持工单。该接口会返回工单的状态、负责人、优先级以及最近的更新信息。`,
  {
    ticket_id: z
      .string()
      .regex/^TK-\d+$/)
      .describe("工单ID的格式为TK-12345"),
  },
  async ({ ticket_id }) => {
    const ctx = getUserContext();

    const response = await fetch(
      `\({process.env.TICKETING_API_URL}/api/v2/tickets/\){ticket_id}`,
      {
        headers: {
          Authorization: `Bearer ${process.env.TICKETING_SERVICE_TOKEN}`,
          "X-On-Behalf-Of": ctx.userId,
        },
      }
    );

    if (response.status === 404) {
      return {
        content: [
          { type: "text", text: `未找到编号为${ticket_id}的工单。` },
        ],
      };
    }

    if (response.status === 403) {
      return {
        content: [
          {
            type: "text",
            text: `您没有权限访问编号为${ticket_id}的工单。` },
          },
        ],
      };
    }

    const ticket = await response.json();

    return {
      content: [
        {
          type: "text",
          text: `
            **\({ticket.id}: \){ticket.title}**`,
            `状态：\({ticket.status} | 优先级：\){ticket.priority}`,
            `负责人：${ticket.assignee?.name ?? "未分配" },
            `创建时间：${ticket.created_at`),
            "",
            `**最新更新：**`,
            ticket.updates?.[0]?.body ?? "尚未有任何更新。",
          ].join("\n"),
        },
      ],
    };
  }
);

在封装内部API时需要注意以下关键点：

  • 使用服务令牌进行服务器之间的身份验证，但需通过`X-On-Behalf-Of`等头部信息传递用户身份。

  • 明确处理HTTP错误——应返回易于理解的提示信息，而不要直接返回原始的错误对象。

  • 验证输入格式——对`ticket_id`应用正则表达式检查，既能防止注入攻击，也能帮助AI理解预期的数据格式。

  • 避免在错误信息中泄露内部实现细节。

为内部文档构建RAG工具

其中一个极具价值的应用场景就是让AI能够搜索你的内部知识库。下面这个工具可以对内部文档存储进行向量搜索：

server.tool(
  "search_internal_docs",
  `在内部知识库中搜索相关文档。这些文档包括工程文档、操作手册、架构设计方案以及政策规定等。当用户询问有关内部流程、系统或决策的信息时，可以使用这个接口。`,
  {
    query: z
      .string()
      .describe("自然语言搜索查询"),
    category: z
      .enum(["engineering", "policy", "runbook", "architecture", "all"])
      .default("all")
      .describe("要搜索的文档类别"),
    limit: z
      .number()
      .min(1)
      .max(10)
      .default(5)
      .describe("返回的结果数量上限"),
  },
  async ({ query, category, limit }) => {
    // 为搜索查询生成嵌入向量
    const embedding = await generateEmbedding(query);

    // 对内部文档存储进行向量相似度搜索
    const results = await pool.query(
      `SELECT
         d.id,
         d.title,
         d.category,
         d.content_chunk,
         d.source_url,
         d.updated_at,
         1 - (dembedding <=> $1::vector) AS similarity
       FROM document_chunks d
       WHERE (\(2 = 'all' OR d.category = \)2)
         AND 1 - (d.embedding <=> $1::vector) > 0.7
       ORDER BY dembedding <=> $1::vector
       LIMIT $3`,
      [JSON.stringify.embedding), category, limit]
    );

    if (results.rows.length === 0) {
      return {
        content: [
          {
            type: "text",
            text: `未找到与${query}相关的文档。` },
          },
        ],
      };
    }

    const formatted = results.rows
      .map(
        (doc, i) =>
          `### \({i + 1}. \){doc.title}\n` +
          `类别：\({doc.category} | 更新时间：\){doc.updated_at} | 相似度：${(doc.similarity * 100).toFixed(0)}%\n\n` +
          `${doc.content_chunk}\n\n` +
          `来源：${doc.source_url}`
      )
      .join("\n\n---\n\n");

    return {
      content: [
        {
          type: "text",
          text: `找到了\({results.rows.length}份相关文档:\n\n\){formatted}`,
        },
      ],
    };
  }
);

这个工具结合了两种操作：嵌入生成与向量相似性搜索。generateEmbedding(query)会调用一个嵌入模型（例如 OpenAI 的 text-embedding-3-small 或者自定义部署的模型），将用户的查询转换成数值向量。随后，SQL 查询会使用 pgvector 的 <=> 运算符来计算查询向量与存储在数据库中的文档嵌入向量之间的余弦距离——距离越小，相似度越高。1 - (embedding <=> $1) > 0.7 这一条件会过滤掉相似度低于 70% 的结果，从而避免 AI 收到无关紧要的干扰信息。查询结果会按照距离从小到大的顺序排列，并且其数量会受到 limit 参数的限制。输出的结果中还会包含相似度百分比，这样 AI 就可以向用户说明自身的判断准确性。

生产环境部署

将 MCP 服务器容器化

FROM node:22-slim AS builder

WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build

FROM node:22-slim AS runtime

WORKDIR /app
COPY --from=builder /app/dist ./dist
COPY --from=builder /app/node_modules ./node_modules
COPY --from=builder /app/package.json ./

ENV NODE_ENV=production
EXPOSE 3100

HEALTHCHECK --interval=30s --timeout=5s \
  CMD curl -f http://localhost:3100/health || exit 1

CMD ["node", "dist/index.js"]

Dockerfile 采用了两阶段构建方式。在 builder 阶段，系统会安装所有依赖项（包括开发依赖项），并将 TypeScript 代码编译成 JavaScript 代码，这些编译后的文件会被保存到 dist/ 目录中。而在 runtime 阶段，系统会从一个新的 Node.js 镜像开始构建过程，只复制编译后的结果文件以及 node_modules 目录中的依赖项——像 TypeScript 这样的开发依赖项不会被复制进去，这样最终生成的容器镜像的大小就会保持较小。HEALTHCHECK 指令会让 Docker 以及 Kubernetes 等调度系统每隔 30 秒检查一次 /health 端点；如果该端点无法正常响应，容器就会被标记为“不健康状态”，系统会自动重启该容器或将其从负载均衡器的调度列表中移除。

健康检查与监控

请添加一个用于检测依赖项是否正常的健康检查端点：

app.get("/health", async (_req, res) => {
  const checks = {
    database: false,
    ticketingApi: false,
  };

  try {
    await pool.query("SELECT 1");
    checks.database = true;
  } catch {}

  try {
    const resp = await fetch(
      `${process.env.TICKETING_API_URL}/health`
    );
    checks.ticketingApi = resp.ok;
  } catch {}

  const healthy = Object.values(checks).every(Boolean);
  res.status(healthy ? 200 : 503).json({
    status: healthy ? "healthy" : "degraded",
    checks,
    uptime: process.uptime(),
  });
});

/health 端点会同时执行两项依赖项检查：首先会运行一个简单的 SELECT 1 查询来确认数据库连接是否正常；其次会向 ticketing API 发送 HTTP 请求进行检测。这两项检查的结果会被保存到 checks 对象中。如果有任何一项检查失败，该端点就会返回 HTTP 503 状态码（表示“服务不可用”），这样负载均衡器和其他容器调度系统就能及时停止将请求路由到这个不健康的实例上。process.uptime() 这个函数被用来获取系统的运行时间，这样你就可以快速判断出一个实例是刚刚出现故障，还是已经运行了数小时之久。

日志记录与审计追踪

凡是针对内部数据使用的工具调用，都应被记录下来：

function createAuditLogger() {
  return {
    logToolCall(params: {
      userId: string;
      tool: string;
      input: Record;
      resultSize: number;
      durationMs: number;
    }) {
      // 将日志发送到您的日志记录系统
      // （例如 Datadog、ELK、CloudWatch 等）
      console.log(
        JSON.stringify({
          event: "mcp_tool_call",
          timestamp: new Date().toISOString(),
          ...params,
        })
      );
    },
  };
}

createAuditLogger 返回的是一个日志记录对象，而不是类实例。这样一来，就可以轻松地更换底层日志传输机制（如标准输出、日志记录 SDK 等），而无需修改调用代码。audited 这个包装函数属于高阶函数：它接受一个工具处理函数作为参数，然后返回一个新的函数，这个新函数的接口保持不变，但在原始调用前后会添加计时和日志记录功能。try/catch 语句能确保即使处理函数抛出异常，也会生成相应的日志记录——因为审计追踪中需要记录所有失败的调用，而不仅仅是成功的调用。将这些日志发送到中央存储系统（如 Datadog、CloudWatch、ELK），就可以回答诸如“上周二这个用户通过 AI 功能访问了哪些数据？”这样的问题，而对于那些处理敏感内部数据的组织来说，这类查询通常是非常必要的。

请为您的工具处理函数添加逻辑，以便自动记录所有的调用操作：

function audited>&(
  handler: (params: T) => Promise,
  toolName: string,
  audit: ReturnType typeof createAuditLogger>
) {
  return async (params: T): Promise => {
    const start = Date.now();
    const ctx = getUserContext();

    try {
      const result = await handler(params);
      audit.logToolCall({
        userId: ctx.userId,
        tool: toolName,
        input: params,
        resultSize: JSON.stringify(result).length,
        durationMs: Date.now() - start,
      });
      return result;
    } catch (error) {
      audit.logToolCall({
        userId: ctx.userId,
        tool: toolName,
        input: params,
        resultSize: 0,
        durationMs: Date.now() - start,
      });
      throw error;
    }
  };
}

将您的 MCP 服务器连接到 AI 客户端

Claude 桌面应用

请在您的 claude_desktop_config.json 文件中添加以下内容：

{
  "mcpServers": {
    "internal-data": {
      "url": "http://localhost:3100/mcp",
      "headers": {
        "Authorization": "Bearer your-internal-token"
      }
    }
  }
}

自定义应用程序（使用 MCP 客户端 SDK）

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const transport = new StreamableHTTPClientTransport(
  new URL("http://localhost:3100/mcp"),
  {
    requestInit: {
      headers: {
        Authorization: `Bearer ${userToken}`,
      },
    },
  }
);

const client = new Client({
  name: "my-ai-app", version: "1.0.0"
});

await client.connect(transport);

// 查找可用的工具
const { tools } = await client.listTools();
console.log("可用工具:", tools.map((t) => t.name));

// 调用某个工具
const result = await client.callTool({
  name: "search_employees",
  arguments: { query: "engineering manager" },
});

console.log(result.content);

StreamableHTTPClientTransport负责管理与MCP服务器之间的HTTP连接，包括在每个请求中添加Authorization头部信息。client.connect(transport)会执行MCP初始化流程——客户端会告知自身具备的功能，而服务器则会返回可用的工具及资源列表。client.listTools()可以获取完整的工具目录，你可以利用这些信息动态生成用户界面，或者直接将其传递给大语言模型的工具调用API。client.callTool()会发送JSON-RPC请求来按名称调用特定的工具，并返回处理结果所在的content数组——这个格式与AI模型接收的数据格式是完全一致的。在实际应用中，你可以将这个content作为工具执行的结果，添加到对话历史记录中。

常见误区

1. 返回过多数据。大语言模型具有上下文处理能力的上限。如果数据库查询返回了500条记录，就不要全部发送给模型。应该对结果进行分页处理、汇总或限制数量，通常25条记录是一个合理的默认值。

2>工具描述过于笼统。如果你提供了search_employees和search_contractors这样的工具，AI模型需要能够区分它们之间的区别。不能仅依赖工具名称，描述内容才是模型真正会读取的信息。

3>缺乏错误处理机制。当数据库查询失败时，应该返回结构化的错误信息，而不是堆栈跟踪。AI模型需要向用户提供有用的反馈，而原始的错误信息可能会暴露实现细节。

4>没有实施速率限制。AI工具调用可能会在循环中频繁发生。如果某个模型在一次对话中多次调用你的工具，就必须设置相应的限流机制：

const rateLimiter = new Map();

function checkRateLimit(userId: string, limit = 30, windowMs = 60000) {
  const now = Date.now();
  const calls = rateLimiter.get(userId) ?? [];
  const recent = calls.filter((t) => now - t < windowMs);

  if (recent.length >= limit) {
    throw new Error(
      `速率限制已被超过。每分钟最多允许调用${limit}次。`
    );
  }

  recent.push(now);
  rateLimiter.set(userId, recent);
}

5>没有在实际的AI模型上进行测试。你的工具在单元测试中可能看起来没有问题，但在实际使用中可能会让模型产生困惑。因此需要完整地测试整个流程：AI模型接收工具定义、决定调用某个工具、获取结果，并根据结果做出相应的处理。要根据模型的实际行为来调整工具描述内容。

总结

构建用于处理内部数据的中间件服务器，主要涉及以下三个方面：

  1. 合理的工具设计——清晰的描述、类型化的参数、结构化的响应结果

  2. 恰当的访问控制机制——对用户进行身份验证、限制数据访问范围、记录所有操作日志

  3. 具备生产环境适用性——定期进行健康检查、实施速率限制、处理错误、进行监控

该协议本身其实非常简单。真正困难的地方在于如何为你的内部系统设计出恰当的抽象层，以便人工智能能够有效地利用这些抽象层，同时避免数据泄露或导致上下文信息量过大而影响系统的正常运行。

首先选择一两种高价值的工具（如员工信息查询功能、文档搜索系统），让真实用户来试用这些工具，然后再根据使用情况逐步扩展功能。最优秀的内部MCP服务器通常都是根据人们实际提出的需求，自然而然地发展起来的。

本指南中的完整源代码可以在GitHub上获取。

Tags:
Comments are closed.