MCP服务器Dart

<picture> <source media="(prefers-color-scheme: dark)" srcset="https://raw.githubusercontent.com/Zfinix/mcp_server_dart/main/logo-dark.svg"> <source media="(prefers-color-scheme: light)" srcset="https://raw.githubusercontent.com/Zfinix/mcp_server_dart/main/logo-light.svg"> <img alt="MCP Dart Framework" src="https://raw.githubusercontent.com/Zfinix/mcp_server_dart/main/image.png" width="400"> </picture> </br> </br>

一个面向开发者的Dart MCP框架，支持注解和代码生成。通过在方法上添加@MCPTool、@MCPResource或@MCPPrompt注解来构建MCP服务器，类似于json_serializable或freezed的工作方式。

✨ 特性

• 🏷️ 注解驱动：使用简单的注解声明MCP工具、资源和提示
• 🔧 代码生成：使用build_runner扩展注册自动生成样板代码
• ✨ 不需要@override：无需继承样板代码的干净方法声明
• 📡 多种传输方式：支持标准I/O、HTTP和带有服务端发送事件（SSE）的流式HTTP
• 🔍 类型安全：从方法签名中自动提取参数的完整Dart类型安全
• 📚 JSON Schema：从方法签名自动生成输入模式
• 🧪 完整测试：具有JSON-RPC命令验证的全面测试套件
• ⚡ 生产就绪：完整的MCP 2025-06-18协议实现，使用Relic HTTP服务器
• 🌐 现代HTTP：基于Relic框架构建，支持中间件、CORS、日志记录和健康检查
• 🔧 监控：内置健康检查端点和连接监控
• 📊 SSE支持：根据MCP 2025-06-18规范支持实时流式传输

🚀 快速开始

1. 添加依赖项

dependencies:
  mcp_server_dart: ^1.1.2
  relic: ^0.5.0  # 现代HTTP框架
  logging: ^1.3.0  # 用于服务器日志记录

dev_dependencies:
  build_runner: ^2.4.13

2. 创建您的MCP服务器

import 'package:mcp_server_dart/mcp_server_dart.dart';

part 'my_server.mcp.dart'; // 生成的文件

class MyMCPServer extends MCPServer {
  MyMCPServer() : super(name: 'my-server', version: '1.0.0') {
    // 使用扩展注册所有生成的处理器
    registerGeneratedHandlers();
  }

  @MCPTool('greet', description: '通过名字问候某人')
  Future<String> greet(String name) async {
    return 'Hello, $name! 👋';
  }

  @MCPTool('calculate', description: '执行基本算术运算')
  Future<double> calculate(double a, double b, String operation) async {
    switch (operation) {
      case 'add': return a + b;
      case 'subtract': return a - b;
      case 'multiply': return a * b;
      case 'divide': return b != 0 ? a / b : throw ArgumentError('除以零');
      default: throw ArgumentError('未知操作：$operation');
    }
  }

  @MCPResource('status', description: '服务器状态信息')
  Future<Map<String, dynamic>> getStatus() async {
    return {
      'server': name,
      'version': version,
      'uptime': DateTime.now().toIso8601String(),
      'status': 'healthy',
    };
  }

  @MCPPrompt('codeReview', description: '生成代码审查提示')
  String codeReviewPrompt(String code, String language) {
    return '''请审核这段 $language 代码：
- 最佳实践和约定
- 潜在的错误或问题
- 性能改进
- 安全考虑

代码：
```$language
$code
```''';
  }
}

3. 生成代码

dart run build_runner build

这会生成my_server.mcp.dart，提供自动注册方法的扩展。

4. 运行您的服务器

import 'dart:io';
import 'package:logging/logging.dart';

void main() async {
  // 启用日志记录以查看服务器活动
  Logger.root.level = Level.INFO;
  Logger.root.onRecord.listen((record) {
    print('${record.level.name}: ${record.time}: ${record.message}');
  });

  final server = MyMCPServer(); // 构造函数中自动注册处理器
  
  // 选择您的传输方式：
  await server.start();        // 用于CLI集成（标准I/O）
  // 或者
  await server.serve(port: 8080);   // 带有健康检查的HTTP服务器
}

服务器特性：

• 🌐 HTTP服务器：使用Relic框架在指定端口运行
• 🔍 健康检查：可在http://localhost:8080/health访问
• 📊 状态端点：服务器指标在http://localhost:8080/status
• 📡 MCP端点：流式HTTP在http://localhost:8080/mcp
• 📝 请求日志：所有HTTP请求带时间戳的日志记录
• 🛡️ CORS支持：默认启用跨域请求
• ⚡ 优雅关闭：正确处理SIGINT/SIGTERM信号
• 🔄 SSE流式传输：用于实时通信的服务端发送事件

📖 注解参考

@MCPTool

标记一个方法作为LLMs可以调用的MCP工具：

@MCPTool('toolName', description: '此工具的作用')
Future<ReturnType> myTool(ParameterType param) async {
  // 实现
}

特性：

• 自动参数提取和类型检查
• 从方法签名生成JSON Schema
• 支持带有默认值的可选参数
• 支持异步和同步方法

@MCPResource

标记一个方法作为提供数据的MCP资源：

@MCPResource('resourceName', 
  description: '此资源包含的内容',
  mimeType: 'application/json'  // 可选
)
Future<Map<String, dynamic>> getResource() async {
  // 返回资源数据
}

@MCPPrompt

标记一个方法作为MCP提示模板：

@MCPPrompt('promptName', description: '此提示的作用')
String generatePrompt(String context, String task) {
  return '基于 $context 和 $task 生成的提示';
}

@MCPParam

为参数提供额外元数据：

@MCPTool('example')
Future<String> example(
  @MCPParam(description: '用户名', example: 'John Doe') 
  String name,
  
  @MCPParam(required: false, description: '年龄（年）')
  int age =  25,
) async {
  return 'Hello $name, 年龄 $age';
}

🌟 完整示例

参见Google Maps MCP示例进行综合演示：

class GoogleMapsMCP extends MCPServer {
  GoogleMapsMCP() : super(name: 'google-maps-mcp', version: '1.0.0') {
    registerGeneratedHandlers();
  }

  @MCPTool('searchPlace', description: '按名称或地址查找地点')
  Future<Map<String, dynamic>> searchPlace(String query, int limit = 5) async {
    // 使用模拟的Google Maps API调用实现
  }

  @MCPTool('getDirections', description: '获取两点之间的路线')
  Future<Map<String, dynamic>> getDirections(
    String origin, 
    String destination, 
    String mode = 'driving'
  ) async {
    // 实现
  }

  @MCPResource('currentLocation', description: '当前用户位置')
  Future<Map<String, dynamic>> getCurrentLocation() async {
    // 实现
  }

  @MCPPrompt('locationSummary', description: '生成位置摘要')
  String locationSummaryPrompt(String location, String summaryType = 'general') {
    // 生成上下文提示
  }
}

🔧 高级用法

自定义参数验证

@MCPTool('validateEmail')
Future<bool> validateEmail(String email) async {
  if (!email.contains('@')) {
    throw ArgumentError('无效的电子邮件格式');
  }
  // 验证逻辑
}

复杂输入模式

@MCPTool('complexTool', inputSchema: {
  'type': 'object',
  'properties': {
    'config': {
      'type': 'object',
      'properties': {
        'timeout': {'type': 'integer', 'minimum': 1},
        'retries': {'type': 'integer', 'maximum': 10}
      }
    }
  }
})
Future<String> complexTool(Map<String, dynamic> config) async {
  // 处理复杂的嵌套参数
}

多种传输支持

服务器支持标准I/O和HTTP传输。您可以扩展上面的基本示例以处理命令行参数：

// 在您的main()函数中添加参数处理：
if (args.contains('--stdio')) {
  print('🔌 正在标准I/O上启动MCP服务器...');
  await server.start();
} else {
  final port = args.contains('--port') 
      ? int.parse(args[args.indexOf('--port') + 1])
      : 8080;
  
  print('🌐 正在端口 $port 上启动HTTP服务器...');
  print('🔍 健康检查：http://localhost:$port/health');
  print('📊 状态：http://localhost:$port/status');
  print('📡 MCP端点：http://localhost:$port/mcp');
  
  await server.serve(port: port);
}

传输选项：

• Stdio：适用于Claude Desktop集成和CLI工具
• HTTP：适用于Web应用程序、测试和调试，使用Relic框架
• 流式HTTP：最新的MCP 2025-06-18规范，支持服务端发送事件
• 健康监测：生产监测内置端点

🌐 流式HTTP传输（MCP 2025-06-18）

MCP Dart框架现在支持最新的流式HTTP传输规范：

关键特性

• 单一MCP端点：POST/GET /mcp处理所有MCP通信
• 服务端发送事件：实时流式传输服务器发起的消息
• 会话管理：自动会话ID生成和验证
• 协议头：支持MCP-Protocol-Version: 2025-06-18
• 安全性：开发时的源验证和本地绑定

使用MCP Inspector测试

# 启动您的服务器
dart run main.dart --example calculator --http --port 8080

# 使用Inspector CLI测试
npx @modelcontextprotocol/inspector --cli http://localhost:8080/mcp --transport streamable-http --method tools/list

# 使用Inspector UI测试
npx @modelcontextprotocol/inspector
# 然后连接到：http://localhost:8080/mcp，使用“流式HTTP”传输

使用cURL测试

# 初始化连接
curl -X POST http://localhost:8080/mcp \
  -H 'Accept: application/json' \
  -H 'Content-Type: application/json' \
  -H 'MCP-Protocol-Version: 2025-06-18' \
  --data '{"jsonrpc":"2.0","id":"1","method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{}}}'

# 列出工具
curl -X POST http://localhost:8080/mcp \
  -H 'Accept: application/json' \
  -H 'Content-Type: application/json' \
  -H 'MCP-Protocol-Version: 2025-06-18' \
  --data '{"jsonrpc":"2.0","id":"2","method":"tools/list"}'

# 打开SSE流
curl -H 'Accept: text/event-stream' http://localhost:8080/mcp

🌐 Relic框架集成

MCP Dart框架使用Relic作为其HTTP服务器基础。Relic是一个现代、类型安全的Web服务器框架，灵感来自Shelf但有了显著改进：

为什么是Relic？

• 🔒 类型安全：不再有dynamic类型——一切都是强类型的
• ⚡ 性能：使用Uint8List而不是List<int>以获得更好的性能
• 🛣️ 高效路由：基于trie的高效路由和参数提取
• 🔧 现代头部：类型化的头部解析和验证
• 🧪 全面测试：扩展的测试覆盖和生产就绪

由Relic提供的服务器特性

// 您的MCP服务器自动获得这些特性：
await server.serve(
  port: 8080,
  address: InternetAddress.anyIPv4,
  enableCors: true,  // CORS中间件
  keepAliveTimeout: Duration(seconds: 30),
);

内置端点：

• GET /health - 包含服务器指标的健康检查
• GET /status - 详细的服务器状态和能力
• POST/GET /mcp - MCP流式HTTP端点
• GET /ws - 即将推出的WebSocket升级端点

中间件堆栈：

• 🌐 CORS中间件：跨域请求支持
• 📝 日志中间件：带时间戳的请求/响应日志记录
• 🛡️ 错误处理：优雅的错误处理和响应
• 🛣️ 路由：自动路由注册和参数提取

有关Relic功能的更多细节，请参阅官方Relic文档。

生产部署的二进制编译

将您的MCP服务器编译成本机二进制文件以获得最佳性能和轻松部署：

# 编译为独立二进制文件
dart compile exe my_server.dart -o mcp-server

# 跨平台编译
dart compile exe my_server.dart -o mcp-server-linux --target-os=linux
dart compile exe my_server.dart -o mcp-server-macos --target-os=macos
dart compile exe my_server.dart -o mcp-server.exe --target-os=windows

二进制部署的好处：

• ✅ 不需要Dart运行时 - 独立可执行文件
• ✅ 更快的启动 - 没有VM开销
• ✅ 轻松分发 - 单文件部署
• ✅ 生产就绪 - 优化了性能

使用二进制配置Claude Desktop：

{
  "mcpServers": {
    "my-server": {
      "command": "/path/to/mcp-server"
    }
  }
}

🧪 测试

该框架包括单元测试和集成测试的全面测试能力。以下是几种方法：

测试HTTP服务器

与Relic集成，您可以轻松测试您的MCP服务器的HTTP端点：

import 'dart:convert';
import 'dart:io';
import 'package:test/test.dart';

void main() {
  group('MCP Server HTTP Tests', () {
    late MyMCPServer server;
    late HttpClient client;
    
    setUpAll(() async {
      server = MyMCPServer(); // 处理器自动注册
      await server.serve(port: 8081); // 使用不同的端口进行测试
      
      client = HttpClient();
    });
    
    tearDownAll(() async {
      await server.shutdown();
      client.close();
    });

    test('健康检查端点工作正常', () async {
      final request = await client.get('localhost', 8081, '/health');
      final response = await request.close();
      
      expect(response.statusCode, equals(200));
      
      final body = await response.transform(utf8.decoder).join();
      final data = jsonDecode(body);
      expect(data['status'], equals('healthy'));
      expect(data['server'], equals('my-server'));
    });

    test('MCP端点使用流式HTTP工作正常', () async {
      final request = await client.post('localhost', 8081, '/mcp');
      request.headers.set('content-type', 'application/json');
      request.headers.set('mcp-protocol-version', '2025-06-18');
      request.write(jsonEncode({
        'jsonrpc': '2.0',
        'id': '1',
        'method': 'tools/list'
      }));
      
      final response = await request.close();
      expect(response.statusCode, equals(200));
      
      final body = await response.transform(utf8.decoder).join();
      final data = jsonDecode(body);
      expect(data['result']['tools'], isA<List>());
    });
  });
}

单元测试单个方法

import 'package:test/test.dart';
import 'my_server.dart';

void main() {
  group('MyMCPServer', () {
    late MyMCPServer server;
    
    setUp(()