再给我来一次“击中我”（或“再狠狠地来一下”）
一个具有插件架构的MCP（模型上下文协议）服务器，用于演示速率限制概念。
了解更多
阅读构建这台服务器的完整故事： 在Elixir中构建限速MCP服务器时学到的7件事
特点
- MCP服务器通过标准输入输出实现完整的JSON-RPC 2.0协议
- 插件架构在请求到达数据库之前进行拦截的中间件管道
- 速率限制固定窗口速率限制器（每10秒5次请求）
- 缓存层响应缓存，TTL（生存时间）为30秒
- SQLite 数据库100条样本记录，支持分页功能
- 综合测试28项测试涵盖所有组件
建筑学
┌─────────────┐
│ MCP Client │
└──────┬──────┘
│ JSON-RPC 2.0
▼
┌─────────────────────────────────┐
│ MCP Server (stdio) │
│ - Protocol Handler │
│ - Tool Registry │
└──────┬──────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ Plugin Pipeline │
│ ┌────────────────────────┐ │
│ │ 1. Rate Limiter │ │
│ │ (5 req/10s) │ │
│ │ ├─ if exceeded ──┐ │ │
│ │ │ │ │ │
│ │ ▼ │ │ │
│ │ 2. Logging Plugin │ │ │
│ │ (logs requests) │ │ │
│ │ ▼ │ │ │
│ │ 3. Cache Plugin │ │ │
│ │ (30s TTL) │ │ │
│ │ ├─ if hit ───┐ │ │ │
│ │ │ │ │ │ │
│ └────┼────────────┼───┼──┘ │
│ │ │ │ │
│ ▼ │ │ │
│ Database │ │ │
│ │ │ │ │
│ └────────────┴───┴────────┤
│ │
│ All short-circuits exit ────┤
│ and return response here │
└─────────────────────────────────┘
│
▼
Response to Client安装
# Install dependencies
mix deps.get
# Compile the project
mix compile
# Run tests
mix test运行服务器
# Start the MCP server
mix run --no-halt服务器通过stdio使用JSON-RPC 2.0协议进行通信。
可用工具
get_records
从数据库中分页检索N条记录。
参数：
limit（数字，可选）：要获取的记录数量（默认：10，最大：100）offset(数字，可选)：要跳过的记录数（默认：0）
示例请求：
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "get_records",
"arguments": {
"limit": 10,
"offset": 0
}
}
}示例回复：
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "{\n \"items\": [\n {\n \"id\": 1,\n \"name\": \"Item 1\",\n \"description\": \"This is a sample item...\",\n \"created_at\": \"2024-01-20T12:00:00Z\"\n }\n ],\n \"pagination\": {\n \"limit\": 10,\n \"offset\": 0,\n \"total\": 100,\n \"has_more\": true\n }\n}"
}
]
}
}插件架构
插件系统允许在请求到达数据库之前进行拦截。这对于实现以下功能非常完美：
- 速率限制
- 缓存
- 认证
- 请求验证
- 日志记录和监控
插件的工作原理
每个插件都实现了 HitMeDbOneMoreTime.Plugins.Behaviour 并且可以：
- 通过让请求继续传递给下一个插件
- 短路立即返回响应，无需访问数据库
- 更新上下文修改下游插件的共享上下文
示例：缓存插件
缓存插件展示了如何为限流请求设置快速通道：
def process(request, context) do
cache_key = generate_cache_key(request)
case get_from_cache(cache_key) do
{:ok, cached_response} ->
# Short-circuit! Return cached response without hitting DB
Logger.info("[Cache Plugin] Cache HIT")
{:respond, cached_response}
:miss ->
# Pass to next plugin/handler
Logger.info("[Cache Plugin] Cache MISS")
{:pass, Map.put(context, :cache_key, cache_key)}
end
end示例：速率限制插件
限流插件展示了请求计数和拒绝机制：
def process(request, context) do
client_id = extract_client_id(request, context)
case check_rate_limit(client_id, max_requests, window_seconds) do
{:ok, current_count} ->
# Within limits - pass through
Logger.info("[Rate Limiter] Request allowed (#{current_count}/#{max_requests})")
{:pass, Map.put(context, :rate_limit_current, current_count)}
{:error, :rate_limit_exceeded, retry_after} ->
# Exceeded - short-circuit with error
Logger.warning("[Rate Limiter] Rate limit exceeded")
{:respond, %{
"error" => "rate_limit_exceeded",
"message" => "Too many requests. Please try again later.",
"details" => %{
"limit" => max_requests,
"retry_after_seconds" => retry_after
}
}}
end
end主要特点：
- 固定窗口算法每10秒窗口内允许5次请求
- ETS存储针对每个客户端/工具的内存中追踪
- 自动重置窗口过期后计数器重置
- 在限制上短路（或：超出限制导致短路）在不触碰数据库或缓存的情况下返回错误
配置：
您可以通过上下文自定义速率限制：
context = %{
rate_limit_max: 10, # 10 requests
rate_limit_window: 60 # per 60 seconds
}添加自定义插件
- 创建一个实现该行为的新模块：
defmodule MyPlugin do
@behaviour HitMeDbOneMoreTime.Plugins.Behaviour
def process(request, context) do
# Your logic here
:pass # or {:respond, response} or {:pass, updated_context}
end
end- 将其添加到流水线中
lib/hit_me_db_one_more_time/mcp/tools.ex：
plugins = [
RateLimiterPlugin, # Check rate limits first
LoggingPlugin, # Log allowed requests
CachePlugin, # Check cache for allowed requests
MyPlugin # HitMeDbOneMoreTime.MCP.Tools.execute_tool("get_records", %{"limit" => 5})为您的博客文章
这个服务器展示了完整的速率限制实现，包含以下关键概念：
1. 插件流水线架构
请求通过中间件管道流动，其中每个插件都可以：
- 检查请求
- 短路并提前返回
- 将上下文传递给下游插件
- 修改或拒绝请求
2. 速率限制的实现
这个 RateLimiterPlugin 展示；证明
- 固定窗口算法在时间窗口内追踪请求（每10秒5个请求）
- ETS存储每个客户端的快速内存计数器
- 自动过期Windows在一段时间后重置
- 优雅降级在错误中返回重试后的时间
3. 多层防御
Request → Rate Limiter → Logger → Cache → Database
└─ Denies └─ Logs └─ Serves └─ Fetches
excessive only cached fresh
requests allowed results data4. 这种方法的优势
演出：
- 限速请求永远不会访问数据库
- 缓存的响应完全绕过数据库
- 日志仅记录允许的请求
可扩展性:
- ETS 提供微秒级的查找速度
- 基本速率限制无需外部依赖
- 在分布式系统中，易于将ETS替换为Redis
可观测性:
- 每一层的结构化日志记录
- 响应上下文中的速率限制指标
- 清晰的错误信息，附带重试指导
5. 生产方面的考量
对于生产用途，您会通过以下方式对其进行增强：
- 滑动窗口 用于更平滑限速的算法
- Redis（注：Redis是一个开源的、基于内存的键值对存储系统，常用于数据库、缓存和消息中间件等场景） 用于跨服务器的分布式速率限制
- 按用户/API密钥 按整体跟踪，而非按工具分别跟踪
- 不同的限制 针对不同的终端或用户等级
- 速率限制头部信息 (X-RateLimit-Limit, X-RateLimit-Remaining) 翻译为中文是：(速率限制上限，剩余速率限制)
- 优雅降级 当速率限制存储失败时的策略
许可证
这是一个用于撰写博客文章的演示项目。