diff --git a/pipekit/drivers/base.py b/pipekit/drivers/base.py
index a03f880..2ab2e97 100644
--- a/pipekit/drivers/base.py
+++ b/pipekit/drivers/base.py
@@ -98,6 +98,14 @@ class Driver(abc.ABC):
     def list_tables(self, conn: dict, **qualifiers) -> list[RemoteTable]:
         """Fetch tables/views matching the qualifiers."""
 
+    def list_schemas(self, conn: dict, **qualifiers) -> list[str]:
+        """Return schema/library names available on the source.
+
+        Default returns []; drivers opt in by overriding. ``qualifiers`` lets
+        drivers like MSSQL scope the lookup to a database/linked-server.
+        """
+        return []
+
     @abc.abstractmethod
     def get_columns(self, conn: dict, table: str, **qualifiers) -> list[RemoteColumn]:
         """Fetch column metadata for one table."""
diff --git a/pipekit/drivers/db2.py b/pipekit/drivers/db2.py
index 59eb04a..6a135a0 100644
--- a/pipekit/drivers/db2.py
+++ b/pipekit/drivers/db2.py
@@ -44,6 +44,12 @@ class DB2Driver(Driver):
                         help="e.g. RLDBF12"),
         ]
 
+    def list_schemas(self, conn, **_) -> list[str]:
+        result = self.query(
+            conn, "SELECT SCHEMA_NAME FROM QSYS2.SYSSCHEMAS "
+                  "ORDER BY SCHEMA_NAME")
+        return [r[0].strip() for r in result.rows if r and r[0]]
+
     def list_tables(self, conn, *, schema: str) -> list[RemoteTable]:
         validate_identifier(schema, "schema")
         sql = (
diff --git a/pipekit/drivers/mssql.py b/pipekit/drivers/mssql.py
index c3c74a6..65a4584 100644
--- a/pipekit/drivers/mssql.py
+++ b/pipekit/drivers/mssql.py
@@ -49,6 +49,20 @@ class MSSQLDriver(Driver):
                         required=False, default="dbo"),
         ]
 
+    def list_schemas(
+        self, conn, *, linked_server: str | None = None,
+        database: str | None = None, **_,
+    ) -> list[str]:
+        self._validate(linked_server, database, None)
+        prefix = self._info_schema_prefix(linked_server, database)
+        sql = (
+            f"SELECT DISTINCT TABLE_SCHEMA FROM {prefix}INFORMATION_SCHEMA.TABLES "
+            f"WHERE TABLE_TYPE IN ('BASE TABLE','VIEW') "
+            f"ORDER BY TABLE_SCHEMA"
+        )
+        result = self.query(conn, sql)
+        return [r[0].strip() for r in result.rows if r and r[0]]
+
     def list_tables(
         self, conn, *, linked_server: str | None = None,
         database: str | None = None, schema: str | None = None,
diff --git a/pipekit/drivers/pg.py b/pipekit/drivers/pg.py
index e159953..5145f2a 100644
--- a/pipekit/drivers/pg.py
+++ b/pipekit/drivers/pg.py
@@ -41,6 +41,15 @@ class PGDriver(Driver):
                         required=False, default="public"),
         ]
 
+    def list_schemas(self, conn, **_) -> list[str]:
+        result = self.query(
+            conn,
+            "SELECT schema_name FROM information_schema.schemata "
+            "WHERE schema_name NOT IN ('pg_catalog','information_schema') "
+            "AND schema_name NOT LIKE 'pg\\_%' ESCAPE '\\' "
+            "ORDER BY schema_name")
+        return [r[0].strip() for r in result.rows if r and r[0]]
+
     def list_tables(self, conn, *, schema: str | None = None) -> list[RemoteTable]:
         if schema:
             validate_identifier(schema, "schema")