10.5 框架核心功能模块
“agent_mcp/core”目录是本项目的核心功能模块，实现了项目关键基础功能的实现与管理。负责处理代理身份验证，确保系统访问安全；进行项目配置管理，集中管控各类配置参数；还涉及全局层面的相关操作与支持，为整个项目的稳定运行和核心功能交互提供底层保障，是项目核心逻辑与基础服务的重要集合地。
10.5.1 系统配置
文件agent_mcp/core/config.py是本项目的核心配置文件，负责系统多方面配置管理。例如下面代码的功能是配置应用程序的全局日志系统。原理是先获取根日志记录器并设置日志级别，清除已有的处理器避免重复，然后根据调试模式等条件，分别添加文件处理器（调试模式下启用，用于将日志写入文件）和控制台处理器（根据配置决定是否启用，带有彩色格式化以增强可读性），同时还会对特
def setup_logging(): """配置应用程序的全局日志记录。""" root_logger = logging.getLogger() # 获取根日志记录器 root_logger.setLevel(LOG_LEVEL) # 在根日志记录器上设置级别 # 清除根日志记录器上任何已存在的处理器，以避免重复 for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) # 1. 文件处理器（仅在调试模式下） debug_mode = os.environ.get("MCP_DEBUG", "false").lower() == "true" if debug_mode: file_formatter = logging.Formatter(LOG_FORMAT_FILE) file_handler = logging.FileHandler( LOG_FILE_NAME, mode="a", encoding="utf-8" ) # 追加模式 file_handler.setFormatter(file_formatter) root_logger.addHandler(file_handler) # 2. 控制台处理器（带颜色，有条件启用） if CONSOLE_LOGGING_ENABLED: console_formatter = ColorfulFormatter( LOG_FORMAT_CONSOLE, datefmt="%H:%M:%S" ) # 控制台使用更简洁的日期格式 console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(console_formatter) # 如果需要，可过滤控制台中不太重要的消息 # console_handler.setLevel(logging.INFO) # 示例：控制台仅显示INFO及以上级别 root_logger.addHandler(console_handler) # 抑制特定库的过度冗长日志，同时作用于文件和控制台 logging.getLogger("watchfiles").setLevel(logging.WARNING) # Uvicorn访问日志由Uvicorn的配置处理（cli.py中access_log=False） # 但如果需要，我们也可以尝试管理其错误日志记录器 logging.getLogger("uvicorn.error").setLevel(logging.WARNING) logging.getLogger("uvicorn").setLevel(logging.WARNING) # 通用uvicorn日志记录器 logging.getLogger("mcp.server.lowlevel.server").propagate = ( False # 防止如果它直接记录时出现重复 )10.5.2 身份验证
文件agent_mcp/core/auth.py实现了本项目的身份验证功能，包括生成安全随机令牌、验证令牌是否有效且具备所需角色，以及根据令牌获取对应的代理ID，通过操作全局变量来管理和验证系统中代理与管理员的身份权限。
def generate_token() -> str: """生成一个安全的随机令牌。""" return secrets.token_hex(16) def verify_token(token: str, required_role: str = "agent") -> bool: """ 验证令牌是否有效且具备所需角色。 使用全局变量 `g.admin_token` 和 `g.active_agents`。 """ if not token: # 增加对空/None令牌的检查 return False if required_role == "admin" and token == g.admin_token: return True # 仅当active_agents不为None且令牌是其中的键时，检查活跃代理 if required_role == "agent" and g.active_agents and token in g.active_agents: return True # 也允许管理员令牌用于代理角色 if required_role == "agent" and token == g.admin_token: return True # 管理员可以充当代理 return False def get_agent_id(token: str) -> Optional[str]: """ 从令牌获取代理ID。 使用全局变量 `g.admin_token` 和 `g.active_agents`。 """ if not token: # 增加对空/None令牌的检查 return None if token == g.admin_token: return "admin" # “admin”是用于管理员操作的特殊代理ID # 仅当active_agents不为None且令牌是其中的键时，检查活跃代理 if g.active_agents and token in g.active_agents: # 确保代理数据字典包含“agent_id” agent_data = g.active_agents[token] if isinstance(agent_data, dict) and "agent_id" in agent_data: return agent_data["agent_id"] return None
