list_tables
List all tables in a MySQL database. Specify the database name and optional connection config to retrieve a dictionary of table names.
Instructions
列出指定数据库中的所有表
Args:
database_name: 数据库名称，如果为None则使用默认数据库
db_config: 数据库连接配置参数，如果为None则使用默认配置
Returns:
包含表列表的字典Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database_name | No | ||
| db_config | No |
Implementation Reference
- mysql-mcp.py:177-218 (handler)The 'list_tables' tool handler: an async function decorated with @mcp.tool() that connects to MySQL, runs SHOW TABLES (optionally for a specific database via SHOW TABLES FROM), and returns the list of tables with success status, database name, and count.
@mcp.tool() async def list_tables(database_name: Optional[str] = None, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """列出指定数据库中的所有表 Args: database_name: 数据库名称，如果为None则使用默认数据库 db_config: 数据库连接配置参数，如果为None则使用默认配置 Returns: 包含表列表的字典 """ try: conn = get_connection(db_config) cursor = conn.cursor() # 如果指定了数据库名称，查询指定数据库的表 if database_name: cursor.execute(f"SHOW TABLES FROM {database_name}") else: cursor.execute("SHOW TABLES") tables = [table[0] for table in cursor.fetchall()] return { "success": True, "database": database_name or conn.database, "tables": tables, "count": len(tables) } except Error as e: error_message = f"获取表列表失败: {str(e)}" if "Access denied" in str(e): error_message += "\n原因：当前用户没有足够权限执行SHOW TABLES命令" elif "Unknown database" in str(e): error_message += f"\n原因：数据库 {database_name} 不存在" return {"error": error_message} except Exception as e: return {"error": f"获取表列表时发生未知错误: {str(e)}"} finally: if 'conn' in locals() and conn.is_connected(): cursor.close() conn.close() - mysql-mcp.py:177-177 (registration)Registration of the 'list_tables' tool via @mcp.tool() decorator on top of the async function.
@mcp.tool() - mysql-mcp.py:63-119 (helper)The get_connection helper function used by list_tables to establish a MySQL database connection with retry logic.
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)