#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
配置管理模块 / Configuration Management Module
处理配置文件的加载、验证和管理 / Handle configuration file loading, validation and management
"""
import json
import os
from typing import Dict, Any
# 导入语言资源
from src.utils.language_resources import get_text
def load_config(config_path: str = "config.json") -> Dict[str, Any]:
"""加载配置文件 / Load configuration file"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
# 验证和补充配置
return validate_and_complete_config(config)
except Exception as e:
print(get_text("config_load_error", str(e)))
# 返回默认配置
return get_default_config()
def validate_and_complete_config(config: Dict[str, Any]) -> Dict[str, Any]:
"""验证和补充配置 / Validate and complete configuration"""
# 确保必需字段存在 / Ensure required fields exist
if "agent_name" not in config:
config["agent_name"] = get_text("agent_name")
if "version" not in config:
config["version"] = "1.0.0"
if "description" not in config:
config["description"] = get_text("agent_description")
# 确保工具目录配置 / Ensure tool directory configuration
if "tools_directory" not in config:
config["tools_directory"] = "./src/tools"
if "utils_directory" not in config:
config["utils_directory"] = "./src/utils"
# 确保日志级别配置 / Ensure log level configuration
if "log_level" not in config:
config["log_level"] = "INFO"
# 确保语言配置 / Ensure language configuration
if "language" not in config:
config["language"] = {
"default": "zh",
"available": ["zh", "en"],
"auto_detect": True
}
# 确保OpenTelemetry配置 / Ensure OpenTelemetry configuration
if "opentelemetry" not in config:
config["opentelemetry"] = get_default_opentelemetry_config()
# 确保服务器配置 / Ensure server configuration
if "port" not in config:
config["port"] = 8001
if "host" not in config:
config["host"] = "localhost"
# 确保LLM配置 / Ensure LLM configuration
if "llm" not in config:
config["llm"] = {
"enabled": False,
"providers": {
"openai": {
"api_key": "",
"model": "gpt-3.5-turbo"
},
"anthropic": {
"api_key": "",
"model": "claude-3-sonnet-20240229"
}
}
}
return config
def get_default_config() -> Dict[str, Any]:
"""获取默认配置 / Get default configuration"""
return {
"agent_name": get_text("agent_name"),
"version": "1.0.0",
"description": get_text("agent_description"),
"tools_directory": "./src/tools",
"utils_directory": "./src/utils",
"log_level": "INFO",
"language": {
"default": "zh",
"available": ["zh", "en"],
"auto_detect": True
},
"opentelemetry": get_default_opentelemetry_config(),
"port": 8001,
"host": "localhost",
"llm": {
"enabled": False,
"providers": {
"openai": {
"api_key": "",
"model": "gpt-3.5-turbo"
},
"anthropic": {
"api_key": "",
"model": "claude-3-sonnet-20240229"
}
}
}
}
def get_default_opentelemetry_config() -> Dict[str, Any]:
"""获取默认的OpenTelemetry配置 / Get default OpenTelemetry configuration"""
return {
"enabled": False,
"service_name": get_text("agent_name"),
"exporter": "console", # console, otlp / 控制台, OTLP
"otlp_endpoint": "http://localhost:4318/v1/traces",
"sampler": "always_on",
"headers": {}, # OTLP导出器的自定义头部 / Custom headers for OTLP exporter
"api_key": "", # LangSmith等平台的API密钥 / API key for platforms like LangSmith
"project_name": "zephyr_mcp_agent" # LangSmith项目名 / LangSmith project name
}
def save_config(config: Dict[str, Any], config_path: str = "config.json") -> bool:
"""保存配置到文件 / Save configuration to file"""
try:
with open(config_path, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
return True
except Exception as e:
print(get_text("config_save_error", str(e)))
return False
def get_config_value(config: Dict[str, Any], key_path: str, default: Any = None) -> Any:
"""通过路径获取配置值 / Get configuration value by path"""
keys = key_path.split('.')
current = config
# 遍历路径键 / Traverse path keys
for key in keys:
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current
def set_config_value(config: Dict[str, Any], key_path: str, value: Any) -> bool:
"""通过路径设置配置值 / Set configuration value by path"""
keys = key_path.split('.')
current = config
# 遍历到最后一个键的父级 / Traverse to the parent of the last key
for key in keys[:-1]:
if key not in current or not isinstance(current[key], dict):
current[key] = {}
current = current[key]
# 设置最终值 / Set the final value
current[keys[-1]] = value
return True
def validate_config(config: Dict[str, Any]) -> Dict[str, Any]:
"""验证配置的有效性 / Validate configuration validity"""
errors = []
# 验证必需字段 / Validate required fields
if not config.get("agent_name"):
errors.append("agent_name is required")
if not config.get("version"):
errors.append("version is required")
# 验证端口范围 / Validate port range
port = config.get("port", 8001)
if not isinstance(port, int) or port < 1 or port > 65535:
errors.append("port must be an integer between 1 and 65535")
# 验证日志级别 / Validate log level
valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
if config.get("log_level") not in valid_log_levels:
errors.append(f"log_level must be one of {valid_log_levels}")
# 验证语言配置 / Validate language configuration
language_config = config.get("language", {})
if language_config.get("default") not in language_config.get("available", []):
errors.append("default language must be in available languages")
# 验证OpenTelemetry配置 / Validate OpenTelemetry configuration
otel_config = config.get("opentelemetry", {})
if otel_config.get("enabled", False):
if otel_config.get("exporter") not in ["console", "otlp"]:
errors.append("opentelemetry.exporter must be 'console' or 'otlp'")
return {
"valid": len(errors) == 0,
"errors": errors
}
def create_sample_config(config_path: str = "config.json") -> bool:
"""创建示例配置文件 / Create sample configuration file"""
sample_config = {
"agent_name": "Zephyr MCP Agent",
"version": "1.0.0",
"description": "Zephyr MCP Agent for Zephyr RTOS development",
"tools_directory": "./src/tools",
"utils_directory": "./src/utils",
"log_level": "INFO",
"port": 8001,
"host": "localhost",
"language": {
"default": "zh",
"available": ["zh", "en"],
"auto_detect": True
},
"opentelemetry": {
"enabled": False,
"service_name": "zephyr_mcp_agent",
"exporter": "console",
"otlp_endpoint": "http://localhost:4318/v1/traces",
"sampler": "always_on"
},
"llm": {
"enabled": False,
"providers": {
"openai": {
"api_key": "your_openai_api_key_here",
"model": "gpt-3.5-turbo"
},
"anthropic": {
"api_key": "your_anthropic_api_key_here",
"model": "claude-3-sonnet-20240229"
}
}
}
}
return save_config(sample_config, config_path)
