mcp通用
   
版本: 0.6.0(原住民) 状态: 生产就绪
______________________________________________________________________
快速链接
概述
mcp常见的是 Oneiric本土基金会图书馆 用于构建生产级MCP(模型上下文协议)服务器。它提供从生产服务器(包括Crackerjack、Session Buddy和FastBlocks)中提取的经过战斗测试的模式。
质量与CI
Crackerjack是标准的质量控制和CI/CD门,用于更改此库和构建在其上的下游MCP服务器。
🎯 这个库提供了什么:
- 刀具轮廓系统 (v0.6.0+)-门控工具注册,以减少MCP上下文开销(5台服务器上的391个工具)
- 描述修剪 (v0.6.0+)-将工具文档字符串修剪为200个字符以提高令牌效率的实用程序
- Oneiric CLI工厂 (v0.3.3+)-使用启动/停止/重启/状态/运行状况命令的标准化服务器生命周期
- HTTP客户端适配器 -使用httpx连接池实现11倍性能
- 提示/通知适配器 🆕 - 具有自动后端检测功能的统一跨平台用户交互
- 安全实用程序 -API密钥验证(缓存速度快90%)和输入净化(速度快2倍)
- 丰富的控制台用户界面 -用于服务器操作的漂亮面板和通知
- 设置管理 -YAML+环境变量配置(基于Pydantic)
- 健康检查系统 -生产就绪健康监测
- 类型安全 -完整的Pydantic验证和类型提示
- 综合测试 -615测试,基于属性和并发测试
设计原则:
- Oneiric原住民 -直接Pydantic、丰富的库和标准模式
- 生产就绪 -从实际生产系统中提取
- 分层配置 -YAML文件+具有明确优先级的环境变量
- 丰富的用户界面 -配备丰富面板的专业控制台输出
- 类型安全 -带有严格MyPy检查的完整类型提示
- 测试良好 -最低覆盖率为90%
______________________________________________________________________
📚 示例
看 examples/ 完整的生产就绪示例:
1.CLI服务器(Oneiric Native)-v0.3.3中的新功能
演示 CLI工厂 对于标准化的服务器生命周期管理:
- 5个生命周期命令(启动、停止、重新启动、状态、运行状况)
- 带安全验证的PID文件管理
- 运行时运行状况快照
- 带信号处理的优雅关机
- 自定义生命周期处理程序
cd examples
python cli_server.py start
python cli_server.py status
python cli_server.py health
python cli_server.py stop2.气象MCP服务器(Oneiric Native)
展示 HTTP适配器 和 FastMCP集成:
- 带连接池的HTTPClientAdapter(性能为11倍)
- 带YAML+环境配置的MCPBaseSettings
- ServerPanels提供美观的终端用户界面
- Oneiric配置模式(直接实例化)
- FastMCP工具集成(可选;单独安装)
cd examples
python weather_server.py完整文档: examples/README.md
______________________________________________________________________
快速开始
安装
pip install mcp-common>=0.3.6这会自动安装Pydantic、Rich和所有必需的依赖项。
如果您计划运行MCP服务器(例如示例),请单独安装FastMCP等协议主机:
pip install fastmcp
# or
uv add fastmcp最小示例
# my_server/settings.py
from mcp_common.config import MCPBaseSettings
from pydantic import Field
class MyServerSettings(MCPBaseSettings):
"""Server configuration following Oneiric pattern.
Loads from (priority order):
1. settings/local.yaml (gitignored)
2. settings/my-server.yaml
3. Environment variables MY_SERVER_*
4. Defaults below
"""
api_key: str = Field(description="API key for service")
timeout: int = Field(default=30, description="Request timeout")
# my_server/main.py
from fastmcp import FastMCP # Optional: install fastmcp separately
from mcp_common import ServerPanels, HTTPClientAdapter, HTTPClientSettings
from my_server.settings import MyServerSettings
# Initialize
mcp = FastMCP("MyServer")
settings = MyServerSettings.load("my-server")
# Initialize HTTP adapter
http_settings = HTTPClientSettings(timeout=settings.timeout)
http_adapter = HTTPClientAdapter(settings=http_settings)
# Define tools
@mcp.tool()
async def call_api():
# Use the global adapter instance
response = await http_adapter.get("https://api.example.com")
return response.json()
# Run server
if __name__ == "__main__":
# Display startup panel
ServerPanels.startup_success(
server_name="My MCP Server",
version="1.0.0",
features=["HTTP Client", "YAML Configuration"],
)
mcp.run()______________________________________________________________________
核心功能
🔌 HTTP客户端适配器
使用httpx进行连接池:
- 比每次请求创建客户端快11倍
- 自动初始化和清理
- 可配置的超时、重试、连接限制
from mcp_common import HTTPClientAdapter, HTTPClientSettings
# Configure HTTP adapter
http_settings = HTTPClientSettings(
timeout=30,
max_connections=50,
retry_attempts=3,
)
# Create adapter
http_adapter = HTTPClientAdapter(settings=http_settings)
# Make requests
response = await http_adapter.get("https://api.example.com")架构概述:
graph TB
subgraph "mcp-common Components"
A[HTTP Client Adapter
with Connection Pooling]
B[Settings Management
YAML + Env Vars]
C[CLI Factory
Lifecycle Management]
D[Rich UI Panels
Console Output]
E[Security Utilities
Validation & Sanitization]
end
subgraph "Integration"
F[FastMCP
Optional]
G[MCP Server
Application]
end
A --> G
B --> G
C --> G
D --> G
E --> G
F --> G
style A fill:#e1f5fe
style B fill:#f3e5f5
style C fill:#e8f5e8
style D fill:#fff3e0
style E fill:#fce4ec注意:此库不提供速率限制。如果你使用FastMCP,它内置 RateLimitingMiddleware 可以启用;否则,请使用特定于项目的配置。
🎯 Oneiric CLI工厂(v0.3.3中的新功能)
生产就绪服务器生命周期管理:
这 MCPServerCLIFactory 受Oneiric操作模式的启发,提供了用于管理MCP服务器生命周期的标准化CLI命令。它处理流程管理、健康监控和开箱即用的优雅关机。
特征:
- 5个标准命令 -
start,stop,restart,status,health - 安全第一 -安全PID文件(0o600)、缓存目录(0o700)、所有权验证
- 工艺验证 -检测过时的PID,防止竞争情况,验证进程身份
- 健康监测 -具有可配置TTL的运行时运行状况快照
- 信号处理 -SIGTERM/SIGINT上的优雅关机
- 自定义处理程序 -针对服务器特定逻辑的可扩展生命周期挂钩
- 双输出 -人类可读和JSON输出模式
- 标准出口代码 -带有语义退出代码的Shell可脚本化
CLI工厂架构:
graph LR
subgraph "User Application"
A[Server Implementation]
B[Custom Handlers]
end
subgraph "mcp-common CLI Factory"
C[MCPServerCLIFactory]
D[MCPServerSettings]
E[PID File Management]
F[Health Snapshots]
G[Signal Handlers]
end
subgraph "Typer CLI"
H[start command]
I[stop command]
J[restart command]
K[status command]
L[health command]
end
A --> C
B --> C
D --> C
C --> E
C --> F
C --> G
C --> H
C --> I
C --> J
C --> K
C --> L
style A fill:#e8f5e8
style B fill:#fff3e0
style C fill:#e3f2fd
style D fill:#f3e5f5
style H fill:#e0f2f1
style I fill:#e0f2f1
style J fill:#e0f2f1
style K fill:#e0f2f1
style L fill:#e0f2f1快速示例:
from mcp_common.cli import MCPServerCLIFactory, MCPServerSettings
# 1. Load settings (YAML + env vars)
settings = MCPServerSettings.load("my-server")
# 2. Define lifecycle handlers
def start_server():
print("Server initialized!")
# Your server startup logic here
def stop_server(pid: int):
print(f"Stopping PID {pid}")
# Your cleanup logic here
def check_health():
# Return current health snapshot
return RuntimeHealthSnapshot(
orchestrator_pid=os.getpid(),
watchers_running=True,
)
# 3. Create CLI factory
factory = MCPServerCLIFactory(
server_name="my-server",
settings=settings,
start_handler=start_server,
stop_handler=stop_server,
health_probe_handler=check_health,
)
# 4. Create and run Typer app
app = factory.create_app()
if __name__ == "__main__":
app()命令用法:
# Start server (creates PID file and health snapshot)
python my_server.py start
# Check status (lightweight process check)
python my_server.py status
# Output: Server running (PID 12345, snapshot age: 2.3s, fresh: True)
# View health (detailed health information)
python my_server.py health
# Live health probe
python my_server.py health --probe
# Stop server (graceful shutdown with SIGTERM)
python my_server.py stop
# Force stop with timeout
python my_server.py stop --timeout 5 --force
# Restart (stop + start)
python my_server.py restart
# JSON output for automation
python my_server.py status --json配置:
设置从多个来源加载(优先级顺序):
settings/local.yaml(gitignored,用于发展)settings/{server-name}.yaml(已签入回购)- 环境变量
MCP_SERVER_* - 默认值
MCPServerSettings
示例 settings/my-server.yaml:
server_name: "My MCP Server"
cache_root: .oneiric_cache
health_ttl_seconds: 60.0
log_level: INFO退出代码:
0-成功1-一般错误2-服务器未运行(状态/停止)3-服务器已在运行(启动)4-健康检查失败5-配置错误6-权限错误7-超时8-过时的PID文件(使用--force)
完整示例:
看 examples/cli_server.py 查看带有自定义命令和健康探测器的完整工作示例。
⚙️ 支持YAML的设置(Oneiric模式)
- 纯Pydantic基模型
- 分层配置:YAML文件+环境变量
- 使用Pydantic进行类型验证
- 路径扩展(
~→ 主目录)
from mcp_common.config import MCPBaseSettings
class ServerSettings(MCPBaseSettings):
api_key: str # Required
timeout: int = 30 # Optional with default
# Load with layered configuration
settings = ServerSettings.load("my-server")
# Loads from:
# 1. settings/my-server.yaml
# 2. settings/local.yaml
# 3. Environment variables MY_SERVER_*
# 4. Defaults📝 标准Python日志
mcp-common使用标准的Python日志记录。根据需要为您的服务器配置:
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
logger.info("Server started")🎨 丰富的控制台用户界面
- 漂亮的启动面板
- 错误显示与上下文
- 统计表
- 进度条
from mcp_common.ui import ServerPanels
ServerPanels.startup_success(
server_name="Mailgun MCP",
http_endpoint="http://localhost:8000",
features=["Rate Limiting", "Security Filters"],
)🧪 测试工具
- 模拟MCP客户端
- HTTP响应模拟
- 共享装置
- DI友好测试
from mcp_common.testing import MockMCPClient, mock_http_response
async def test_tool():
with mock_http_response(status=200, json={"ok": True}):
result = await my_tool()
assert result["success"]🔧 刀具轮廓系统(v0.6.0中的新功能)
通过在启动时选择注册哪些工具来减少MCP上下文开销。每个服务器读取一个 {SERVER_NAME}_TOOL_PROFILE 环境变量(minimal, standard,或 full)并且仅注册相应的工具组。
为什么? 一个拥有170个工具的服务器在每次请求时都会向Claude发送大约70k个工具定义令牌。配置文件门控将每日开发的代币减少到约10-20k。
ToolProfile枚举:
from mcp_common.tools import ToolProfile, trim_description, MANDATORY_TOOLS
# Resolve from environment (defaults to FULL for backward compatibility)
profile = ToolProfile.from_env("MY_SERVER_TOOL_PROFILE")
# Ordering comparisons work
assert ToolProfile.MINIMAL < ToolProfile.STANDARD < ToolProfile.FULL
# Safe fallback for invalid values
assert ToolProfile.from_string("unknown") == ToolProfile.FULL描述修剪:
# Strip Args/Returns/Raises sections, keep first paragraph, max 200 chars
trimmed = trim_description("""Check health of a service.
Args:
service_name: Name of the service
port: Port number
Returns:
Health status dictionary""")
# Result: "Check health of a service."每服务器配置文件.py模式:
# my_server/mcp/tools/profiles.py
from mcp_common.tools import ToolProfile
MINIMAL_REGISTRATIONS = ["register_health_tools"]
STANDARD_REGISTRATIONS = MINIMAL_REGISTRATIONS + ["register_core_tools"]
FULL_REGISTRATIONS = STANDARD_REGISTRATIONS + ["register_advanced_tools"]
PROFILE_REGISTRATIONS = {
ToolProfile.MINIMAL: MINIMAL_REGISTRATIONS,
ToolProfile.STANDARD: STANDARD_REGISTRATIONS,
ToolProfile.FULL: FULL_REGISTRATIONS,
}
def get_active_profile(env_var="MY_SERVER_TOOL_PROFILE"):
return ToolProfile.from_env(env_var)整个生态系统的配置文件层次:
| 服务器 | 最小 | 标准 | 满 |
|---|---|---|---|
| 会话伙伴 | 4组(~12个工具) | 13组(~35个工具) | 32组(~151个工具) |
| 马哈维什努 | 1组(健康) | 7组 | 14组(约174个工具) |
| crackerjack | 2组 | 7组 | 12组(~60个工具) |
| akosha | 1组(健康) | 2组 | 4组(~5个工具) |
| dhara | 1组(kv/店) | 3组 | 3组(~17个工具) |
discover_tools元工具: 每个服务器注册一个 discover_tools(query) 始终可用的工具,让Claude找到卸载的工具并建议更改配置文件。
______________________________________________________________________
文档
- 示例/README.md - 从这里开始 -示例服务器和使用模式
- **ONEIRIC_CLI_FACTORY\_\*.md** -CLI工厂文档和实施指南
______________________________________________________________________
完整示例
看 examples/ 用于演示MCP常见模式的完整生产就绪Weather MCP服务器。
展示的关键模式:
- Oneiric设置 -YAML+环境变量配置
.load() - 超文本传输协议适配器 -带连接池的HTTPClientAdapter
- 丰富的用户界面 -ServerPanels用于启动/错误/状态
- 工具组织 -使用FastMCP进行模块化工具注册
- 配置分层 -具有明确优先级的多个配置源
- 类型安全 -全程进行完整的Pydantic验证
- 错误处理 -ServerPanels显示优美的错误
______________________________________________________________________
性能基准
✨ 第4阶段优化(v0.6.0)
消毒早期退出优化:
| 场景 | 之前 | 之后 | 加速 |
|---|---|---|---|
| 干净文本(无敏感数据) | 22μs | 10μs | 速度提高2.2倍 ⚡ |
| 包含敏感数据的文本 | 22μs | 22μs | 无变化 |
API密钥验证缓存:
| 通话类型 | 时间 | 加速 |
|---|---|---|
| 第一次调用(未缓存) | 100μs | 基线 |
| 后续调用(缓存) | 10μs | 快10倍 ⚡ |
影响:
- 干净文本净化速度提高2倍(最常见的情况)
- 重复API密钥验证速度提高10倍
- 缓存大小:128个最新条目
- 零突破性变化
HTTP客户端适配器(与每个请求的新客户端相比)
Before: 100 requests in 45 seconds, 500MB memory
After: 100 requests in 4 seconds, 50MB memory
Result: 11x faster, 10x less memory费率限制器开销
Without: 1000 requests in 1.2 seconds
With: 1000 requests in 1.25 seconds
Result: +4% overhead (negligible vs network I/O)📊 测试性能
测试套件增长:
| 版本 | 测试 | 覆盖率 | 执行时间 |
|---|---|---|---|
| v0.5.2 | 564 | 94% | 约110秒 |
| v0.6.0 | 615 | 99%+ | 约120秒 |
测试能力:
- ✅ 20个基于属性的测试(假设)
- ✅ 10个并发测试(线程安全)
- ✅ 7次性能优化测试
- ✅ 保持100%的向后兼容性
______________________________________________________________________
使用模式
模式1:使用YAML配置设置
from mcp_common.config import MCPBaseSettings
from pydantic import Field
class MySettings(MCPBaseSettings):
api_key: str = Field(description="API key")
timeout: int = Field(default=30, description="Timeout")
# Load from settings/my-server.yaml + env vars
settings = MySettings.load("my-server")
# Access configuration
print(f"Using API key: {settings.get_masked_key()}")模式2:使用HTTP客户端适配器
from mcp_common import HTTPClientAdapter, HTTPClientSettings
# Configure HTTP client
http_settings = HTTPClientSettings(
timeout=30,
max_connections=50,
retry_attempts=3,
)
# Create adapter
http = HTTPClientAdapter(settings=http_settings)
# Make requests
@mcp.tool()
async def call_api():
response = await http.get("https://api.example.com/data")
return response.json()
# Cleanup when done
await http._cleanup_resources()模式3:显示丰富的UI面板
from mcp_common import ServerPanels
# Startup panel
ServerPanels.startup_success(
server_name="My Server",
version="1.0.0",
features=["Feature 1", "Feature 2"],
)
# Error panel
ServerPanels.error(
title="API Error",
message="Failed to connect",
suggestion="Check your API key",
)
# Status table
ServerPanels.status_table(
title="Health Check",
rows=[
("API", "✅ Healthy", "200 OK"),
("Database", "⚠️ Degraded", "Slow queries"),
],
)______________________________________________________________________
发展
设置
git clone https://github.com/lesaker/mcp-common.git
cd mcp-common
pip install -e ".[dev]"运行测试
# Run all tests with coverage
pytest --cov=mcp_common --cov-report=html
# Run specific test
pytest tests/test_http_adapter.py -v
# Run integration tests
pytest tests/integration/ -v代码质量
# Format code
ruff format
# Lint code
ruff check
# Type checking
mypy mcp_common tests
# Run all quality checks
crackerjack --all______________________________________________________________________
版本控制
最新版本:
- 0.6.0 -刀具轮廓系统,描述修整,MANDATORY_TOOLS
- 0.3.6 -Oneiric本地(生产就绪)
- 0.3.3 -添加Oneiric CLI工厂
- 0.3.0 -初始Oneiric模式
兼容性:
- 需要Python 3.13+
- 可选:与FastMCP 2.0兼容+
- 使用Pydantic 2.12+,富14.2+
______________________________________________________________________
成功指标
当前状态:
- ✅ 所有组件均配备专业丰富的用户界面
- ✅ 保持90%以上的测试覆盖率
- ✅ 零生产事故
- ✅ Oneiric本地模式贯穿始终
- ✅ 标准化CLI生命周期管理
- ✅ 干净的依赖树(无框架锁定)
______________________________________________________________________
许可证
BSD-3条款许可-见 许可证 详情
______________________________________________________________________
贡献
欢迎投稿!拜托:
- 阅读
examples/README.md关于使用模式 - 遵循Oneiric模式(见示例)
- 分叉并创建特征分支
- 添加测试(覆盖率≥90%)
- 确保所有质量检查通过(
ruff format && ruff check && mypy && pytest) - 提交拉取请求
______________________________________________________________________
致谢
基于从9台生产MCP服务器中提取的模式构建:
主要模式来源:
- 能手 -MCP服务器结构、丰富的UI面板、CLI模式
- 会话伙伴 -配置模式、健康检查
- 快速挡块 -适配器组织、设置管理
其他贡献者:
- raindropio mcp(HTTP客户端模式)
- excalidraw mcp(测试模式)
- 歌剧云mcp
- mailgun mcp
- unifi-mcp
______________________________________________________________________
支持
如需支持,请查看 docs/ 目录或在存储库中创建问题。
______________________________________________________________________
准备好开始了吗? 结账 examples/ 用于演示所有功能的工作示例!
