AI导哈

NestJS MCP模块

NestJS MCP模块允许通过装饰器自动检测、验证和集成OpenAI功能调用，创建用于LLM或HTTP的'工具'、'提示'和'资源'。

  • by Muzikanto
  • TypeScript
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 Muzikanto/nestjs-mcp

使用说明

@穆兹坎托/nestjs-mcp

NestJS MCP（模型上下文协议）模块——允许您为LLM或HTTP创建“工具”、“提示”、“资源”，并通过装饰器自动检测、验证和与OpenAI函数调用集成。

目录

特性

  • 使用注册MCP工具 @McpTool() 装饰器
  • 使用注册MCP提示 @McpPrompt() 装饰器
  • 自动检测模块中的所有提供者（工具）
  • 输入数据验证
  • MCP的SSE端点(GET /mcp/sse, POST /mcp/messages)
  • 调用工具的端点(POST /mcp/tools)
  • 所有工具列表的端点(GET /mcp/tools)
  • 调用提示的端点(POST /mcp/prompts/:name)
  • 所有提示列表的终结点(GET /mcp/prompts)
  • 易于与LLM（OpenAI函数调用）集成
  • 支持http适配器（默认fastify）
  • 完整的TypeScript类型
  • 支持 @UseGuards, @UseInterceptors, @UseFilters

例子

安装

yarn add @muzikanto/nestjs-mcp

对等依赖关系： @nestjs/common, @nestjs/core, @nestjs/swagger, @nestjs/axios, reflect-metadata

用法

连接MCP模块

import { Module } from "@nestjs/common";
import { McpModule } from "@muzikanto/nestjs-mcp";
import { TelegramSendMessageTool } from "./tools/telegram-send-message.tool";
import { TelegramAutoReplyPrompt } from "./prompts/telegram-auto-reply.prompt";

@Module({
  imports: [
    McpModule.forRoot({
      providers: [TelegramSendMessageTool, TelegramAutoReplyPrompt],
      transports: {
        sse: true, // SSE endpoints for mcp
        http: true, // HTTP endpoints for custom server
      },
    }),
  ],
})
export class AppModule {}

创建MCP工具

import { IMcpTool, McpTool } from "@muzikanto/nestjs-mcp";
import { Telegraf } from "telegraf";
import z from "zod/v3";

const schema = {
  chatId: z.number().describe("Telegram chat id"), // строка с описанием
  text: z.string().describe("Message text"), // строка с описанием
};

@McpTool()
export class TelegramSendMessageTool implements IMcpTool<
  { chatId: string; text: string },
  { success: boolean }
> {
  name = "telegram.sendMessage";
  inputSchema = schema;

  constructor(private readonly bot: Telegraf) {}

  async execute(input: { chatId: string; text: string }) {
    await this.bot.telegram.sendMessage(input.chatId, input.text);
    return {
      structuredContent: { success: true },
      messages: [{ type: "text", text: "Success sent to user" }],
    };
  }
}

通过HTTP调用MCP工具

POST/mcp/工具

{
  "type": "telegram.sendMessage",
  "payload": {
    "chatId": "...",
    "text": "Hi, how are you ?"
  }
}

回答

{
  "structuredContent": {
    "success": true
  },
  "messages": [{ "type": "text", "text": "Success sent to user" }]"
}

获取所有工具

GET/mcp/工具

{
  "tools": [
    {
      "name": "telegram.sendMessage",
      "description": "Sent message via Telegram",
      "inputSchema": {
        "type": "object",
        "properties": {
          "chatId": {
            "type": "number",
            "description": "Telegram chat ID"
          },
          "text": {
            "type": "string",
            "description": "Message text"
          }
        },
        "required": ["chatId", "text"]
      }
    }
  ]
}

创建MCP提示

import { IMcpPrompt, McpPrompt } from "@muzikanto/nestjs-mcp";
import z from "zod/v3";

const schema = {
  chatId: z.number().describe("Telegram chat id"), // строка с описанием
  text: z.string().describe("Message text"), // строка с описанием
};

@McpPrompt()
export class TelegramAutoReplyPrompt implements IMcpPrompt<{
  text: string;
  chatId: number;
}> {
  name = "telegram_auto_reply";
  description =
    "Generate a short, friendly reply to an incoming Telegram message and send it back to the same chat using telegram.sendMessage tool";
  schema = schema;

  async execute({ text, chatId }: { text: string; chatId: number }) {
    return {
      messages: [
        {
          role: "system",
          content: `You are a friendly Telegram bot. Reply briefly and to the point.`,
        },
        {
          role: "user",
          content: text,
        },
        {
          role: "assistant",
          tool_call: {
            name: "telegram.sendMessage",
            arguments: {
              chatId,
              text: "{{model_output}}",
            },
          },
        },
      ],
    };
  }
}

获取所有提示

GET/mcp/提示

{
  "prompts": [
    {
      "name": "telegram_auto_reply",
      "description": "Generate a short, friendly reply to an incoming Telegram message and send it back to the same chat using telegram.sendMessage tool",
      "inputSchema": {
        "type": "object",
        "properties": {
          "chatId": {
            "type": "number",
            "description": "Telegram chat ID"
          },
          "text": {
            "type": "string",
            "description": "Message text"
          }
        },
        "required": ["chatId", "text"]
      }
    }
  ]
}

通过HTTP调用MCP提示符

POST/mcp/提示/电报_自动回复

{
  "text": "Hi. How are you ?",
  "chatId": 100000000
}

响应

{
  "messages": [
    {
      "role": "system",
      "content": "You are a friendly Telegram bot. Reply briefly and to the point."
    },
    {
      "role": "user",
      "content": "Hi. How are you ?"
    },
    {
      "role": "assistant",
      "tool_call": {
        "name": "telegram.sendMessage",
        "arguments": {
          "chatId": 100000000,
          "text": "{{model_output}}"
        }
      }
    }
  ]
}

创建MCP资源

import { IMcpResource, McpResource } from "@muzikanto/nestjs-mcp";

@McpResource()
export class TestResource implements IMcpResource<{ userId: string }> {
  name = "users.get";
  uri = "users://{userId}";
  title = "Get test user";
  description = "Get user by id";

  async execute(uri: string, vars: { userId: string }) {
    return [{ uri, text: `Hello ${vars.userId}` }];
  }
}

通过HTTP调用MCP资源

POST/mcp/resources/users.get

{
  "uri": "users://user/1"
}

响应

{
  "contents": [
    {
      "uri": "users://user/1",
      "content": "Hello 1"
    }
  ]
}

获取所有资源

GET/mcp/资源

{
  "resources": [
    {
      "name": "users.get",
      "title": "Get user by id",
      "uri": "users://user/{userId}"
    }
  ]
}

动态创建

import { McpDynamicService } from "@muzikanto/nestjs-mcp";
import { Injectable } from "@nestjs/common";
import { ExampleInterceptor } from "./example.interceptor";
import { ExampleGuard } from "./example.guard";

@Injectable()
export class McpDynamic {
  constructor(protected readonly mcpDynamicService: McpDynamicService) {}

  onModuleInit() {
    this.mcpDynamicService.registerTool({
      name: "dynamic_tool",
      title: "Dynamic tool",
      execute: () =>
        Promise.resolve({
          structuredContent: { text: "test" },
          messages: [{ type: "text" as const, text: "test" }],
        }),
      guards: [ExampleGuard],
      interceptors: [ExampleInterceptor],
    });

    this.mcpDynamicService.registerPrompt({
      name: "dynamic_prompt",
      title: "Dynamic prompt",
      execute: () =>
        Promise.resolve({ messages: [{ role: "assistant", content: "test" }] }),
      guards: [ExampleGuard],
      interceptors: [ExampleInterceptor],
    });

    this.mcpDynamicService.registerResource<{ testId: string }>({
      name: "dynamic_resource",
      title: "Dynamic resource",
      uri: "dynamic://test/{testId}",
      execute: (uri, input) =>
        Promise.resolve([{ uri: uri.href, text: `ID: ${input.testId}` }]),
      guards: [ExampleGuard],
      interceptors: [ExampleInterceptor],
      list: async () => {
        return [
          { uri: "dynamic://test/1", name: "dynamice_1", title: "Dynamic 1" },
        ];
      },
    });
  }
}

守卫

首先

import { IMcpTool, McpTool } from "@muzikanto/nestjs-mcp";
import { UseGuards } from "@nestjs/common";
import { CanActivate, ExecutionContext, Injectable } from "@nestjs/common";
import { Observable } from "rxjs";

@Injectable()
class TestGuard implements CanActivate {
  canActivate(
    context: ExecutionContext,
  ): boolean | Promise<boolean> | Observable<boolean> {
    console.log("TestGuard called");
    return true;
  }
}

@UseGuards(TestGuard)
@McpTool()
export class TelegramSendMessageTool implements IMcpTool<
  { chatId: string; text: string },
  { success: boolean }
> {
  name = "telegram.sendMessage";

  async execute() {
    // await this.bot.telegram.sendMessage(input.chatId, input.text);
    return {
      structuredContent: { success: true },
      messages: { type: "text", text: "Success sent" },
    };
  }
}

对于所有人

import { McpModule } from "@muzikanto/nestjs-mcp";
import { Module } from "@nestjs/common";
import { TestGuard } from "./other/test.guard";
import { ToolWithoutGuards } from "./test.tool";
import { APP_GUARD } from "@nestjs/core";

@Module({
  imports: [
    McpModule.forRoot({
      providers: [
        ToolWithoutGuards,
        TestGuard,
        { provide: APP_GUARD, useExisting: TestGuard },
      ],
    }),
  ],
})
export class TestModule {}

拦截器

首先

import { IMcpTool, McpTool } from "@muzikanto/nestjs-mcp";
import {
  ExecutionContext,
  NestInterceptor,
  CallHandler,
  Injectable,
  UseInterceptors,
} from "@nestjs/common";
import { map } from "rxjs";

@Injectable()
export class ExampleInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler) {
    console.log("Before execute");

    return next.handle().pipe(
      map((data: unknown) => {
        console.log("After execute", data);

        return data;
      }),
    );
  }
}

@UseInterceptors(ExampleInterceptor)
@McpTool()
export class TelegramSendMessageTool implements IMcpTool<
  { chatId: string; text: string },
  { success: boolean }
> {
  name = "telegram.sendMessage";

  async execute() {
    // await this.bot.telegram.sendMessage(input.chatId, input.text);
    return {
      success: true,
      messagse: [{ type: "text", text: "Success sent" }],
    };
  }
}

对于所有人

import { McpModule } from "@muzikanto/nestjs-mcp";
import { Module } from "@nestjs/common";
import { TestInterceptor } from "./other/test.interceptor";
import { ToolWithoutInterceptors } from "./test.tool";
import { APP_INTERCEPTOR } from "@nestjs/core";

@Module({
  imports: [
    McpModule.forRoot({
      providers: [
        ToolWithoutInterceptors,
        TestInterceptor,
        { provide: APP_INTERCEPTOR, useExisting: TestInterceptor },
      ],
    }),
  ],
})
export class TestModule {}

过滤器

import {
  IMcpTool,
  McpTool,
  McpUnauthorizedException,
} from "@muzikanto/nestjs-mcp";
import {
  ExceptionFilter,
  Catch,
  ArgumentsHost,
  UseFilters,
} from "@nestjs/common";

@Catch(NotImplementedException)
class NotImplExceptionFilter implements ExceptionFilter {
  catch(exception: unknown, host: ArgumentsHost) {
    return {
      isError: true,
      messages: [{ type: "text", text: "Not implemented tool" }],
    };
  }
}

@Catch(McpUnauthorizedException)
class AuthExceptionFilter implements ExceptionFilter {
  catch(exception: unknown, host: ArgumentsHost) {
    return {
      isError: true,
      messages: [{ type: "text", text: "No access" }],
    };
  }
}

@UseGuards(AuthGuard)
@UseFilters(AuthExceptionFilter, NotImplExceptionFilter)
@McpTool()
export class TelegramSendMessageTool implements IMcpTool<
  { chatId: string; text: string },
  { success: boolean }
> {
  name = "telegram.sendMessage";

  async execute() {
    throw new NotImplementedException();
  }
}

与OpenAI函数调用集成

示例 openai: 6.25.0

import axios from "axios";
import OpenAI from "openai";

const MCP_TOOLS_URL = "http://localhost:3000/mcp/tools";
const MCP_TELEGRAM_PROMPT_URL =
  "http://localhost:3000/mcp/prompts/telegram_auto_reply";

// Create OpenAI client
const client = new OpenAI({
  baseUrl: "https://openrouter.ai/api/v1",
  apiKey: process.env.OPENROUTER_API_KEY,
});

/**
 * Get all tools
 */
async function getMcpTools() {
  const response = await axios.get(MCP_TOOLS_URL);

  return response.data.tools.map((el) => ({
    type: "function",
    function: {
      name: el.name,
      description: el.description,
      parameters: el.inputSchema,
    },
  }));
}

/**
 * Get all prompts
 */
async function getMcpPrompt() {
  const response = await axios.post(MCP_TELEGRAM_PROMPT_URL, {
    /* prompt generation arguments */
  });
  return response.data;
}

/**
 * Request mcp tool
 */
async function callMcpTool(toolName: string, payload: Record<string, any>) {
  const response = await axios.post(MCP_TOOLS_URL, { type: toolName, payload });
  return response.data;
}

/**
 * Example
 */
(async () => {
  const toolsResponse = await getMcpTools();
  const promptResponse = await getMcpPrompt();

  const completion = await client.chat.completions.create({
    model: "arcee-ai/trinity-large-preview:free",
    messages: promptResponse.messages,
    tools: toolsResponse,
    tool_choice: "auto",
  });

  const message = completion.choices[0].message;

  if (message.tool_calls?.length) {
    const { name, arguments: argsJson } = message.tool_calls[0].function;
    const args = JSON.parse(argsJson);

    const { messages, structuredContent } = await callMcpTool(name, args);
    console.log("Result from MCP tool:", messages, structuredContent);
  }
})();

贡献

欢迎投稿！请打开问题或提交PR。

更新日志

看 更新日志 查看详细的版本历史和更新。

相关 MCP