MCP安全本地服务器
一种生产就绪、安全第一的模型上下文协议(MCP)服务器,在本地运行时具有严格的安全控制,同时允许对特定用例(如网络搜索)进行受控的外部网络访问。
特性
- 安全第一设计:所有操作都根据可配置的安全策略进行验证
- 网络防火墙:阻止所有外部网络访问,明确分配的端点除外
- 输入验证:JSON模式验证、路径遍历保护、命令清理
- 速率限制:每个工具的费率限制,以防止滥用
- 审计日志:JSON行格式日志记录,带有敏感数据编辑
- 插件系统:用于添加新工具的可扩展架构
- 符合MCP协议:通过STDIO传输的完整JSON-RPC 2.0
快速开始
安装
# Clone the repository
git clone
cd mcp-server
# Install dependencies with uv
uv sync运行服务器
# Run with default policy
uv run python main.py
# Run with custom policy file
uv run python main.py --policy /path/to/policy.yaml
# Show version
uv run python main.py --version与MCP客户端集成
此服务器可与任何兼容MCP的客户端配合使用。将以下内容添加到客户端的MCP配置中:
{
"mcpServers": {
"secure-local": {
"command": "uv",
"args": ["run", "python", "/path/to/mcp-server/main.py"],
"env": {}
}
}
}客户端配置位置示例:
- 克劳德桌面版:
~/Library/Application Support/Claude/claude_desktop_config.json(macOS) - 其他MCP客户端:有关配置文件位置,请参阅客户的文档
建筑
mcp-server/
├── main.py # CLI entry point
├── config/
│ └── policy.yaml # Security policy configuration
├── src/
│ ├── server.py # Main MCP server
│ ├── protocol/
│ │ ├── jsonrpc.py # JSON-RPC 2.0 parsing
│ │ ├── transport.py # STDIO transport
│ │ ├── lifecycle.py # MCP lifecycle management
│ │ └── tools.py # tools/list & tools/call handlers
│ ├── plugins/
│ │ ├── base.py # Plugin base class
│ │ ├── loader.py # Plugin discovery
│ │ ├── dispatcher.py # Tool call routing
│ │ ├── discovery.py # Built-in: Progressive disclosure tools
│ │ ├── websearch.py # Example: DuckDuckGo search plugin
│ │ └── bugtracker.py # Example: Bug tracking plugin
│ └── security/
│ ├── policy.py # Policy loader
│ ├── firewall.py # Network access control
│ ├── validator.py # Input validation
│ ├── engine.py # Integrated security engine
│ └── audit.py # Audit logging
└── tests/ # Test suite (343 tests, 96%+ coverage)安全策略
安全策略以YAML格式定义。看 config/policy.yaml 举一个完整的例子。
网络安全
network:
# Allowed local network ranges
allowed_ranges:
- "127.0.0.0/8"
- "10.0.0.0/8"
- "192.168.0.0/16"
# Explicitly allowed external endpoints
allowed_endpoints:
- host: "lite.duckduckgo.com"
ports: [443]
description: "DuckDuckGo search"
# Blocked ports (even on local network)
blocked_ports:
- 22 # SSH
# DNS settings
allow_dns: true
dns_allowlist:
- "lite.duckduckgo.com"文件系统安全
filesystem:
# Allowed paths (supports globs and env vars)
allowed_paths:
- "${HOME}/projects/**"
- "/tmp/mcp-workspace/**"
# Denied paths (takes precedence)
denied_paths:
- "**/.ssh/**"
- "**/.aws/**"
- "**/*.pem"
- "**/.env"工具配置
tools:
# Rate limits (requests per minute)
rate_limits:
default: 60
web_search: 20
# Execution timeout
timeout: 30审计日志
audit:
log_file: "${HOME}/.mcp-secure/audit.log"
log_level: "INFO"内置工具
服务器自动注册发现工具以进行渐进式披露,使代理能够高效地找到并仅加载他们需要的工具。
搜索工具
按关键字或类别搜索可用工具。使用 detail_level 以控制上下文使用。
输入架构:
{
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Keyword to search in tool names and descriptions"
},
"category": {
"type": "string",
"description": "Filter by plugin category (e.g., 'bugtracker')"
},
"detail_level": {
"type": "string",
"enum": ["name", "summary", "full"],
"description": "Level of detail: 'name' (just names), 'summary' (names + descriptions), 'full' (complete schemas)"
}
}
}示例-使用最少的上下文查找与bug相关的工具:
{
"name": "search_tools",
"arguments": {
"query": "bug",
"detail_level": "name"
}
}
// Returns: ["add_bug", "get_bug", "update_bug", "close_bug", "list_bugs", "search_bugs_global"]示例-获取特定类别的完整架构:
{
"name": "search_tools",
"arguments": {
"category": "websearch",
"detail_level": "full"
}
}列表_类别
列出所有可用的工具类别(插件)和工具计数。在搜索之前,使用此功能发现功能。
输入架构:
{
"type": "object",
"properties": {}
}示例响应:
[
{
"category": "discovery",
"version": "1.0.0",
"tool_count": 2,
"tools": ["search_tools", "list_categories"]
},
{
"category": "websearch",
"version": "1.0.0",
"tool_count": 1,
"tools": ["web_search"]
},
{
"category": "bugtracker",
"version": "1.0.0",
"tool_count": 7,
"tools": ["init_bugtracker", "add_bug", "get_bug", "update_bug", "close_bug", "list_bugs", "search_bugs_global"]
}
]示例插件
服务器包括示例插件来演示插件架构。这些是作为参考实现提供的,展示了如何为任何用例构建自己的插件。
web_search(示例插件)
一个使用DuckDuckGo搜索网络的示例插件。演示如何构建在安全策略内发出外部网络请求的插件。
输入架构:
{
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"max_results": {
"type": "integer",
"description": "Maximum results to return (default: 5)"
}
},
"required": ["query"]
}例子:
{
"name": "web_search",
"arguments": {
"query": "Python asyncio tutorial",
"max_results": 3
}
}Bug跟踪器(示例插件)
一个使用集中式SQLite数据库实现本地错误跟踪系统的示例插件。演示如何构建管理本地状态、支持多个项目和执行复杂查询的插件。
T错误跟踪器
初始化项目的错误跟踪。
输入架构:
{
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "Path to project directory (defaults to cwd)"
}
}
}add_bug
在跟踪器中添加一个新bug。
输入架构:
{
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Brief title for the bug"
},
"description": {
"type": "string",
"description": "Detailed description"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"],
"description": "Bug priority (default: medium)"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Tags for categorization"
},
"project_path": {
"type": "string",
"description": "Path to project directory (defaults to cwd)"
}
},
"required": ["title"]
}get_bug
按ID检索错误。
输入架构:
{
"type": "object",
"properties": {
"bug_id": {
"type": "string",
"description": "The bug ID to retrieve"
},
"project_path": {
"type": "string",
"description": "Path to project directory (defaults to cwd)"
}
},
"required": ["bug_id"]
}update_bug
更新现有bug的状态、优先级、标签或相关bug。支持仅用于进度跟踪的注释更新。
输入架构:
{
"type": "object",
"properties": {
"bug_id": {
"type": "string",
"description": "The bug ID to update"
},
"status": {
"type": "string",
"enum": ["open", "in_progress", "closed"]
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "New tags (replaces existing)"
},
"related_bugs": {
"type": "array",
"description": "Related bugs with relationship type"
},
"note": {
"type": "string",
"description": "Note for the history entry"
},
"project_path": {
"type": "string"
}
},
"required": ["bug_id"]
}close_bug
使用解决方案注释关闭错误。
输入架构:
{
"type": "object",
"properties": {
"bug_id": {
"type": "string",
"description": "The bug ID to close"
},
"resolution": {
"type": "string",
"description": "Resolution note explaining how the bug was fixed"
},
"project_path": {
"type": "string"
}
},
"required": ["bug_id"]
}list_bugs
列出带有可选过滤功能的错误。
输入架构:
{
"type": "object",
"properties": {
"status": {
"type": "string",
"enum": ["open", "in_progress", "closed"]
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Filter by tags (must have ALL specified tags)"
},
"project_path": {
"type": "string"
}
}
}search_bugs_global
在所有索引项目中搜索错误。
输入架构:
{
"type": "object",
"properties": {
"status": {
"type": "string",
"enum": ["open", "in_progress", "closed"]
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
},
"tags": {
"type": "array",
"items": {"type": "string"}
}
}
}示例-创建并跟踪错误:
// Add a bug
{
"name": "add_bug",
"arguments": {
"title": "Login button not responding",
"description": "The login button on the home page doesn't trigger the auth flow",
"priority": "high",
"tags": ["ui", "auth"]
}
}
// Update with progress
{
"name": "update_bug",
"arguments": {
"bug_id": "BUG-001",
"status": "in_progress",
"note": "Identified missing onClick handler"
}
}
// Close with resolution
{
"name": "close_bug",
"arguments": {
"bug_id": "BUG-001",
"resolution": "Added onClick handler to LoginButton component"
}
}创建自定义插件
Python插件
插件必须继承自 PluginBase 并实施所需的方法:
from src.plugins.base import PluginBase, ToolDefinition, ToolResult
class MyPlugin(PluginBase):
@property
def name(self) -> str:
return "my_plugin"
@property
def version(self) -> str:
return "1.0.0"
def get_tools(self) -> list[ToolDefinition]:
return [
ToolDefinition(
name="my_tool",
description="Does something useful",
input_schema={
"type": "object",
"properties": {
"input": {"type": "string"}
},
"required": ["input"]
},
)
]
def execute(self, tool_name: str, arguments: dict) -> ToolResult:
if tool_name == "my_tool":
result = do_something(arguments["input"])
return ToolResult(
content=[{"type": "text", "text": result}]
)
return ToolResult(
content=[{"type": "text", "text": "Unknown tool"}],
is_error=True
)在中注册插件 main.py:
from my_plugin import MyPlugin
server.register_plugin(MyPlugin())外部插件(非Python)
插件系统可以通过子流程包装器方法支持用任何语言(Rust、JavaScript、TypeScript、Go等)编写的工具。这是一个计划中的功能——欢迎投稿。
架构概述
外部插件作为单独的进程运行,通过stdin/stdout上的JSON与Python包装器通信:
┌─────────────────────────────────────────────────────────────┐
│ MCP Server (Python) │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ WebSearch │ │ BugTracker │ │ External │ │
│ │ (Python) │ │ (Python) │ │ Plugin │ │
│ └──────────────┘ └──────────────┘ │ (Wrapper) │ │
│ └──────┬───────┘ │
│ │ │
└─────────────────────────────────────────────────┼────────────┘
│ JSON/stdin/stdout
▼
┌──────────────┐
│ my-rust-tool │
│ (subprocess) │
└──────────────┘运作原理
- Python包装器:薄
ExternalPlugin类继承自PluginBase并处理子流程生命周期 - 清单A.
manifest.yaml声明工具定义并指向可执行文件 - 合同:外部工具在stdin上接收JSON并将JSON写入stdout
清单格式
name: my-rust-tools
version: "1.0.0"
type: external
executable: ./target/release/my-rust-tool
tools:
- name: calculate_hash
description: Calculate cryptographic hash of input
input_schema:
type: object
properties:
algorithm:
type: string
enum: [sha256, sha512, blake3]
input:
type: string
required: [algorithm, input]外部工具合同
外部可执行文件必须:
- 接受 stdin上的JSON对象:
{
"tool": "calculate_hash",
"arguments": {
"algorithm": "sha256",
"input": "hello world"
}
}- 返回 stdout上的JSON对象:
{
"content": [
{"type": "text", "text": "sha256: b94d27b9934d3e08..."}
],
"isError": false
}- 退出 成功时代码为0,失败时代码为非零
示例:Rust工具
use serde::{Deserialize, Serialize};
use std::io::{self, BufRead, Write};
#[derive(Deserialize)]
struct Request {
tool: String,
arguments: serde_json::Value,
}
#[derive(Serialize)]
struct Response {
content: Vec,
#[serde(rename = "isError")]
is_error: bool,
}
#[derive(Serialize)]
struct Content {
#[serde(rename = "type")]
content_type: String,
text: String,
}
fn main() {
let stdin = io::stdin();
let line = stdin.lock().lines().next().unwrap().unwrap();
let request: Request = serde_json::from_str(&line).unwrap();
let result = match request.tool.as_str() {
"calculate_hash" => calculate_hash(request.arguments),
_ => Err(format!("Unknown tool: {}", request.tool)),
};
let response = match result {
Ok(text) => Response {
content: vec![Content { content_type: "text".into(), text }],
is_error: false,
},
Err(e) => Response {
content: vec![Content { content_type: "text".into(), text: e }],
is_error: true,
},
};
println!("{}", serde_json::to_string(&response).unwrap());
}示例:Node.js工具
const readline = require('readline');
const rl = readline.createInterface({ input: process.stdin });
rl.on('line', (line) => {
const request = JSON.parse(line);
let response;
try {
const result = handleTool(request.tool, request.arguments);
response = {
content: [{ type: 'text', text: result }],
isError: false
};
} catch (e) {
response = {
content: [{ type: 'text', text: e.message }],
isError: true
};
}
console.log(JSON.stringify(response));
process.exit(0);
});
function handleTool(tool, args) {
switch (tool) {
case 'format_json':
return JSON.stringify(JSON.parse(args.input), null, 2);
default:
throw new Error(`Unknown tool: ${tool}`);
}
}外部插件的安全注意事项
- 进程隔离:外部工具在具有自己内存空间的单独进程中运行
- 超时强制:包装器会杀死超过配置超时的子进程
- 无网络继承:子进程网络访问受操作系统级控制
- 可执行许可列表:只能调用已注册清单中列出的可执行文件
- 输入验证:JSON模式在传递给子流程之前经过验证
权衡
| 特性 | Python插件 | 外部插件 |
|---|---|---|
| 启动延迟 | 无 | 每次通话约10-50ms |
| 记忆 | 与服务器共享 | 独立进程 |
| 语言 | 仅限Python | 任何语言 |
| 调试 | 容易 | 难(单独过程) |
| 安全 | 共享内存空间 | 进程隔离 |
何时使用外部插件
- 性能关键工具:Rust/Go用于CPU密集型操作
- 现有CLI工具:包装现有二进制文件而不重写
- 特定语言库:使用npm包、货物箱等。
- 团队专业知识:让团队使用他们喜欢的语言
发展
运行测试
# Run all tests
uv run pytest
# Run with coverage report
uv run pytest --cov=src --cov-report=term-missing
# Run specific test file
uv run pytest tests/test_server.py -v掉毛
# Check for issues
uv run ruff check .
# Auto-fix issues
uv run ruff check --fix .
# Format code
uv run ruff format .项目结构
| 目录 | 目的 |
|---|---|
src/protocol/ | MCP协议实现(JSON-RPC、STDIO、生命周期) |
src/plugins/ | 插件系统和内置插件 |
src/security/ | 安全层(防火墙、验证、审计) |
tests/ | 测试套件 |
config/ | 配置文件 |
MCP协议支持
此服务器实现MCP协议版本 2025-11-25 支持:
| 方法 | 说明 |
|---|---|
initialize | 初始化连接 |
notifications/initialized | 确认初始化完成 |
tools/list | 列出可用工具 |
tools/call | 执行工具 |
安全考虑
- 网络隔离:默认情况下,所有外部网络访问都被阻止。只能到达明确分配的端点。
- 路径横向保护:所有文件路径都根据允许/拒绝模式进行验证,以防止访问敏感文件。
- 命令注入预防:命令经过净化,以阻止shell运算符等危险模式。
- 速率限制:每个工具的费率限制可防止滥用和资源枯竭。
- 审计跟踪:所有操作都记录了时间戳、请求ID和经过净化的参数。
许可证
MIT许可证-有关详细信息,请参阅许可证文件。
