瑞士AI MCP Commons
瑞士人工智能模型上下文协议(MCP)服务器的共享基础设施和数据模型。
版本: 1.0.0 许可证: 麻省理工学院 状态: 生产就绪
概述
swiss-ai-mcp-commons 是一个Python包,为瑞士旅行和旅游MCP提供标准化组件:
- 数据模型:位置、天气、定价和时间的标准化Pydantic模型
- HTTP客户端:异步HTTP客户端,具有缓存、重试和结构化日志记录功能
- 验证:日期、货币、瑞士地区等的输入验证实用程序
- 错误处理:带有JSON-RPC错误代码的标准异常层次结构
- 日志记录:支持JSON输出的结构化日志配置
特性
标准化数据模型
from swiss_ai_mcp_commons import Location, Coordinates, Weather, Price
# Geographic location
location = Location(
name="Bern",
coordinates=Coordinates(latitude=46.947, longitude=7.447),
region=Region(canton="BE"),
)
# Weather data
weather = Weather(
timestamp=datetime.now(),
description="Sunny",
temperature=Temperature(value=15.2, min=8.5, max=19.3),
humidity_percent=65,
)
# Pricing information
price = Price(amount=125.50, currency="CHF")带缓存的HTTP客户端
from swiss_ai_mcp_commons import CachedHttpClient
async with CachedHttpClient(
base_url="https://api.example.com",
cache_ttl_seconds=120,
) as client:
# Automatically cached for 120 seconds
data = await client.get("/v1/endpoint", params={"key": "value"})
# Retries with exponential backoff on server errors
data = await client.post("/v1/create", json={"data": "value"})输入验证
from swiss_ai_mcp_commons import (
validate_date_range,
validate_currency_code,
validate_swiss_canton,
)
# Date range validation
start = date(2024, 7, 15)
end = date(2024, 7, 22)
validate_date_range(start, end, min_days=1, max_days=365)
# Currency code validation
currency = validate_currency_code("CHF") # "CHF"
# Swiss canton validation
canton = validate_swiss_canton("be") # "BE"结构化日志记录
from swiss_ai_mcp_commons import configure_logging, get_logger
# Configure logging
configure_logging(
app_name="my-app",
version="1.0.0",
json_output=True,
)
# Get logger
logger = get_logger(__name__)
logger.info("event_name", key="value", number=42)
# Output: {"timestamp": "2024-01-20T14:30:00Z", "level": "info", "event": "event_name", "key": "value", "number": 42}错误处理
from swiss_ai_mcp_commons import (
StandardMcpException,
ValidationError,
ApiError,
ConfigurationError,
)
# Validation error
try:
validate_date_range(invalid_start, invalid_end)
except ValidationError as e:
error_dict = e.to_dict() # {"code": -32001, "message": "...", "data": {...}}
# API error
try:
await client.get("/endpoint")
except ApiError as e:
print(f"API {e.details['api']} failed: {e.message}")
# Configuration error
if not os.environ.get("API_KEY"):
raise ConfigurationError(
"Missing required API key",
config_key="API_KEY"
)安装
来自PyPI(发布时)
pip install swiss-ai-mcp-commons从地方发展
# Clone repository
git clone https://github.com/your-org/swiss-ai-mcp-commons.git
cd swiss-ai-mcp-commons
# Install with development dependencies
pip install -e ".[dev]"
# Or using uv
uv syncMCP中的使用
示例:天气MCP
from fastmcp import FastMCP
from swiss_ai_mcp_commons import (
configure_logging,
CachedHttpClient,
Weather,
Location,
)
configure_logging(app_name="open-meteo-mcp", version="2.1.0")
mcp = FastMCP("Open-Meteo Weather")
@mcp.tool()
async def get_weather(location: Location) -> Weather:
"""Get current weather for a location."""
async with CachedHttpClient(base_url="https://api.open-meteo.com") as client:
data = await client.get(
"/v1/forecast",
params={
"latitude": location.coordinates.latitude,
"longitude": location.coordinates.longitude,
}
)
return Weather(
timestamp=datetime.now(),
description=data["current"]["weather"],
temperature=Temperature(value=data["current"]["temperature"]),
)示例:旅程MCP
from swiss_ai_mcp_commons import (
validate_date_range,
DateRange,
StandardMcpException,
)
@mcp.tool()
def plan_journey(start_date: str, end_date: str) -> dict:
"""Plan a journey with date validation."""
try:
start = date.fromisoformat(start_date)
end = date.fromisoformat(end_date)
validate_date_range(start, end, min_days=1, max_days=365)
date_range = DateRange(start_date=start, end_date=end)
# ... rest of journey planning logic
return {"duration_days": date_range.days}
except ValidationError as e:
raise StandardMcpException(e.message, code=e.code)数据模型
区位模型
- 坐标:经过验证的地理坐标(-90至90纬度,-180至180经度)
- 区域:瑞士州和地区信息
- 位置:包含坐标、区域和元数据的完整位置
天气模型
- 温度:最低/最高温度和表观温度
- 雪况:雪深、新雪、雪崩风险、质量
- 空气质量:含污染物水平和花粉数据的空气质量指数
- 天气:以上综合天气加上风、湿度、降水
定价模型
- 价格:单一货币价格
- 票价选项:票价等级、限制、行李、折扣
- 价格信息:标准价与折扣价,含票价选项
时间模型
- 时间范围:验证的时间范围
- 日期范围:带属性的日期范围(天、is_past、is_future、is_current)
验证实用程序
validate_date_range():使用最小/最大天数验证日期范围validate_currency_code():ISO 4217货币代码验证validate_swiss_canton():瑞士州代码验证(2个字母的代码)validate_email():电子邮件地址验证validate_phone():电话号码验证(瑞士格式)validate_price():价格金额验证,最小/最大范围
异常层次结构
所有异常都继承自 StandardMcpException JSON-RPC错误代码:
- 验证错误 (-32001):输入验证失败
- 接口错误 (-32006):外部API调用失败
- 配置错误 (-32009):配置问题
- 身份验证错误 (-32002):身份验证失败
- RateLimitError (-32007):超出费率限制
- 超时错误 (-32008):操作超时
测试
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=swiss_ai_mcp_commons
# Run specific test file
uv run pytest tests/test_models.py
# Run with verbose output
uv run pytest -v代码质量
# Format code
uv run black src/ tests/
# Lint
uv run ruff check src/ tests/
# Type checking
uv run mypy src/
# All checks
uv run black src/ tests/ && uv run ruff check src/ tests/ && uv run mypy src/建筑
项目结构
swiss-ai-mcp-commons/
├── src/swiss_ai_mcp_commons/
│ ├── __init__.py # Main package exports
│ ├── models/ # Data models (Pydantic)
│ │ ├── __init__.py
│ │ ├── location.py # Location, Coordinates, Region
│ │ ├── weather.py # Weather, Temperature, Snow, AirQuality
│ │ ├── pricing.py # Price, FareOption, PricingInfo
│ │ └── time.py # TimeRange, DateRange
│ ├── http/ # HTTP client
│ │ ├── __init__.py
│ │ └── client.py # CachedHttpClient with retries
│ ├── logging/ # Structured logging
│ │ ├── __init__.py
│ │ └── setup.py # Logging configuration
│ └── validation/ # Validation & exceptions
│ ├── __init__.py
│ ├── validators.py # Input validators
│ └── exceptions.py # Exception hierarchy
├── tests/ # Test suite
│ ├── conftest.py # Shared fixtures
│ ├── test_models.py # Model tests
│ └── test_validation.py # Validator & exception tests
├── pyproject.toml # Project configuration
└── README.md # This file设计原则
- 标准化:所有MCP使用的通用模型可减少重复
- 验证优先:输入验证发生在模型边界
- 错误清晰度:JSON-RPC错误代码支持清晰的错误处理
- 异步本机:HTTP客户端使用async/await实现可扩展性
- 可观察对象:结构化日志记录支持生产调试
- 类型安全:带类型提示的完整Pydantic验证
演出
- 缓存:缓存120秒的HTTP响应(可配置)
- 重试:瞬态错误的自动指数回退
- 异步:并发请求的非阻塞I/O
- 最小依赖性:只有Pydantic、httpx、structlog
贡献
瑞士人工智能mcp公地贡献指南:
- 保持向后兼容性(semver)
- 为新功能添加测试
- 保持模型的专注性和可组合性
- 记录复杂验证器
- 使用一致的命名(函数使用snake_case,类使用PascalCase)
许可证
MIT许可证-有关详细信息,请参阅许可证文件
支持
对于问题和疑问:
______________________________________________________________________
内置于❤️ 瑞士人工智能MCP生态系统
