use_database
Switch to a MySQL database by specifying its name and optional connection configuration, enabling direct database operations.
Instructions
切换到指定的数据库
Args:
database_name: 数据库名称
db_config: 数据库连接配置参数，如果为None则使用默认配置
Returns:
包含切换结果的字典Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database_name | Yes | ||
| db_config | No |
Implementation Reference
- mysql-mcp.py:458-511 (handler)The async function that implements the 'use_database' tool. It switches the active database by updating the database name in the connection config, testing the connection, and updating the global config.
async def use_database(database_name: str, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """切换到指定的数据库 Args: database_name: 数据库名称 db_config: 数据库连接配置参数，如果为None则使用默认配置 Returns: 包含切换结果的字典 """ global GLOBAL_DB_CONFIG if not database_name: return {"error": "数据库名称不能为空"} try: # 创建新的配置 if db_config is None: if GLOBAL_DB_CONFIG is not None: db_config = GLOBAL_DB_CONFIG.copy() else: db_config = DEFAULT_DB_CONFIG.copy() else: db_config = db_config.copy() # 更新数据库名称 db_config["database"] = database_name # 测试连接 conn = get_connection(db_config) cursor = conn.cursor() cursor.execute("SELECT DATABASE()") current_db = cursor.fetchone()[0] cursor.close() conn.close() # 更新全局配置 if GLOBAL_DB_CONFIG is not None: GLOBAL_DB_CONFIG["database"] = database_name return { "success": True, "message": f"已切换到数据库 {database_name}", "current_database": current_db } except Error as e: error_message = f"切换数据库失败: {str(e)}" if "Unknown database" in str(e): error_message += f"\n原因：数据库 {database_name} 不存在" elif "Access denied" in str(e): error_message += "\n原因：当前用户没有访问该数据库的权限" return {"error": error_message} except Exception as e: return {"error": f"切换数据库时发生未知错误: {str(e)}"} - mysql-mcp.py:457-458 (registration)The @mcp.tool() decorator that registers 'use_database' as an MCP tool.
@mcp.tool() async def use_database(database_name: str, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - mysql-mcp.py:458-458 (schema)Type signature defining input parameters: database_name (str) and optional db_config (Dict[str, Any]), returning a Dict[str, Any].
async def use_database(database_name: str, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - mysql-mcp.py:63-119 (helper)The get_connection helper function used by use_database to test connections to the specified database.
def get_connection(db_config=None): """获取数据库连接 Args: db_config: 数据库连接配置参数，如果为None则使用默认配置 Returns: 数据库连接对象 """ # 如果没有提供配置，先尝试使用全局配置，再使用默认配置 if db_config is None: if GLOBAL_DB_CONFIG is not None: db_config = GLOBAL_DB_CONFIG.copy() else: db_config = DEFAULT_DB_CONFIG.copy() else: # 合并用户提供的配置和全局/默认配置 if GLOBAL_DB_CONFIG is not None: config = GLOBAL_DB_CONFIG.copy() else: config = DEFAULT_DB_CONFIG.copy() config.update(db_config) db_config = config retry_count = 0 last_error = None max_retries = db_config.get("connect_retry_count", 3) while retry_count < max_retries: try: # 创建一个配置字典的副本，移除自定义的配置项 db_config_copy = db_config.copy() db_config_copy.pop("connect_retry_count", None) # 将connection_timeout转换为mysql.connector需要的connect_timeout参数 if "connection_timeout" in db_config_copy: db_config_copy["connect_timeout"] = db_config_copy.pop("connection_timeout") conn = mysql.connector.connect(**db_config_copy) return conn except Error as e: last_error = e retry_count += 1 if retry_count < max_retries: # 只有在还有重试机会的情况下打印重试信息 print(f"第 {retry_count} 次连接失败，正在重试... 错误: {e}") # 所有重试都失败后，构建详细的错误信息 error_message = f"数据库连接错误(重试 {retry_count} 次后): {last_error}" if "Can't connect to MySQL server" in str(last_error): error_message += f"\n无法连接到MySQL服务器，请检查主机 {db_config['host']} 和端口 {db_config['port']} 是否正确" error_message += f"\n连接超时时间为 {db_config.get('connection_timeout', 10)} 秒" elif "Access denied" in str(last_error): error_message += f"\n访问被拒绝，请检查用户名 {db_config['user']} 和密码是否正确" elif "Unknown database" in str(last_error): error_message += f"\n未知数据库 {db_config['database']}，请确认数据库名称是否正确" raise Exception(error_message) - mysql-mcp.py:24-58 (helper)The DEFAULT_DB_CONFIG dictionary used as a fallback configuration when no db_config is provided.
DEFAULT_DB_CONFIG = { "host": os.getenv("MYSQL_HOST", "localhost"), "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER", "root"), "password": os.getenv("MYSQL_PASSWORD", "root"), "database": os.getenv("MYSQL_DATABASE", ""), "connection_timeout": int(os.getenv("MYSQL_CONNECTION_TIMEOUT", "10")), # 连接超时时间(秒) "connect_retry_count": int(os.getenv("MYSQL_CONNECT_RETRY_COUNT", "3")) # 连接重试次数 } # 从命令行参数获取配置 def get_config_from_args(): args = parse_args() cmd_config = {} if args.host: cmd_config["host"] = args.host if args.port: cmd_config["port"] = args.port if args.user: cmd_config["user"] = args.user if args.password: cmd_config["password"] = args.password if args.database: cmd_config["database"] = args.database if args.connection_timeout: cmd_config["connection_timeout"] = args.connection_timeout if args.connect_retry_count: cmd_config["connect_retry_count"] = args.connect_retry_count # 合并配置 config = DEFAULT_DB_CONFIG.copy() config.update(cmd_config) return config