"""配置管理模块 - 支持通过工具进行配置，无需手动编辑文件."""
import json
import os
from pathlib import Path
from typing import Optional, Dict, Any
from pydantic import BaseModel, Field
class Neo4jConfig(BaseModel):
"""Neo4j 数据库配置."""
uri: str = Field(..., description="Neo4j 连接 URI，例如: bolt://localhost:7687")
username: str = Field(..., description="Neo4j 用户名")
password: str = Field(..., description="Neo4j 密码")
database: str = Field(default="neo4j", description="数据库名称")
class APIConfig(BaseModel):
"""AI API 配置（可选）."""
provider: str = Field(default="openai", description="API 提供商: openai, anthropic, etc.")
api_key: Optional[str] = Field(default=None, description="API Key（可选，如果不提供将使用 Cursor 内置 AI）")
base_url: Optional[str] = Field(default=None, description="API Base URL（可选，用于 DeepSeek 等兼容 OpenAI 接口的模型）")
model: Optional[str] = Field(default=None, description="模型名称")
class ServerConfig(BaseModel):
"""服务器完整配置."""
neo4j: Optional[Neo4jConfig] = None
api: Optional[APIConfig] = None
group_id: str = Field(default="default", description="用于数据隔离的组 ID")
class ConfigManager:
"""配置管理器 - 支持通过工具进行配置."""
def __init__(self, config_path: Optional[Path] = None):
"""
初始化配置管理器.
Args:
config_path: 配置文件路径，默认为 ~/.grace-mcp/config.json
"""
if config_path is None:
config_dir = Path.home() / ".grace-mcp"
config_dir.mkdir(parents=True, exist_ok=True)
config_path = config_dir / "config.json"
self.config_path = config_path
self._config: Optional[ServerConfig] = None
self._last_mtime: float = 0.0
def load_config(self) -> ServerConfig:
"""加载配置 (支持热重载)."""
if self.config_path.exists():
try:
current_mtime = self.config_path.stat().st_mtime
# 如果文件修改时间变了，或者从未加载过，则重新加载
if self._config is None or current_mtime > self._last_mtime:
with open(self.config_path, "r", encoding="utf-8") as f:
data = json.load(f)
self._config = ServerConfig(**data)
self._last_mtime = current_mtime
except Exception as e:
# 如果配置文件损坏，保留旧配置或创建新配置
import sys
print(f"警告: 配置文件加载失败: {e}", file=sys.stderr)
if self._config is None:
self._config = ServerConfig()
else:
# 文件不存在，使用默认配置
if self._config is None:
self._config = ServerConfig()
return self._config
def save_config(self) -> bool:
"""保存配置到文件."""
if self._config is None:
return False
try:
# 确保目录存在
self.config_path.parent.mkdir(parents=True, exist_ok=True)
# 保存配置（密码等敏感信息需要特殊处理）
config_dict = self._config.model_dump(mode="json")
with open(self.config_path, "w", encoding="utf-8") as f:
json.dump(config_dict, f, indent=2, ensure_ascii=False)
# 更新 mtime，避免下次 load_config 重新加载
self._last_mtime = self.config_path.stat().st_mtime
return True
except Exception as e:
# 使用 stderr 输出错误（不会干扰 MCP 协议）
import sys
print(f"错误: 保存配置失败: {e}", file=sys.stderr)
return False
def configure_neo4j(
self,
uri: str,
username: str,
password: str,
database: str = "neo4j"
) -> Dict[str, Any]:
"""
配置 Neo4j 连接.
Args:
uri: Neo4j URI
username: 用户名
password: 密码
database: 数据库名称
Returns:
配置结果
"""
config = self.load_config()
config.neo4j = Neo4jConfig(
uri=uri,
username=username,
password=password,
database=database
)
self._config = config
if self.save_config():
return {
"success": True,
"message": "Neo4j 配置已保存",
"config": {
"uri": uri,
"username": username,
"database": database
}
}
else:
return {
"success": False,
"message": "Neo4j 配置保存失败"
}
def configure_api(
self,
provider: str = "openai",
api_key: Optional[str] = None,
base_url: Optional[str] = None,
model: Optional[str] = None
) -> Dict[str, Any]:
"""
配置 AI API（可选）.
Args:
provider: API 提供商
api_key: API Key（可选）
base_url: API Base URL（可选）
model: 模型名称（可选）
Returns:
配置结果
"""
config = self.load_config()
config.api = APIConfig(
provider=provider,
api_key=api_key,
base_url=base_url,
model=model
)
self._config = config
if self.save_config():
return {
"success": True,
"message": "API 配置已保存",
"config": {
"provider": provider,
"has_api_key": api_key is not None,
"base_url": base_url,
"model": model
}
}
else:
return {
"success": False,
"message": "API 配置保存失败"
}
def set_group_id(self, group_id: str) -> Dict[str, Any]:
"""
设置组 ID（用于数据隔离）.
Args:
group_id: 组 ID
Returns:
配置结果
"""
config = self.load_config()
config.group_id = group_id
self._config = config
if self.save_config():
return {
"success": True,
"message": f"组 ID 已设置为: {group_id}",
"group_id": group_id
}
else:
return {
"success": False,
"message": "组 ID 设置失败"
}
def get_config_status(self) -> Dict[str, Any]:
"""
获取当前配置状态.
Returns:
配置状态信息
"""
config = self.load_config()
status = {
"neo4j_configured": config.neo4j is not None,
"api_configured": config.api is not None and config.api.api_key is not None,
"group_id": config.group_id,
"config_path": str(self.config_path)
}
if config.neo4j:
status["neo4j"] = {
"uri": config.neo4j.uri,
"username": config.neo4j.username,
"database": config.neo4j.database
}
if config.api:
status["api"] = {
"provider": config.api.provider,
"has_api_key": config.api.api_key is not None,
"base_url": config.api.base_url,
"model": config.api.model
}
return status
def reset_config(self) -> Dict[str, Any]:
"""
重置所有配置.
Returns:
重置结果
"""
self._config = ServerConfig()
if self.save_config():
return {
"success": True,
"message": "配置已重置"
}
else:
return {
"success": False,
"message": "配置重置失败"
}
def get_neo4j_config(self) -> Optional[Neo4jConfig]:
"""获取 Neo4j 配置."""
config = self.load_config()
return config.neo4j
def get_api_config(self) -> Optional[APIConfig]:
"""获取 API 配置."""
config = self.load_config()
return config.api
def get_group_id(self) -> str:
"""获取组 ID."""
config = self.load_config()
return config.group_id
