Ruby SDK

Model Context Protocol (MCP) 的 Ruby SDK 提供了构建 MCP 服务器和客户端的优雅 Ruby 解决方案，支持现代 Ruby 特性和惯用法。

安装

Gem 安装

gem install model_context_protocol

Gemfile

gem 'model_context_protocol', '~> 1.0'

# 可选依赖
gem 'async', '~> 2.0'  # 异步支持
gem 'websocket-client-simple', '~> 0.8'  # WebSocket 支持

Bundler

bundle install

快速开始

创建 MCP 服务器

require 'model_context_protocol'

# 创建服务器
server = ModelContextProtocol::Server.new(
  name: 'my-ruby-server',
  version: '1.0.0',
  description: 'Ruby MCP 服务器示例'
)

# 定义工具
class EchoTool < ModelContextProtocol::Tool
  def name
    'echo'
  end
  
  def description
    '回显输入的消息'
  end
  
  def input_schema
    {
      type: 'object',
      properties: {
        message: {
          type: 'string',
          description: '要回显的消息'
        }
      },
      required: ['message']
    }
  end
  
  def call(arguments)
    message = arguments['message']
    return error('缺少 message 参数') unless message
    
    success(echo: message)
  end
end

# 注册工具
server.register_tool(EchoTool.new)

# 启动服务器
server.start

创建 MCP 客户端

require 'model_context_protocol'

# 创建传输层
transport = ModelContextProtocol::StdioTransport.new

# 创建客户端
client = ModelContextProtocol::Client.new(
  transport: transport,
  connect_timeout: 10,
  request_timeout: 30
)

begin
  # 连接到服务器
  client.connect
  
  # 列出可用工具
  tools = client.list_tools
  puts "可用工具: #{tools.map(&:name).join(', ')}"
  
  # 调用工具
  result = client.call_tool('echo', message: 'Hello from Ruby!')
  puts "工具调用结果: #{result}"
ensure
  client.close
end

核心功能

服务器功能

工具注册

# 计算器工具
class CalculatorTool < ModelContextProtocol::Tool
  def name
    'calculator'
  end
  
  def description
    '执行基本数学运算'
  end
  
  def input_schema
    {
      type: 'object',
      properties: {
        operation: {
          type: 'string',
          enum: %w[add subtract multiply divide],
          description: '运算类型'
        },
        a: {
          type: 'number',
          description: '第一个数字'
        },
        b: {
          type: 'number',
          description: '第二个数字'
        }
      },
      required: %w[operation a b]
    }
  end
  
  def call(arguments)
    operation = arguments['operation']
    a = arguments['a']
    b = arguments['b']
    
    return error('无效的参数') unless operation && a && b
    
    result = case operation
             when 'add'
               a + b
             when 'subtract'
               a - b
             when 'multiply'
               a * b
             when 'divide'
               return error('除数不能为零') if b.zero?
               a / b
             else
               return error("不支持的运算: #{operation}")
             end
    
    success(result: result)
  end
end

# 注册工具
server.register_tool(CalculatorTool.new)

# 或者使用块语法
server.register_tool('simple_tool') do |arguments|
  # 工具逻辑
  success(message: 'Hello from block tool!')
end

资源管理

class FileResourceProvider < ModelContextProtocol::ResourceProvider
  def initialize(base_path)
    @base_path = base_path
  end
  
  def get_resource(uri)
    file_path = File.join(@base_path, uri.path[1..-1])
    
    unless File.exist?(file_path)
      raise ModelContextProtocol::ResourceNotFoundError, "资源不存在: #{uri}"
    end
    
    content = File.read(file_path)
    
    ModelContextProtocol::Resource.new(
      uri: uri,
      mime_type: 'text/plain',
      content: content
    )
  end
  
  def list_resources
    Dir.glob(File.join(@base_path, '**', '*'))
       .select { |path| File.file?(path) }
       .map do |path|
         relative_path = path.sub(@base_path, '')
         ModelContextProtocol::ResourceURI.new("file://#{relative_path}")
       end
  end
end

# 注册资源提供者
server.register_resource_provider(FileResourceProvider.new('/path/to/files'))

提示模板

class CodeReviewPromptProvider < ModelContextProtocol::PromptProvider
  def get_prompt(name, arguments)
    case name
    when 'code-review'
      create_code_review_prompt(arguments)
    else
      raise ModelContextProtocol::PromptNotFoundError, "未知提示: #{name}"
    end
  end
  
  def list_prompts
    ['code-review']
  end
  
  private
  
  def create_code_review_prompt(arguments)
    code = arguments['code']
    raise ArgumentError, '缺少 code 参数' unless code
    
    language = arguments['language'] || 'unknown'
    
    content = <<~PROMPT
      请审查以下 #{language} 代码并提供改进建议:
      
      ```#{language}
      #{code}
      ```
    PROMPT
    
    ModelContextProtocol::Prompt.new(
      name: 'code-review',
      description: '代码审查提示',
      content: content
    )
  end
end

# 注册提示提供者
server.register_prompt_provider(CodeReviewPromptProvider.new)

客户端功能

连接管理

# 自动重连客户端
client = ModelContextProtocol::Client.new(
  transport: transport,
  auto_reconnect: true,
  max_reconnect_attempts: 5,
  reconnect_delay: 2
)

# 连接状态回调
client.on_connected do
  puts '已连接到服务器'
end

client.on_disconnected do
  puts '与服务器断开连接'
end

client.on_error do |error|
  puts "连接错误: #{error.message}"
end

并发操作

require 'async'

# 并发调用多个工具
Async do
  tasks = [
    Async { client.call_tool('tool1', args1) },
    Async { client.call_tool('tool2', args2) },
    Async { client.call_tool('tool3', args3) }
  ]
  
  results = tasks.map(&:wait)
  
  results.each_with_index do |result, index|
    puts "结果 #{index + 1}: #{result}"
  end
end

流式操作

# 流式工具调用
client.call_tool_stream('streaming-tool', args) do |chunk|
  print chunk
end

传输协议

Stdio 传输

transport = ModelContextProtocol::StdioTransport.new(
  command: ['python', 'server.py'],
  working_directory: '/path/to/server',
  environment: { 'ENV_VAR' => 'value' }
)

WebSocket 传输

transport = ModelContextProtocol::WebSocketTransport.new(
  uri: 'ws://localhost:8080/mcp',
  headers: { 'Authorization' => 'Bearer token' },
  connect_timeout: 10
)

HTTP SSE 传输

transport = ModelContextProtocol::SSETransport.new(
  uri: 'http://localhost:8080/mcp',
  headers: { 'Authorization' => 'Bearer token' },
  read_timeout: 30
)

高级特性

依赖注入支持

require 'dry-container'
require 'dry-auto_inject'

# 创建容器
class Container
  extend Dry::Container::Mixin
  
  # 注册服务
  register(:database_service) { DatabaseService.new }
  register(:email_service) { EmailService.new }
  
  # 注册工具
  register(:database_tool) { DatabaseTool.new }
  register(:email_tool) { EmailTool.new }
end

# 自动注入
Import = Dry::AutoInject(Container)

# 带依赖注入的工具
class DatabaseTool < ModelContextProtocol::Tool
  include Import[:database_service]
  
  def name
    'database-query'
  end
  
  def description
    '执行数据库查询'
  end
  
  def input_schema
    {
      type: 'object',
      properties: {
        query: {
          type: 'string',
          description: 'SQL 查询语句'
        }
      },
      required: ['query']
    }
  end
  
  def call(arguments)
    query = arguments['query']
    return error('缺少 query 参数') unless query
    
    begin
      result = database_service.execute_query(query)
      success(result)
    rescue => e
      error("查询执行失败: #{e.message}")
    end
  end
end

# 创建服务器并注册工具
server = ModelContextProtocol::Server.new(name: 'my-server', version: '1.0.0')
server.register_tool(Container[:database_tool])
server.register_tool(Container[:email_tool])

中间件支持

class LoggingMiddleware < ModelContextProtocol::Middleware
  def call(request, &next_middleware)
    puts "处理请求: #{request.method}"
    start_time = Time.now
    
    begin
      response = next_middleware.call(request)
      duration = ((Time.now - start_time) * 1000).round
      puts "请求完成，耗时: #{duration}ms"
      response
    rescue => e
      duration = ((Time.now - start_time) * 1000).round
      puts "请求处理失败，耗时: #{duration}ms"
      raise e
    end
  end
end

class AuthenticationMiddleware < ModelContextProtocol::Middleware
  def call(request, &next_middleware)
    # 验证逻辑
    auth_header = request.headers['Authorization']
    unless valid_token?(auth_header)
      raise ModelContextProtocol::UnauthorizedError, '无效的认证令牌'
    end
    
    next_middleware.call(request)
  end
  
  private
  
  def valid_token?(token)
    # 实现令牌验证逻辑
    token&.start_with?('Bearer ')
  end
end

# 注册中间件
server.use(LoggingMiddleware.new)
server.use(AuthenticationMiddleware.new)

配置管理

require 'yaml'

# 配置类
class McpConfig
  attr_reader :server, :logging, :transport
  
  def initialize(config_hash)
    @server = ServerConfig.new(config_hash['server'] || {})
    @logging = LoggingConfig.new(config_hash['logging'] || {})
    @transport = TransportConfig.new(config_hash['transport'] || {})
  end
  
  def self.load_from_file(file_path)
    config_hash = YAML.load_file(file_path)
    new(config_hash)
  end
  
  class ServerConfig
    attr_reader :name, :version, :description, :max_connections, :request_timeout
    
    def initialize(config)
      @name = config['name'] || ''
      @version = config['version'] || ''
      @description = config['description'] || ''
      @max_connections = config['max_connections'] || 100
      @request_timeout = config['request_timeout'] || 30
    end
  end
  
  class LoggingConfig
    attr_reader :level, :file
    
    def initialize(config)
      @level = config['level'] || 'INFO'
      @file = config['file']
    end
  end
  
  class TransportConfig
    attr_reader :type, :buffer_size
    
    def initialize(config)
      @type = config['type'] || 'stdio'
      @buffer_size = config['buffer_size'] || 8192
    end
  end
end

# config.yml
# server:
#   name: my-server
#   version: 1.0.0
#   description: 我的 MCP 服务器
#   max_connections: 100
#   request_timeout: 30
# logging:
#   level: INFO
#   file: /var/log/mcp-server.log
# transport:
#   type: stdio
#   buffer_size: 8192

# 加载配置
config = McpConfig.load_from_file('config.yml')

# 使用配置创建服务器
server = ModelContextProtocol::Server.new(
  name: config.server.name,
  version: config.server.version,
  description: config.server.description,
  max_connections: config.server.max_connections,
  request_timeout: config.server.request_timeout
)

事件系统

class EventEmitter
  def initialize
    @listeners = Hash.new { |h, k| h[k] = [] }
  end
  
  def on(event, &block)
    @listeners[event] << block
  end
  
  def emit(event, *args)
    @listeners[event].each { |listener| listener.call(*args) }
  end
end

class McpServerWithEvents < ModelContextProtocol::Server
  include EventEmitter
  
  def register_tool(tool)
    super
    emit(:tool_registered, tool)
  end
  
  def handle_request(request)
    emit(:request_received, request)
    response = super
    emit(:response_sent, response)
    response
  end
end

# 使用事件
server = McpServerWithEvents.new(name: 'event-server', version: '1.0.0')

server.on(:tool_registered) do |tool|
  puts "工具已注册: #{tool.name}"
end

server.on(:request_received) do |request|
  puts "收到请求: #{request.method}"
end

server.on(:response_sent) do |response|
  puts "发送响应: #{response.status}"
end

测试

单元测试 (RSpec)

require 'rspec'
require 'model_context_protocol'

RSpec.describe EchoTool do
  let(:tool) { EchoTool.new }
  
  describe '#call' do
    context 'with valid message' do
      it 'returns echo response' do
        result = tool.call('message' => 'test')
        
        expect(result).to be_success
        expect(result.data['echo']).to eq('test')
      end
    end
    
    context 'with missing message' do
      it 'returns error' do
        result = tool.call({})
        
        expect(result).to be_error
        expect(result.error_message).to include('缺少 message 参数')
      end
    end
  end
end

RSpec.describe CalculatorTool do
  let(:tool) { CalculatorTool.new }
  
  describe '#call' do
    it 'performs addition correctly' do
      result = tool.call(
        'operation' => 'add',
        'a' => 2,
        'b' => 3
      )
      
      expect(result).to be_success
      expect(result.data['result']).to eq(5)
    end
    
    it 'handles division by zero' do
      result = tool.call(
        'operation' => 'divide',
        'a' => 10,
        'b' => 0
      )
      
      expect(result).to be_error
      expect(result.error_message).to include('除数不能为零')
    end
  end
end

集成测试

require 'rspec'
require 'model_context_protocol'

RSpec.describe 'MCP Integration' do
  let(:server) do
    ModelContextProtocol::Server.new(name: 'test-server', version: '1.0.0')
  end
  
  let(:client) do
    transport = ModelContextProtocol::StdioTransport.new
    ModelContextProtocol::Client.new(transport: transport)
  end
  
  before do
    server.register_tool(EchoTool.new)
    
    # 启动服务器（在后台线程中）
    @server_thread = Thread.new { server.start }
    sleep 0.1  # 等待服务器启动
    
    client.connect
  end
  
  after do
    client.close
    @server_thread.kill
  end
  
  it 'calls tool successfully' do
    result = client.call_tool('echo', message: 'test')
    
    expect(result['echo']).to eq('test')
  end
  
  it 'lists available tools' do
    tools = client.list_tools
    
    expect(tools.map(&:name)).to include('echo')
  end
end

性能测试

require 'benchmark'

# 性能基准测试
def benchmark_tool_calls(client, iterations = 1000)
  Benchmark.bm(15) do |x|
    x.report('tool_calls:') do
      iterations.times do
        client.call_tool('echo', message: 'benchmark')
      end
    end
  end
end

# 并发性能测试
def benchmark_concurrent_calls(client, threads = 10, calls_per_thread = 100)
  start_time = Time.now
  
  threads_array = Array.new(threads) do
    Thread.new do
      calls_per_thread.times do
        client.call_tool('echo', message: 'concurrent')
      end
    end
  end
  
  threads_array.each(&:join)
  
  end_time = Time.now
  total_calls = threads * calls_per_thread
  duration = end_time - start_time
  
  puts "总调用次数: #{total_calls}"
  puts "总耗时: #{duration.round(2)}秒"
  puts "平均 QPS: #{(total_calls / duration).round(2)}"
end

部署

Rack 应用集成

require 'rack'
require 'model_context_protocol'

class McpRackApp
  def initialize
    @server = ModelContextProtocol::Server.new(
      name: 'rack-mcp-server',
      version: '1.0.0'
    )
    @server.register_tool(EchoTool.new)
  end
  
  def call(env)
    request = Rack::Request.new(env)
    
    if request.path == '/mcp' && request.post?
      handle_mcp_request(request)
    else
      [404, {}, ['Not Found']]
    end
  end
  
  private
  
  def handle_mcp_request(request)
    begin
      body = request.body.read
      response = @server.handle_request(body)
      
      [200, { 'Content-Type' => 'application/json' }, [response]]
    rescue => e
      error_response = { error: e.message }.to_json
      [500, { 'Content-Type' => 'application/json' }, [error_response]]
    end
  end
end

# config.ru
run McpRackApp.new

Sinatra 集成

require 'sinatra'
require 'model_context_protocol'

class McpSinatraApp < Sinatra::Base
  def initialize
    super
    @server = ModelContextProtocol::Server.new(
      name: 'sinatra-mcp-server',
      version: '1.0.0'
    )
    @server.register_tool(EchoTool.new)
  end
  
  post '/mcp' do
    content_type :json
    
    begin
      response = @server.handle_request(request.body.read)
      response
    rescue => e
      status 500
      { error: e.message }.to_json
    end
  end
  
  get '/health' do
    { status: 'ok' }.to_json
  end
end

# 启动应用
McpSinatraApp.run!

Docker 部署

FROM ruby:3.2-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# 复制 Gemfile
COPY Gemfile Gemfile.lock ./

# 安装 gems
RUN bundle install --without development test

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 4567

# 启动应用
CMD ["ruby", "app.rb"]

Systemd 服务

# /etc/systemd/system/mcp-server.service
[Unit]
Description=MCP Server
After=network.target

[Service]
Type=simple
User=mcp
WorkingDirectory=/opt/mcp-server
ExecStart=/usr/local/bin/ruby server.rb
Restart=always
RestartSec=5
Environment=RACK_ENV=production

[Install]
WantedBy=multi-user.target

# 启用和启动服务
sudo systemctl enable mcp-server
sudo systemctl start mcp-server
sudo systemctl status mcp-server

相关资源

• Ruby SDK GitHub 仓库
• Ruby 官方文档
• RubyGems 文档
• Bundler 文档
• MCP 规范文档
• 示例项目

社区支持

• GitHub Discussions
• Discord 社区
• Ruby 社区论坛