Show database version in db.list and sql.query tool description

Adds SQLExecutor.get_version() using SELECT VERSION() / sqlite_version().
db.list now includes a version field per connection. sql.query description
shows the exact engine and version (e.g. "MySQL 5.7.41") so AI agents
can choose the correct SQL dialect.

Author: loglux

app/config.py

@@ -150,6 +150,15 @@ def db_type(self) -> str:
             return "SQLite"
         return "SQL"
 
+    @property
+    def db_version(self) -> str:
+        from app.sql.executor import SQLExecutor
+
+        try:
+            return SQLExecutor(self.db_url).get_version()
+        except Exception:
+            return ""
+
     @classmethod
     def from_env(cls) -> "Config":
         return cls.load()

app/mcp/tools.py

@@ -490,7 +490,8 @@ def db_use(payload: Dict[str, Any]) -> Dict[str, Any]:
                 name="sql.query",
                 title="SQL Query",
                 description=(
-                    f"Execute a read-only SQL query against {config.db_type}. "
+                    f"Execute a read-only SQL query against {config.db_type}"
+                    f"{' ' + config.db_version if config.db_version else ''}. "
                     f"Use {config.db_type}-compatible syntax. "
                     "SELECT, WITH, EXPLAIN, SHOW, and DESCRIBE statements are allowed. "
                     "For INSERT, UPDATE, DELETE, or DDL statements use db.apply."
@@ -588,6 +589,7 @@ def db_use(payload: Dict[str, Any]) -> Dict[str, Any]:
                                     "id": {"type": "integer"},
                                     "name": {"type": "string"},
                                     "db_type": {"type": "string"},
+                                    "version": {"type": "string"},
                                     "mode": {"type": "string"},
                                     "host": {"type": "string"},
                                     "is_active": {"type": "boolean"},

app/session_db.py

@@ -58,6 +58,16 @@ def get_engine_for_session(self, session_id: str | None) -> Engine:
         url = self.get_db_url(session_id)
         return self._get_engine(url)
 
+    def get_db_version(self, session_id: str | None) -> str:
+        """Return the DB version string for a session's active connection."""
+        from app.sql.executor import SQLExecutor
+
+        url = self.get_db_url(session_id)
+        try:
+            return SQLExecutor(url).get_version()
+        except Exception:
+            return ""
+
     def get_db_type(self, session_id: str | None) -> str:
         """Return the DB type name for a session's active connection."""
         url = self.get_db_url(session_id).lower()
@@ -78,15 +88,25 @@ def get_mode(self, session_id: str | None, default_mode: str) -> str:
 
     def list_connections(self) -> list[dict[str, Any]]:
         """Return all registered connections with metadata (no secrets)."""
+        from app.sql.executor import SQLExecutor
+
         connections = settings_db.get_all_db_connections()
         result = []
         for c in connections:
             url = c.get("url", "")
+            version = ""
+            if url:
+                try:
+                    ex = SQLExecutor(url)
+                    version = ex.get_version()
+                except Exception:
+                    pass
             result.append(
                 {
                     "id": c["id"],
                     "name": c.get("name", ""),
                     "db_type": c.get("db_type", ""),
+                    "version": version,
                     "mode": c.get("mode", "read-only"),
                     "is_active": bool(c.get("is_active")),
                     "host": _extract_host(url),

app/sql/executor.py

@@ -25,6 +25,21 @@ class SQLExecutor:
     def __init__(self, db_url: str) -> None:
         self.engine: Engine = create_engine(db_url)
 
+    def get_version(self) -> str:
+        """Return the database server version string."""
+        dialect = self.engine.dialect.name
+        try:
+            with self.engine.connect() as conn:
+                if dialect == "sqlite":
+                    row = conn.execute(text("SELECT sqlite_version()")).fetchone()
+                else:
+                    row = conn.execute(text("SELECT VERSION()")).fetchone()
+                if row:
+                    return str(row[0])
+        except SQLAlchemyError:
+            pass
+        return ""
+
     def execute(
         self,
         sql: str,