diff --git a/scripts/query_mysql.py b/scripts/query_mysql.py
new file mode 100644
index 0000000..157939f
--- /dev/null
+++ b/scripts/query_mysql.py
@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+通用 MySQL 查询脚本
+用法：
+  python scripts/query_mysql.py "SELECT * FROM diagnosis_session ORDER BY created_at DESC LIMIT 5"
+  python scripts/query_mysql.py                      # 交互模式
+依赖：pip install pymysql
+"""
+
+import sys
+import os
+
+# Windows 控制台 UTF-8 输出
+if sys.stdout.encoding and sys.stdout.encoding.lower() != 'utf-8':
+    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
+
+try:
+    import pymysql
+    import pymysql.cursors
+except ImportError:
+    print("缺少依赖，请先执行: pip install pymysql")
+    sys.exit(1)
+
+# 从 application.yml 读取的连接信息
+DB_CONFIG = {
+    "host": "119.29.78.52",
+    "port": 33306,
+    "user": "root",
+    "password": "!Fucker123..",
+    "database": "superbiz_agent",
+    "charset": "utf8mb4",
+    "cursorclass": pymysql.cursors.DictCursor,
+}
+
+
+def run_query(sql: str):
+    conn = pymysql.connect(**DB_CONFIG)
+    try:
+        with conn.cursor() as cur:
+            cur.execute(sql)
+            if sql.strip().upper().startswith("SELECT") or sql.strip().upper().startswith("SHOW"):
+                rows = cur.fetchall()
+                if not rows:
+                    print("(空结果)")
+                    return
+                # 打印列头
+                cols = list(rows[0].keys())
+                col_widths = {c: max(len(c), max(len(str(r[c])) for r in rows)) for c in cols}
+                header = " | ".join(c.ljust(col_widths[c]) for c in cols)
+                print(header)
+                print("-" * len(header))
+                for row in rows:
+                    print(" | ".join(str(row[c]).ljust(col_widths[c]) for c in cols))
+                print(f"\n({len(rows)} 行)")
+            else:
+                conn.commit()
+                print(f"OK，影响行数: {cur.rowcount}")
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1:
+        sql = " ".join(sys.argv[1:])
+        run_query(sql)
+    else:
+        print("MySQL 交互模式（输入 exit 退出）")
+        print(f"连接：{DB_CONFIG['user']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}")
+        print("-" * 50)
+        while True:
+            try:
+                sql = input("sql> ").strip()
+                if sql.lower() in ("exit", "quit", "q"):
+                    break
+                if not sql:
+                    continue
+                run_query(sql)
+            except KeyboardInterrupt:
+                break
+            except Exception as e:
+                print(f"错误: {e}")