MCP Secure Local Server

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) │
                                          └──────────────┘

运作原理

1. Python包装器：薄 ExternalPlugin 类继承自 PluginBase 并处理子流程生命周期
2. 清单A. manifest.yaml 声明工具定义并指向可执行文件
3. 合同：外部工具在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]

外部工具合同

外部可执行文件必须：

1. 接受 stdin上的JSON对象：

{
  "tool": "calculate_hash",
  "arguments": {
    "algorithm": "sha256",
    "input": "hello world"
  }
}

1. 返回 stdout上的JSON对象：

{
  "content": [
    {"type": "text", "text": "sha256: b94d27b9934d3e08..."}
  ],
  "isError": false
}

1. 退出 成功时代码为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}`);
  }
}

外部插件的安全注意事项

1. 进程隔离：外部工具在具有自己内存空间的单独进程中运行
2. 超时强制：包装器会杀死超过配置超时的子进程
3. 无网络继承：子进程网络访问受操作系统级控制
4. 可执行许可列表：只能调用已注册清单中列出的可执行文件
5. 输入验证：JSON模式在传递给子流程之前经过验证

权衡

何时使用外部插件

• 性能关键工具：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 .

项目结构

MCP协议支持

此服务器实现MCP协议版本 2025-11-25 支持：

安全考虑

1. 网络隔离：默认情况下，所有外部网络访问都被阻止。只能到达明确分配的端点。

1. 路径横向保护：所有文件路径都根据允许/拒绝模式进行验证，以防止访问敏感文件。

1. 命令注入预防：命令经过净化，以阻止shell运算符等危险模式。

1. 速率限制：每个工具的费率限制可防止滥用和资源枯竭。

1. 审计跟踪：所有操作都记录了时间戳、请求ID和经过净化的参数。

许可证

MIT许可证-有关详细信息，请参阅许可证文件。