From 71527017e579ae71a15f5412747a01a34f960b07 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 27 May 2026 11:56:17 +0200
Subject: [PATCH 1/4] fix workspace login type for snowflake

---
 src/keboola_agent_cli/client.py               |  8 +-
 src/keboola_agent_cli/constants.py            |  3 +
 .../services/workspace_service.py             | 11 ++-
 tests/test_client.py                          | 65 ++++++++++++++
 tests/test_workspace_service.py               | 88 ++++++++++++++++++-
 5 files changed, 172 insertions(+), 3 deletions(-)

diff --git a/src/keboola_agent_cli/client.py b/src/keboola_agent_cli/client.py
index 5d73813d..bde607b4 100644
--- a/src/keboola_agent_cli/client.py
+++ b/src/keboola_agent_cli/client.py
@@ -2531,6 +2531,7 @@ def create_config_workspace(
         component_id: str,
         config_id: str,
         backend: str = "snowflake",
+        login_type: str | None = None,
     ) -> dict[str, Any]:
         """Create a workspace tied to a specific configuration.
 
@@ -2539,16 +2540,21 @@ def create_config_workspace(
             component_id: Component ID (e.g. keboola.snowflake-transformation).
             config_id: Configuration ID.
             backend: Workspace backend.
+            login_type: Optional Storage API loginType. Omitted when None.
 
         Returns:
             Workspace dict including connection credentials.
         """
         safe_component = quote(component_id, safe="")
         safe_config = quote(config_id, safe="")
+        payload: dict[str, Any] = {"backend": backend}
+        if login_type is not None:
+            payload["loginType"] = login_type
+
         response = self._request(
             "POST",
             f"/v2/storage/branch/{branch_id}/components/{safe_component}/configs/{safe_config}/workspaces",
-            json={"backend": backend},
+            json=payload,
         )
         return response.json()
 
diff --git a/src/keboola_agent_cli/constants.py b/src/keboola_agent_cli/constants.py
index f440cac6..e516cbfe 100644
--- a/src/keboola_agent_cli/constants.py
+++ b/src/keboola_agent_cli/constants.py
@@ -303,14 +303,17 @@
 # Verified 2026-05-18 against project 901 on connection.keboola.com:
 #   snowflake-service-keypair: PASS
 #   snowflake-person-sso:      PASS
+#   snowflake-person-keypair:  PASS (required for new Snowflake sandboxes)
 #   snowflake-legacy-service:  PASS here, FAIL on GCP us-east4 (issue #304)
 #   default (legacy 2016 ws):  FAIL ('JWT token is invalid')
 #
 # Extend ONLY after empirical confirmation across at least one non-AWS stack.
+SNOWFLAKE_WORKSPACE_LOGIN_TYPE: str = "snowflake-person-keypair"
 QUERY_SERVICE_COMPATIBLE_LOGIN_TYPES: frozenset[str] = frozenset(
     {
         "snowflake-service-keypair",
         "snowflake-person-sso",
+        SNOWFLAKE_WORKSPACE_LOGIN_TYPE,
     }
 )
 
diff --git a/src/keboola_agent_cli/services/workspace_service.py b/src/keboola_agent_cli/services/workspace_service.py
index 6238ade8..481b9ac9 100644
--- a/src/keboola_agent_cli/services/workspace_service.py
+++ b/src/keboola_agent_cli/services/workspace_service.py
@@ -8,7 +8,7 @@
 import logging
 from typing import Any
 
-from ..constants import QUERY_SERVICE_COMPATIBLE_LOGIN_TYPES
+from ..constants import QUERY_SERVICE_COMPATIBLE_LOGIN_TYPES, SNOWFLAKE_WORKSPACE_LOGIN_TYPE
 from ..errors import ConfigError, ErrorCode, KeboolaApiError
 from ..models import ProjectConfig
 from .base import BaseService
@@ -28,6 +28,13 @@ def _classify_qs_compatibility(login_type: str) -> bool:
     return login_type in QUERY_SERVICE_COMPATIBLE_LOGIN_TYPES
 
 
+def _workspace_login_type_for_backend(backend: str) -> str | None:
+    """Return the loginType kbagent should request for newly created workspaces."""
+    if backend.lower() == "snowflake":
+        return SNOWFLAKE_WORKSPACE_LOGIN_TYPE
+    return None
+
+
 def find_storage_workspace_for_sandbox_config(
     workspaces: list[dict[str, Any]],
     config_id: str,
@@ -230,6 +237,7 @@ def _create_workspace_direct(
             component_id="keboola.sandboxes",
             config_id=config_id,
             backend=backend,
+            login_type=_workspace_login_type_for_backend(backend),
         )
 
         connection = ws_data.get("connection", {})
@@ -883,6 +891,7 @@ def create_from_transformation(
                 component_id=component_id,
                 config_id=config_id,
                 backend=effective_backend,
+                login_type=_workspace_login_type_for_backend(effective_backend),
             )
 
             workspace_id = ws_data.get("id")
diff --git a/tests/test_client.py b/tests/test_client.py
index f6bcee38..af902432 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -509,6 +509,71 @@ def test_storage_api_token_header(self, httpx_mock) -> None:
         client.close()
 
 
+class TestConfigWorkspaces:
+    """Tests for config-tied workspace endpoints."""
+
+    def test_create_config_workspace_includes_login_type_when_requested(
+        self, httpx_mock
+    ) -> None:
+        """Explicit loginType is included in the Storage API workspace payload."""
+        httpx_mock.add_response(
+            url=(
+                "https://connection.keboola.com/v2/storage/branch/123/components/"
+                "keboola.sandboxes/configs/cfg-1/workspaces"
+            ),
+            json={"id": 42},
+            status_code=201,
+        )
+
+        client = KeboolaClient(
+            stack_url="https://connection.keboola.com",
+            token="901-55555-fakeTestTokenDoNotUseXXXXXXXX",
+        )
+        result = client.create_config_workspace(
+            branch_id=123,
+            component_id="keboola.sandboxes",
+            config_id="cfg-1",
+            backend="snowflake",
+            login_type="snowflake-person-keypair",
+        )
+
+        assert result == {"id": 42}
+        request = httpx_mock.get_request()
+        assert json.loads(request.content) == {
+            "backend": "snowflake",
+            "loginType": "snowflake-person-keypair",
+        }
+        client.close()
+
+    def test_create_config_workspace_omits_login_type_when_default(self, httpx_mock) -> None:
+        """A None login type is omitted so non-Snowflake backends keep Storage defaults."""
+        httpx_mock.add_response(
+            url=(
+                "https://connection.keboola.com/v2/storage/branch/123/components/"
+                "keboola.sandboxes/configs/cfg-1/workspaces"
+            ),
+            json={"id": 42},
+            status_code=201,
+        )
+
+        client = KeboolaClient(
+            stack_url="https://connection.keboola.com",
+            token="901-55555-fakeTestTokenDoNotUseXXXXXXXX",
+        )
+        result = client.create_config_workspace(
+            branch_id=123,
+            component_id="keboola.sandboxes",
+            config_id="cfg-1",
+            backend="bigquery",
+            login_type=None,
+        )
+
+        assert result == {"id": 42}
+        request = httpx_mock.get_request()
+        assert json.loads(request.content) == {"backend": "bigquery"}
+        client.close()
+
+
 class TestContextManager:
     """Tests for context manager support."""
 
diff --git a/tests/test_workspace_service.py b/tests/test_workspace_service.py
index 4e118010..7ade0ff8 100644
--- a/tests/test_workspace_service.py
+++ b/tests/test_workspace_service.py
@@ -143,6 +143,7 @@ def test_create_workspace_success(self, tmp_config_dir: Path) -> None:
             component_id="keboola.sandboxes",
             config_id="cfg-123",
             backend="snowflake",
+            login_type="snowflake-person-keypair",
         )
         # close() called twice: once in _resolve_branch_id, once in create_workspace
         assert mock_client.close.call_count == 2
@@ -215,12 +216,73 @@ def test_create_workspace_in_dev_branch(self, tmp_config_dir: Path) -> None:
             component_id="keboola.sandboxes",
             config_id="cfg-456",
             backend="snowflake",
+            login_type="snowflake-person-keypair",
         )
 
 
 class TestAutoDetectBackend:
     """Tests for automatic backend detection when --backend is omitted."""
 
+    def test_create_workspace_snowflake_uses_person_keypair_login_type(
+        self, tmp_config_dir: Path
+    ) -> None:
+        """Snowflake sandbox workspaces request the Query-Service-compatible login type."""
+        mock_client = MagicMock()
+        mock_client.verify_token.return_value = SAMPLE_TOKEN_VERIFY
+        mock_client.list_dev_branches.return_value = [{"id": 123, "isDefault": True}]
+        mock_client.create_sandbox_config.return_value = {"id": "cfg-1", "name": "ws"}
+        mock_client.create_config_workspace.return_value = SAMPLE_WORKSPACE
+
+        store = setup_single_project(tmp_config_dir)
+        svc = WorkspaceService(
+            config_store=store,
+            client_factory=lambda url, token: mock_client,
+        )
+
+        result = svc.create_workspace(alias="prod", name="ws")
+
+        assert result["backend"] == "snowflake"
+        mock_client.create_config_workspace.assert_called_once_with(
+            branch_id=123,
+            component_id="keboola.sandboxes",
+            config_id="cfg-1",
+            backend="snowflake",
+            login_type="snowflake-person-keypair",
+        )
+
+    def test_create_workspace_bigquery_keeps_default_login_type(
+        self, tmp_config_dir: Path
+    ) -> None:
+        """BigQuery sandbox workspaces omit loginType so Storage uses its default."""
+        mock_client = MagicMock()
+        mock_client.verify_token.return_value = SAMPLE_TOKEN_VERIFY_BIGQUERY
+        mock_client.list_dev_branches.return_value = [{"id": 123, "isDefault": True}]
+        mock_client.create_sandbox_config.return_value = {"id": "cfg-1", "name": "ws"}
+        mock_client.create_config_workspace.return_value = {
+            "id": 42,
+            "connection": {
+                "backend": "bigquery",
+                "schema": "WORKSPACE_42",
+            },
+        }
+
+        store = setup_single_project(tmp_config_dir)
+        svc = WorkspaceService(
+            config_store=store,
+            client_factory=lambda url, token: mock_client,
+        )
+
+        result = svc.create_workspace(alias="prod", name="ws")
+
+        assert result["backend"] == "bigquery"
+        mock_client.create_config_workspace.assert_called_once_with(
+            branch_id=123,
+            component_id="keboola.sandboxes",
+            config_id="cfg-1",
+            backend="bigquery",
+            login_type=None,
+        )
+
     def test_create_workspace_auto_detects_snowflake(self, tmp_config_dir: Path) -> None:
         """create_workspace auto-detects snowflake backend from project."""
         mock_client = MagicMock()
@@ -244,6 +306,7 @@ def test_create_workspace_auto_detects_snowflake(self, tmp_config_dir: Path) ->
             component_id="keboola.sandboxes",
             config_id="cfg-1",
             backend="snowflake",
+            login_type="snowflake-person-keypair",
         )
 
     def test_create_workspace_auto_detects_bigquery(self, tmp_config_dir: Path) -> None:
@@ -276,6 +339,7 @@ def test_create_workspace_auto_detects_bigquery(self, tmp_config_dir: Path) -> N
             component_id="keboola.sandboxes",
             config_id="cfg-1",
             backend="bigquery",
+            login_type=None,
         )
 
     def test_explicit_backend_skips_auto_detect(self, tmp_config_dir: Path) -> None:
@@ -341,6 +405,7 @@ def test_from_transformation_auto_detects_bigquery(self, tmp_config_dir: Path) -
             component_id="keboola.snowflake-transformation",
             config_id="456",
             backend="bigquery",
+            login_type=None,
         )
 
 
@@ -1092,6 +1157,7 @@ def test_create_from_transformation_success(self, tmp_config_dir: Path) -> None:
             component_id="keboola.snowflake-transformation",
             config_id="456",
             backend="snowflake",
+            login_type="snowflake-person-keypair",
         )
         mock_client.load_workspace_tables.assert_called_once()
         # close() called twice: once in _resolve_branch_id, once in create_from_transformation
@@ -1371,6 +1437,21 @@ def test_list_exposes_login_type_and_qs_compatible(self, tmp_config_dir: Path) -
                 "component": "keboola.snowflake-transformation",
                 "configurationId": "cfg-2",
             },
+            {
+                "id": 3,
+                "name": "person-keypair",
+                "connection": {
+                    "backend": "snowflake",
+                    "host": "h",
+                    "schema": "S3",
+                    "user": "U3",
+                    "loginType": "snowflake-person-keypair",
+                },
+                "readOnlyStorageAccess": True,
+                "created": "2026-05-18T00:00:00Z",
+                "component": "keboola.sandboxes",
+                "configurationId": "cfg-3",
+            },
         ]
         mock_client.list_component_configs.return_value = []
 
@@ -1383,14 +1464,19 @@ def test_list_exposes_login_type_and_qs_compatible(self, tmp_config_dir: Path) -
         result = svc.list_workspaces(aliases=["prod"])
 
         workspaces = result["workspaces"]
-        assert len(workspaces) == 2
+        assert len(workspaces) == 3
         compat = next(w for w in workspaces if w["id"] == 1)
         legacy = next(w for w in workspaces if w["id"] == 2)
+        person_keypair = next(w for w in workspaces if w["id"] == 3)
 
         assert compat["login_type"] == "snowflake-service-keypair"
         assert compat["read_only"] is True
         assert compat["qs_compatible"] is True
 
+        assert person_keypair["login_type"] == "snowflake-person-keypair"
+        assert person_keypair["read_only"] is True
+        assert person_keypair["qs_compatible"] is True
+
         assert legacy["login_type"] == "default"
         assert legacy["read_only"] is False
         # ``default`` is intentionally OFF the whitelist (legacy 2016 ws,

From d3bb570cf65f8456672e1dbcda3d7c31415bb90d Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 27 May 2026 12:28:19 +0200
Subject: [PATCH 2/4] format workspace login type tests

---
 tests/test_client.py            | 4 +---
 tests/test_workspace_service.py | 4 +---
 2 files changed, 2 insertions(+), 6 deletions(-)

diff --git a/tests/test_client.py b/tests/test_client.py
index af902432..4c07b2ae 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -512,9 +512,7 @@ def test_storage_api_token_header(self, httpx_mock) -> None:
 class TestConfigWorkspaces:
     """Tests for config-tied workspace endpoints."""
 
-    def test_create_config_workspace_includes_login_type_when_requested(
-        self, httpx_mock
-    ) -> None:
+    def test_create_config_workspace_includes_login_type_when_requested(self, httpx_mock) -> None:
         """Explicit loginType is included in the Storage API workspace payload."""
         httpx_mock.add_response(
             url=(
diff --git a/tests/test_workspace_service.py b/tests/test_workspace_service.py
index 7ade0ff8..82856bba 100644
--- a/tests/test_workspace_service.py
+++ b/tests/test_workspace_service.py
@@ -250,9 +250,7 @@ def test_create_workspace_snowflake_uses_person_keypair_login_type(
             login_type="snowflake-person-keypair",
         )
 
-    def test_create_workspace_bigquery_keeps_default_login_type(
-        self, tmp_config_dir: Path
-    ) -> None:
+    def test_create_workspace_bigquery_keeps_default_login_type(self, tmp_config_dir: Path) -> None:
         """BigQuery sandbox workspaces omit loginType so Storage uses its default."""
         mock_client = MagicMock()
         mock_client.verify_token.return_value = SAMPLE_TOKEN_VERIFY_BIGQUERY

From 33c1c4f34b864d4b246e58f11b5be923d732bb66 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Thu, 28 May 2026 10:45:41 +0200
Subject: [PATCH 3/4] fix snowflake workspace keypair creation

---
 pyproject.toml                                |  1 +
 src/keboola_agent_cli/client.py               |  4 ++
 src/keboola_agent_cli/commands/workspace.py   | 20 ++++--
 .../services/workspace_service.py             | 65 +++++++++++++++----
 tests/test_client.py                          |  3 +
 tests/test_workspace_cli.py                   | 47 ++++++++++++++
 tests/test_workspace_service.py               | 39 ++++++++---
 uv.lock                                       |  2 +
 8 files changed, 154 insertions(+), 27 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index 1193872b..769e8485 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,6 +21,7 @@ dependencies = [
     "prompt-toolkit>=3.0",
     "kai-client>=0.11.0",
     "croniter>=2.0",
+    "cryptography>=46",
 ]
 
 [project.optional-dependencies]
diff --git a/src/keboola_agent_cli/client.py b/src/keboola_agent_cli/client.py
index bde607b4..43d179fd 100644
--- a/src/keboola_agent_cli/client.py
+++ b/src/keboola_agent_cli/client.py
@@ -2532,6 +2532,7 @@ def create_config_workspace(
         config_id: str,
         backend: str = "snowflake",
         login_type: str | None = None,
+        public_key: str | None = None,
     ) -> dict[str, Any]:
         """Create a workspace tied to a specific configuration.
 
@@ -2541,6 +2542,7 @@ def create_config_workspace(
             config_id: Configuration ID.
             backend: Workspace backend.
             login_type: Optional Storage API loginType. Omitted when None.
+            public_key: Optional public key for key-pair workspaces. Omitted when None.
 
         Returns:
             Workspace dict including connection credentials.
@@ -2550,6 +2552,8 @@ def create_config_workspace(
         payload: dict[str, Any] = {"backend": backend}
         if login_type is not None:
             payload["loginType"] = login_type
+        if public_key is not None:
+            payload["publicKey"] = public_key
 
         response = self._request(
             "POST",
diff --git a/src/keboola_agent_cli/commands/workspace.py b/src/keboola_agent_cli/commands/workspace.py
index 72a4d39e..e6d06829 100644
--- a/src/keboola_agent_cli/commands/workspace.py
+++ b/src/keboola_agent_cli/commands/workspace.py
@@ -91,9 +91,15 @@ def workspace_create(
                 c.print(f"[bold]Host:[/bold] {d['host']}"),
                 c.print(f"[bold]Schema:[/bold] {d['schema']}"),
                 c.print(f"[bold]User:[/bold] {d['user']}"),
-                c.print(f"[bold yellow]Password:[/bold yellow] {d['password']}"),
                 c.print(
-                    "\n[bold yellow]Warning:[/bold yellow] Save the password now -- it cannot be retrieved later!"
+                    f"[bold yellow]Private key:[/bold yellow]\n{d['private_key']}"
+                    if d.get("private_key")
+                    else f"[bold yellow]Password:[/bold yellow] {d['password']}"
+                ),
+                c.print(
+                    "\n[bold yellow]Warning:[/bold yellow] Save the private key now -- it cannot be retrieved later!"
+                    if d.get("private_key")
+                    else "\n[bold yellow]Warning:[/bold yellow] Save the password now -- it cannot be retrieved later!"
                 ),
             ),
         )
@@ -645,10 +651,16 @@ def workspace_from_transformation(
                 c.print(f"[bold]Host:[/bold] {d['host']}"),
                 c.print(f"[bold]Schema:[/bold] {d['schema']}"),
                 c.print(f"[bold]User:[/bold] {d['user']}"),
-                c.print(f"[bold yellow]Password:[/bold yellow] {d['password']}"),
+                c.print(
+                    f"[bold yellow]Private key:[/bold yellow]\n{d['private_key']}"
+                    if d.get("private_key")
+                    else f"[bold yellow]Password:[/bold yellow] {d['password']}"
+                ),
                 c.print(f"[bold]Tables loaded:[/bold] {', '.join(d.get('tables_loaded', []))}"),
                 c.print(
-                    "\n[bold yellow]Warning:[/bold yellow] Save the password now -- it cannot be retrieved later!"
+                    "\n[bold yellow]Warning:[/bold yellow] Save the private key now -- it cannot be retrieved later!"
+                    if d.get("private_key")
+                    else "\n[bold yellow]Warning:[/bold yellow] Save the password now -- it cannot be retrieved later!"
                 ),
             ),
         )
diff --git a/src/keboola_agent_cli/services/workspace_service.py b/src/keboola_agent_cli/services/workspace_service.py
index 481b9ac9..c355e0be 100644
--- a/src/keboola_agent_cli/services/workspace_service.py
+++ b/src/keboola_agent_cli/services/workspace_service.py
@@ -8,6 +8,9 @@
 import logging
 from typing import Any
 
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+
 from ..constants import QUERY_SERVICE_COMPATIBLE_LOGIN_TYPES, SNOWFLAKE_WORKSPACE_LOGIN_TYPE
 from ..errors import ConfigError, ErrorCode, KeboolaApiError
 from ..models import ProjectConfig
@@ -35,6 +38,32 @@ def _workspace_login_type_for_backend(backend: str) -> str | None:
     return None
 
 
+def _generate_snowflake_workspace_key_pair() -> tuple[str, str]:
+    """Generate the unencrypted PEM key pair required by Snowflake workspaces."""
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    private_key_pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode("ascii")
+    public_key_pem = (
+        private_key.public_key()
+        .public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+        .decode("ascii")
+    )
+    return private_key_pem, public_key_pem
+
+
+def _workspace_key_pair_for_backend(backend: str) -> tuple[str | None, str | None]:
+    """Return private/public key material for backends that require it."""
+    if backend.lower() == "snowflake":
+        return _generate_snowflake_workspace_key_pair()
+    return None, None
+
+
 def find_storage_workspace_for_sandbox_config(
     workspaces: list[dict[str, Any]],
     config_id: str,
@@ -171,8 +200,10 @@ def create_workspace(
         - Default (headless): fast (~1s) via Storage API. Not visible in Keboola UI.
         - UI mode (--ui): slower (~15s) via Queue job. Visible in UI Workspaces tab.
 
-        IMPORTANT: Password is only available on creation (headless mode).
-        In UI mode, password must be retrieved via 'workspace password' command.
+        IMPORTANT: Credentials are only available on creation (headless mode).
+        Snowflake returns a generated private key; password-based workspaces
+        return a password. In UI mode, password must be retrieved via
+        'workspace password' command.
 
         Args:
             alias: Project alias.
@@ -232,16 +263,19 @@ def _create_workspace_direct(
         read_only: bool,
     ) -> dict[str, Any]:
         """Create workspace via Storage API (fast, headless)."""
+        private_key, public_key = _workspace_key_pair_for_backend(backend)
         ws_data = client.create_config_workspace(
             branch_id=branch_id,
             component_id="keboola.sandboxes",
             config_id=config_id,
             backend=backend,
             login_type=_workspace_login_type_for_backend(backend),
+            public_key=public_key,
         )
 
         connection = ws_data.get("connection", {})
-        return {
+        credential_label = "private key" if private_key else "password"
+        result = {
             "project_alias": alias,
             "workspace_id": ws_data.get("id"),
             "name": name,
@@ -255,11 +289,12 @@ def _create_workspace_direct(
             "password": connection.get("password", ""),
             "read_only": read_only,
             "ui_mode": False,
-            "message": (
-                f"Workspace '{name}' created in project '{alias}'. "
-                "Save the password -- it cannot be retrieved later!"
-            ),
+            "message": f"Workspace '{name}' created in project '{alias}'. "
+            f"Save the {credential_label} -- it cannot be retrieved later!",
         }
+        if private_key is not None:
+            result["private_key"] = private_key
+        return result
 
     def _create_workspace_via_job(
         self,
@@ -886,12 +921,14 @@ def create_from_transformation(
                 )
 
             # Create config-tied workspace
+            private_key, public_key = _workspace_key_pair_for_backend(effective_backend)
             ws_data = client.create_config_workspace(
                 branch_id=branch_id,
                 component_id=component_id,
                 config_id=config_id,
                 backend=effective_backend,
                 login_type=_workspace_login_type_for_backend(effective_backend),
+                public_key=public_key,
             )
 
             workspace_id = ws_data.get("id")
@@ -923,7 +960,8 @@ def create_from_transformation(
                     workspace_id, table_defs, branch_id=branch_id, preserve=preserve
                 )
 
-            return {
+            credential_label = "private key" if private_key else "password"
+            result = {
                 "project_alias": alias,
                 "workspace_id": workspace_id,
                 "branch_id": branch_id,
@@ -938,11 +976,12 @@ def create_from_transformation(
                 "user": connection.get("user", ""),
                 "password": connection.get("password", ""),
                 "tables_loaded": source_tables,
-                "message": (
-                    f"Workspace {workspace_id} created from transformation "
-                    f"'{config_id}' with {len(source_tables)} table(s) loaded. "
-                    "Save the password -- it cannot be retrieved later!"
-                ),
+                "message": f"Workspace {workspace_id} created from transformation "
+                f"'{config_id}' with {len(source_tables)} table(s) loaded. "
+                f"Save the {credential_label} -- it cannot be retrieved later!",
             }
+            if private_key is not None:
+                result["private_key"] = private_key
+            return result
         finally:
             client.close()
diff --git a/tests/test_client.py b/tests/test_client.py
index 4c07b2ae..71ef7c89 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -533,6 +533,7 @@ def test_create_config_workspace_includes_login_type_when_requested(self, httpx_
             config_id="cfg-1",
             backend="snowflake",
             login_type="snowflake-person-keypair",
+            public_key="-----BEGIN PUBLIC KEY-----\ntest\n-----END PUBLIC KEY-----\n",
         )
 
         assert result == {"id": 42}
@@ -540,6 +541,7 @@ def test_create_config_workspace_includes_login_type_when_requested(self, httpx_
         assert json.loads(request.content) == {
             "backend": "snowflake",
             "loginType": "snowflake-person-keypair",
+            "publicKey": "-----BEGIN PUBLIC KEY-----\ntest\n-----END PUBLIC KEY-----\n",
         }
         client.close()
 
@@ -564,6 +566,7 @@ def test_create_config_workspace_omits_login_type_when_default(self, httpx_mock)
             config_id="cfg-1",
             backend="bigquery",
             login_type=None,
+            public_key=None,
         )
 
         assert result == {"id": 42}
diff --git a/tests/test_workspace_cli.py b/tests/test_workspace_cli.py
index 53138f02..589a8949 100644
--- a/tests/test_workspace_cli.py
+++ b/tests/test_workspace_cli.py
@@ -101,6 +101,53 @@ def test_workspace_create_success_json(self, tmp_path: Path) -> None:
         assert output["data"]["password"] == "s3cret!Passw0rd"
         assert output["data"]["backend"] == "snowflake"
 
+    def test_workspace_create_human_outputs_private_key_when_present(self, tmp_path: Path) -> None:
+        """workspace create human output shows the generated private key when returned."""
+        config_dir = tmp_path / "config"
+        config_dir.mkdir()
+
+        store = _setup_config(config_dir, {"prod": {"token": TEST_TOKEN}})
+
+        mock_ws = _make_workspace_mock()
+        mock_ws.create_workspace.return_value = {
+            "project_alias": "prod",
+            "workspace_id": 42,
+            "name": "my-workspace",
+            "config_id": "cfg-123",
+            "backend": "snowflake",
+            "host": "account.snowflakecomputing.com",
+            "warehouse": "KEBOOLA_PROD",
+            "database": "KEBOOLA_258",
+            "schema": "WORKSPACE_42",
+            "user": "KEBOOLA_WORKSPACE_42",
+            "private_key": "-----BEGIN PRIVATE KEY-----\ntest\n-----END PRIVATE KEY-----\n",
+            "read_only": True,
+            "message": "Workspace 'my-workspace' (42) created in project 'prod'. Save the private key!",
+        }
+
+        with (
+            patch("keboola_agent_cli.cli.ConfigStore") as MockStore,
+            patch("keboola_agent_cli.cli.ProjectService") as MockProjService,
+            patch("keboola_agent_cli.cli.ConfigService") as MockCfgService,
+            patch("keboola_agent_cli.cli.JobService") as MockJobService,
+            patch("keboola_agent_cli.cli.WorkspaceService") as MockWsService,
+        ):
+            MockStore.return_value = store
+            MockProjService.return_value = ProjectService(config_store=store)
+            MockCfgService.return_value = ConfigService(config_store=store)
+            MockJobService.return_value = JobService(config_store=store)
+            MockWsService.return_value = mock_ws
+
+            result = runner.invoke(
+                app,
+                ["workspace", "create", "--project", "prod"],
+            )
+
+        assert result.exit_code == 0, f"Exit code {result.exit_code}: {result.output}"
+        assert "Private key:" in result.output
+        assert "-----BEGIN PRIVATE KEY-----" in result.output
+        assert "Password:" not in result.output
+
     def test_workspace_create_api_error(self, tmp_path: Path) -> None:
         """workspace create with API error returns correct exit code."""
         config_dir = tmp_path / "config"
diff --git a/tests/test_workspace_service.py b/tests/test_workspace_service.py
index 82856bba..d9272bd5 100644
--- a/tests/test_workspace_service.py
+++ b/tests/test_workspace_service.py
@@ -6,7 +6,7 @@
 """
 
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import ANY, MagicMock
 
 import pytest
 
@@ -100,7 +100,7 @@ class TestCreateWorkspace:
     """Tests for WorkspaceService.create_workspace()."""
 
     def test_create_workspace_success(self, tmp_config_dir: Path) -> None:
-        """create_workspace returns workspace details including password."""
+        """create_workspace returns workspace details including Snowflake key-pair credentials."""
         mock_client = MagicMock()
         mock_client.list_dev_branches.return_value = [{"id": 123, "isDefault": True}]
         mock_client.create_sandbox_config.return_value = {
@@ -130,8 +130,9 @@ def test_create_workspace_success(self, tmp_config_dir: Path) -> None:
         assert result["schema"] == "WORKSPACE_42"
         assert result["user"] == "KEBOOLA_WORKSPACE_42"
         assert result["password"] == "s3cret!Passw0rd"
+        assert result["private_key"].startswith("-----BEGIN PRIVATE KEY-----")
         assert result["read_only"] is True
-        assert "Save the password" in result["message"]
+        assert "Save the private key" in result["message"]
 
         mock_client.create_sandbox_config.assert_called_once_with(
             name="test-ws",
@@ -144,7 +145,10 @@ def test_create_workspace_success(self, tmp_config_dir: Path) -> None:
             config_id="cfg-123",
             backend="snowflake",
             login_type="snowflake-person-keypair",
+            public_key=ANY,
         )
+        public_key = mock_client.create_config_workspace.call_args.kwargs["public_key"]
+        assert public_key.startswith("-----BEGIN PUBLIC KEY-----")
         # close() called twice: once in _resolve_branch_id, once in create_workspace
         assert mock_client.close.call_count == 2
 
@@ -217,7 +221,10 @@ def test_create_workspace_in_dev_branch(self, tmp_config_dir: Path) -> None:
             config_id="cfg-456",
             backend="snowflake",
             login_type="snowflake-person-keypair",
+            public_key=ANY,
         )
+        public_key = mock_client.create_config_workspace.call_args.kwargs["public_key"]
+        assert public_key.startswith("-----BEGIN PUBLIC KEY-----")
 
 
 class TestAutoDetectBackend:
@@ -242,13 +249,14 @@ def test_create_workspace_snowflake_uses_person_keypair_login_type(
         result = svc.create_workspace(alias="prod", name="ws")
 
         assert result["backend"] == "snowflake"
-        mock_client.create_config_workspace.assert_called_once_with(
-            branch_id=123,
-            component_id="keboola.sandboxes",
-            config_id="cfg-1",
-            backend="snowflake",
-            login_type="snowflake-person-keypair",
-        )
+        assert result["private_key"].startswith("-----BEGIN PRIVATE KEY-----")
+        call_kwargs = mock_client.create_config_workspace.call_args.kwargs
+        assert call_kwargs["branch_id"] == 123
+        assert call_kwargs["component_id"] == "keboola.sandboxes"
+        assert call_kwargs["config_id"] == "cfg-1"
+        assert call_kwargs["backend"] == "snowflake"
+        assert call_kwargs["login_type"] == "snowflake-person-keypair"
+        assert call_kwargs["public_key"].startswith("-----BEGIN PUBLIC KEY-----")
 
     def test_create_workspace_bigquery_keeps_default_login_type(self, tmp_config_dir: Path) -> None:
         """BigQuery sandbox workspaces omit loginType so Storage uses its default."""
@@ -279,6 +287,7 @@ def test_create_workspace_bigquery_keeps_default_login_type(self, tmp_config_dir
             config_id="cfg-1",
             backend="bigquery",
             login_type=None,
+            public_key=None,
         )
 
     def test_create_workspace_auto_detects_snowflake(self, tmp_config_dir: Path) -> None:
@@ -305,7 +314,10 @@ def test_create_workspace_auto_detects_snowflake(self, tmp_config_dir: Path) ->
             config_id="cfg-1",
             backend="snowflake",
             login_type="snowflake-person-keypair",
+            public_key=ANY,
         )
+        public_key = mock_client.create_config_workspace.call_args.kwargs["public_key"]
+        assert public_key.startswith("-----BEGIN PUBLIC KEY-----")
 
     def test_create_workspace_auto_detects_bigquery(self, tmp_config_dir: Path) -> None:
         """create_workspace auto-detects bigquery backend from project."""
@@ -338,6 +350,7 @@ def test_create_workspace_auto_detects_bigquery(self, tmp_config_dir: Path) -> N
             config_id="cfg-1",
             backend="bigquery",
             login_type=None,
+            public_key=None,
         )
 
     def test_explicit_backend_skips_auto_detect(self, tmp_config_dir: Path) -> None:
@@ -404,6 +417,7 @@ def test_from_transformation_auto_detects_bigquery(self, tmp_config_dir: Path) -
             config_id="456",
             backend="bigquery",
             login_type=None,
+            public_key=None,
         )
 
 
@@ -1143,8 +1157,10 @@ def test_create_from_transformation_success(self, tmp_config_dir: Path) -> None:
         assert result["row_id"] is None
         assert result["backend"] == "snowflake"
         assert result["password"] == "ws-secret-pwd"
+        assert result["private_key"].startswith("-----BEGIN PRIVATE KEY-----")
         assert result["tables_loaded"] == ["in.c-main.orders", "in.c-main.products"]
         assert "2 table(s) loaded" in result["message"]
+        assert "Save the private key" in result["message"]
 
         mock_client.get_config_detail.assert_called_once_with(
             "keboola.snowflake-transformation",
@@ -1156,7 +1172,10 @@ def test_create_from_transformation_success(self, tmp_config_dir: Path) -> None:
             config_id="456",
             backend="snowflake",
             login_type="snowflake-person-keypair",
+            public_key=ANY,
         )
+        public_key = mock_client.create_config_workspace.call_args.kwargs["public_key"]
+        assert public_key.startswith("-----BEGIN PUBLIC KEY-----")
         mock_client.load_workspace_tables.assert_called_once()
         # close() called twice: once in _resolve_branch_id, once in create_from_transformation
         assert mock_client.close.call_count == 2
diff --git a/uv.lock b/uv.lock
index d4fb1dfd..b3b5c76f 100644
--- a/uv.lock
+++ b/uv.lock
@@ -500,6 +500,7 @@ version = "0.47.1"
 source = { editable = "." }
 dependencies = [
     { name = "croniter" },
+    { name = "cryptography" },
     { name = "httpx" },
     { name = "jsonschema" },
     { name = "kai-client" },
@@ -535,6 +536,7 @@ dev = [
 [package.metadata]
 requires-dist = [
     { name = "croniter", specifier = ">=2.0" },
+    { name = "cryptography", specifier = ">=46" },
     { name = "fastapi", marker = "extra == 'server'", specifier = ">=0.115" },
     { name = "httpx", specifier = ">=0.27" },
     { name = "jsonschema", specifier = ">=4.20" },

From 8423f76b83e1e7ceead31797a5a2f222ccf80f1a Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Thu, 28 May 2026 20:30:41 +0200
Subject: [PATCH 4/4] Address Snowflake workspace review feedback

---
 plugins/kbagent/agents/keboola-expert.md      |  8 +++-
 .../kbagent/references/commands-reference.md  |  2 +-
 .../skills/kbagent/references/gotchas.md      | 23 +++++++++++
 src/keboola_agent_cli/changelog.py            |  1 +
 src/keboola_agent_cli/commands/context.py     |  1 +
 .../hints/definitions/workspace.py            |  7 +++-
 .../services/workspace_service.py             | 40 ++++++++++++-------
 tests/test_e2e.py                             | 17 ++++++++
 tests/test_workspace_cli.py                   |  4 +-
 9 files changed, 85 insertions(+), 18 deletions(-)

diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md
index 099f6677..95799d5c 100644
--- a/plugins/kbagent/agents/keboola-expert.md
+++ b/plugins/kbagent/agents/keboola-expert.md
@@ -1,6 +1,6 @@
 ---
 name: keboola-expert
-description: Keboola Connection operations specialist. MUST BE USED proactively for any task touching Keboola projects -- config browsing/updates, jobs, flows, schedules, storage, migrations, dev branches, debugging. Enforces fresh-fetch discipline, --dry-run on writes, CLI over REST, and refuses tasks it cannot safely complete with the installed kbagent version. Delegates write operations through two-step (dry-run -> confirm -> apply) flow without exception.
+description: Keboola Connection ops specialist. Enforces fresh-fetch, dry-run, CLI-over-REST, version gate and confirmed apply.
 tools: Bash, Read, Edit, Write, Grep, Glob, TodoWrite, WebFetch
 model: sonnet
 color: blue
@@ -111,6 +111,7 @@ a critical failure.
    `data-app logs` = 0.43.8+,
    `kbagent agent <verb>` (CLI parity /agents REST) = 0.44.0+,
    `semantic-layer search-context|get-context`, `storage create-table --if-not-exists`, `sync push|pull|diff --branch`, `sync push --no-name-drift-warnings`, fresh-CREATE writeback + KBC.* = 0.47.0+,
+   Snowflake `workspace create` `private_key` = 0.47.1+,
    `storage retype` is a future composite), you
    MUST refuse the task and return a handoff message to the parent:
    `"Cannot proceed safely on kbagent <version>. Missing: <commands>.
@@ -219,6 +220,11 @@ success, not a failure.
   lower-level schema. `config new --push` does NOT inherit the MCP
   refusal because it calls Storage API directly.
 
+- **Snowflake workspace credentials** (0.47.1+): headless
+  `kbagent workspace create` returns `private_key` for Snowflake; `password`
+  is empty/unusable. Use the one-time PKCS8 PEM for key-pair auth. BigQuery
+  keeps the prior shape.
+
 - **`script[]` string-vs-array runtime crash** (0.28.0+ auto-fix; #245):
   the Storage API silently accepts `parameters.blocks[].codes[].script`
   as a string, but the runtime validator rejects it (`Expected array,
diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md
index 12e40136..0150d1f7 100644
--- a/plugins/kbagent/skills/kbagent/references/commands-reference.md
+++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md
@@ -128,7 +128,7 @@ All seven commands authenticate via `KBC_MANAGE_API_TOKEN` (Manage API), not the
 - `branch metadata-delete --project NAME --metadata-id ID [--branch ID|default]` -- delete a metadata entry by its numeric ID (from `metadata-list`)
 
 ## Workspaces (SQL Debugging)
-- `workspace create --project ALIAS [--name NAME] [--ui] [--read-only]` -- create workspace (headless ~1s, `--ui` ~15s)
+- `workspace create --project ALIAS [--name NAME] [--ui] [--read-only]` -- create workspace (headless ~1s, `--ui` ~15s). Since v0.47.1: Snowflake headless workspaces return a `private_key` PEM field; `password` is empty. BigQuery workspaces keep the default password credential shape.
 - `workspace list [--project NAME ...] [--orphaned] [--branch ID] [--qs-compatible]` -- list workspaces. `--project` repeatable; `--orphaned` filters to workspaces whose backing `keboola.sandboxes` config is missing. **Since v0.42.0 (#304)**: each entry carries `login_type`, `read_only`, `qs_compatible`, `database`, `warehouse`. New `Login Type` / `RO` / `QS` columns in human mode. `--qs-compatible` pre-filters to RO + whitelisted-loginType workspaces (the canonical data-app shape). `--branch` requires exactly one `--project`; without `--branch`, the command behaves like `storage buckets` and uses production with an `Info: Using production branch for read (active dev branch X ignored; pass --branch X to override)` banner when an alias is pinned to a dev branch
 - `workspace detail --project ALIAS --workspace-id ID [--branch ID]` -- show connection details. **Since v0.42.0 (#304)**: response carries `login_type`, `read_only`, `qs_compatible`; human mode adds `Login type:` / `Read-only:` / `Query Service compatible:` rows. `--branch` opt-in mirrors `workspace list`
 - `workspace delete --project ALIAS --workspace-id ID` -- delete workspace
diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md
index 7f0d256e..20f8f79e 100644
--- a/plugins/kbagent/skills/kbagent/references/gotchas.md
+++ b/plugins/kbagent/skills/kbagent/references/gotchas.md
@@ -148,6 +148,7 @@ plus a derived `qs_compatible: bool`.
 
 - `snowflake-service-keypair` -- confirmed PASS
 - `snowflake-person-sso` -- confirmed PASS
+- `snowflake-person-keypair` -- confirmed PASS (since v0.47.1)
 - `snowflake-legacy-service` -- explicitly OFF the list (works on
   `connection.keboola.com` but FAILED on GCP us-east4 stack in the
   original #304 incident -- keep it off until cross-stack confirmation)
@@ -159,6 +160,28 @@ confirmed-good whitelist". For an unknown loginType, `workspace list`
 renders it as `?` (yellow) in the QS column so callers know the policy
 is uncertain rather than confirmed-bad.
 
+## Snowflake `workspace create` returns `private_key`, not password (since v0.47.1)
+
+Headless `workspace create` on Snowflake requests
+`loginType: snowflake-person-keypair`, generates an RSA key pair locally,
+passes the public key to the Storage API, and returns the private key once in
+the creation envelope:
+
+```jsonc
+{
+  "backend": "snowflake",
+  "user": "KEBOOLA_WORKSPACE_42",
+  "password": "",
+  "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n"
+}
+```
+
+For successful Snowflake creates, `private_key` is the credential to save and
+use for key-pair authentication; `password` remains in the envelope for
+backward compatibility but should be treated as empty/unusable. BigQuery
+workspaces keep the previous password-based/default backend shape and do not
+return `private_key`.
+
 **Filter (data-app pre-selection):**
 
 ```bash
diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py
index 088da5d2..c996a277 100644
--- a/src/keboola_agent_cli/changelog.py
+++ b/src/keboola_agent_cli/changelog.py
@@ -10,6 +10,7 @@
 CHANGELOG: dict[str, list[str]] = {
     "0.47.1": [
         'Fix (`storage create-table --if-not-exists`, keboola/cli#349): the `action: "skipped"` envelope now reports the EXISTING table\'s actual schema instead of re-echoing the caller\'s request. Pre-0.47.1, `columns` / `primary_key` / `name` on a skip mirrored the args the caller passed in, so a caller probing the skipped envelope to discover the real shape of a pre-existing table got the wrong values whenever the existing table differed from the request. The `get_table_detail(target_id)` lookup that already runs to confirm the table exists is now also the source of the returned schema. The caller\'s requested values are preserved under two new fields, `requested_columns` and `requested_primary_key`, and a new `schema_drift: bool` flags when the existing table diverges from the request (set comparison on columns and primary key). Human-mode output prints the actual schema on a skip and emits a `Warning:` line when `schema_drift` is true. `action: "created"` envelope is unchanged. No new flag, no signature change. Tests: `tests/test_storage_write.py` (skipped returns actual schema, drift flag set on divergence, no drift on match, human-mode warning render); `tests/test_e2e.py::TestE2E_0_47_0_NewSurfaces` extended to assert the skipped envelope reports actual columns + `requested_*` mirror.',
+        'Fix (`workspace create`, keboola/cli#351): new Snowflake sandbox workspaces now request `loginType: "snowflake-person-keypair"` and generate a local RSA key pair for the Storage API `publicKey` field, so the created workspace uses the Query-Service-compatible login type instead of the backend default. The one-time creation envelope now includes `private_key` for Snowflake workspaces and keeps `password` for compatibility, usually empty on key-pair workspaces; human output prints the private key and warns that it cannot be retrieved later. BigQuery workspaces still omit `loginType` and `publicKey`. Tests cover the Storage client payload, service-layer Snowflake/BigQuery branching, CLI JSON/human output, and Snowflake E2E `private_key` presence.',
     ],
     "0.47.0": [
         "Fix (sync push, fresh-CREATE): pre-existing placeholder manifest entries -- the FIIA / scaffold emit pattern, where a downstream tool seeds `.keboola/manifest.json` with placeholder ids and (optionally) `KBC.configuration.*` metadata before the first push -- are now updated **in place** by the create path instead of unconditionally appended. Pre-0.47.0 every create did `manifest.configurations.append(ManifestConfiguration(...))` (and `parent.rows.append(...)` for rows), so N placeholders -> 2N manifest entries after one push, every placeholder still looked `added` on re-push (spurious duplicates on remote), and any `metadata.KBC.configuration.folderName` declared in the placeholder was silently dropped on the floor. Two new private helpers do the work: `SyncService._writeback_create_config_in_manifest(...)` finds the placeholder by `(branch_id, component_id, path)` -- branch is part of the key so a multi-branch manifest with the same logical path under two branches updates the right entry -- and refreshes its id + pull_hash / pull_config_hash while preserving every non-bookkeeping metadata key; `SyncService._writeback_create_row_in_manifest(...)` does the same for rows under their parent. Idempotency on re-push falls out for free: the now-real config id flows through the existing diff engine and the second push reports `status: no_changes, created: 0`. Tests: `tests/test_sync_service.py::TestFreshCreateWriteback` (7 cases incl. an end-to-end placeholder + KBC-metadata round-trip). Manifest contract change for downstream parsers: a single CREATE now produces a single manifest entry (not placeholder + new). Downstream tooling that has been working around the duplication by post-processing must drop that workaround. Live-validated against project 1143 / dev branch 388071: placeholder with `KBC.configuration.folderName: 'Area B E2E Folder'` -> `created=1, errors=0`, manifest length 1, folderName visible via `config metadata-list`, re-push -> `no_changes`.",
diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py
index 02dca1cf..c069bf40 100644
--- a/src/keboola_agent_cli/commands/context.py
+++ b/src/keboola_agent_cli/commands/context.py
@@ -610,6 +610,7 @@
 
   kbagent workspace create --project ALIAS [--name NAME] [--backend TYPE] [--ui] [--read-only/--no-read-only]
     Create workspace. Backend auto-detected from project (or override with --backend). Default: headless (~1s). --ui: visible in KBC UI (~15s).
+    Since 0.47.1, Snowflake headless creates return private_key and an empty password field; use key-pair auth.
 
   kbagent workspace list [--project NAME] [--orphaned] [--branch ID] [--qs-compatible]
     List workspaces. Read command: ignores active dev branch (production endpoint) with an Info banner;
diff --git a/src/keboola_agent_cli/hints/definitions/workspace.py b/src/keboola_agent_cli/hints/definitions/workspace.py
index 17694956..40810c32 100644
--- a/src/keboola_agent_cli/hints/definitions/workspace.py
+++ b/src/keboola_agent_cli/hints/definitions/workspace.py
@@ -14,7 +14,11 @@
                 comment="Create workspace (headless mode)",
                 client=ClientCall(
                     method="create_config_workspace",
-                    args={"backend": "{backend}"},
+                    args={
+                        "backend": "{backend}",
+                        "login_type": "login_type",
+                        "public_key": "public_key_pem",
+                    },
                     result_var="workspace",
                     result_hint="dict",
                 ),
@@ -33,6 +37,7 @@
         ],
         notes=[
             "Service layer handles sandbox config creation + workspace provisioning.",
+            "For Snowflake direct client calls, generate an RSA key pair, pass login_type='snowflake-person-keypair' and public_key=public_key_pem, then save the private key returned by your generator. For BigQuery, pass None for both.",
             "With --ui flag, creates via job run (slower, ~15s) for UI visibility.",
         ],
     )
diff --git a/src/keboola_agent_cli/services/workspace_service.py b/src/keboola_agent_cli/services/workspace_service.py
index c355e0be..39dec856 100644
--- a/src/keboola_agent_cli/services/workspace_service.py
+++ b/src/keboola_agent_cli/services/workspace_service.py
@@ -6,6 +6,7 @@
 """
 
 import logging
+from dataclasses import dataclass
 from typing import Any
 
 from cryptography.hazmat.primitives import serialization
@@ -19,6 +20,14 @@
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class SnowflakeWorkspaceKeyPair:
+    """PEM key material for Snowflake key-pair workspace authentication."""
+
+    private_pem: str
+    public_pem: str
+
+
 def _classify_qs_compatibility(login_type: str) -> bool:
     """Map a Storage API workspace ``connection.loginType`` to Query-Service compat.
 
@@ -38,7 +47,7 @@ def _workspace_login_type_for_backend(backend: str) -> str | None:
     return None
 
 
-def _generate_snowflake_workspace_key_pair() -> tuple[str, str]:
+def _generate_snowflake_workspace_key_pair() -> SnowflakeWorkspaceKeyPair:
     """Generate the unencrypted PEM key pair required by Snowflake workspaces."""
     private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
     private_key_pem = private_key.private_bytes(
@@ -54,14 +63,14 @@ def _generate_snowflake_workspace_key_pair() -> tuple[str, str]:
         )
         .decode("ascii")
     )
-    return private_key_pem, public_key_pem
+    return SnowflakeWorkspaceKeyPair(private_pem=private_key_pem, public_pem=public_key_pem)
 
 
-def _workspace_key_pair_for_backend(backend: str) -> tuple[str | None, str | None]:
+def _workspace_key_pair_for_backend(backend: str) -> SnowflakeWorkspaceKeyPair | None:
     """Return private/public key material for backends that require it."""
     if backend.lower() == "snowflake":
         return _generate_snowflake_workspace_key_pair()
-    return None, None
+    return None
 
 
 def find_storage_workspace_for_sandbox_config(
@@ -263,18 +272,18 @@ def _create_workspace_direct(
         read_only: bool,
     ) -> dict[str, Any]:
         """Create workspace via Storage API (fast, headless)."""
-        private_key, public_key = _workspace_key_pair_for_backend(backend)
+        key_pair = _workspace_key_pair_for_backend(backend)
         ws_data = client.create_config_workspace(
             branch_id=branch_id,
             component_id="keboola.sandboxes",
             config_id=config_id,
             backend=backend,
             login_type=_workspace_login_type_for_backend(backend),
-            public_key=public_key,
+            public_key=key_pair.public_pem if key_pair else None,
         )
 
         connection = ws_data.get("connection", {})
-        credential_label = "private key" if private_key else "password"
+        credential_label = "private key" if key_pair else "password"
         result = {
             "project_alias": alias,
             "workspace_id": ws_data.get("id"),
@@ -292,8 +301,8 @@ def _create_workspace_direct(
             "message": f"Workspace '{name}' created in project '{alias}'. "
             f"Save the {credential_label} -- it cannot be retrieved later!",
         }
-        if private_key is not None:
-            result["private_key"] = private_key
+        if key_pair is not None:
+            result["private_key"] = key_pair.private_pem
         return result
 
     def _create_workspace_via_job(
@@ -305,6 +314,9 @@ def _create_workspace_via_job(
         backend: str,
     ) -> dict[str, Any]:
         """Create workspace via Queue job (slower, visible in UI)."""
+        # The Queue job path does not expose a publicKey/loginType input. Keep
+        # returning a reset password here; headless Snowflake creates use the
+        # key-pair path in _create_workspace_direct().
         job = client.create_job(
             component_id="keboola.sandboxes",
             config_id=config_id,
@@ -921,14 +933,14 @@ def create_from_transformation(
                 )
 
             # Create config-tied workspace
-            private_key, public_key = _workspace_key_pair_for_backend(effective_backend)
+            key_pair = _workspace_key_pair_for_backend(effective_backend)
             ws_data = client.create_config_workspace(
                 branch_id=branch_id,
                 component_id=component_id,
                 config_id=config_id,
                 backend=effective_backend,
                 login_type=_workspace_login_type_for_backend(effective_backend),
-                public_key=public_key,
+                public_key=key_pair.public_pem if key_pair else None,
             )
 
             workspace_id = ws_data.get("id")
@@ -960,7 +972,7 @@ def create_from_transformation(
                     workspace_id, table_defs, branch_id=branch_id, preserve=preserve
                 )
 
-            credential_label = "private key" if private_key else "password"
+            credential_label = "private key" if key_pair else "password"
             result = {
                 "project_alias": alias,
                 "workspace_id": workspace_id,
@@ -980,8 +992,8 @@ def create_from_transformation(
                 f"'{config_id}' with {len(source_tables)} table(s) loaded. "
                 f"Save the {credential_label} -- it cannot be retrieved later!",
             }
-            if private_key is not None:
-                result["private_key"] = private_key
+            if key_pair is not None:
+                result["private_key"] = key_pair.private_pem
             return result
         finally:
             client.close()
diff --git a/tests/test_e2e.py b/tests/test_e2e.py
index dd84bdb8..edcbc527 100644
--- a/tests/test_e2e.py
+++ b/tests/test_e2e.py
@@ -6189,6 +6189,23 @@ def _run(self, *args: str) -> Any:
     def _run_ok(self, *args: str) -> dict[str, Any]:
         return _json_ok(self._run(*args))
 
+    def test_snowflake_workspace_create_returns_private_key(self) -> None:
+        """Snowflake workspace creation returns the generated private key once."""
+        _step(1, "workspace create returns private_key on Snowflake")
+        result = self._run("workspace", "create", "--project", self.alias)
+        if result.exit_code != 0:
+            pytest.skip(f"workspace create not supported: {result.output}")
+
+        data = _json_ok(result)["data"]
+        ws_id = int(data["workspace_id"])
+        self._created_workspace_ids.append(ws_id)
+
+        if data.get("backend") != "snowflake":
+            pytest.skip("Snowflake private_key assertion requires a Snowflake stack")
+
+        assert "private_key" in data
+        assert data["private_key"].startswith("-----BEGIN PRIVATE KEY-----")
+
     def test_issue_304_discoverability_roundtrip(self) -> None:
         """list/detail expose loginType; sandbox config annotation resolves real workspace ID."""
         _step(1, "workspace create (RO sandbox)")
diff --git a/tests/test_workspace_cli.py b/tests/test_workspace_cli.py
index 589a8949..f76c9c16 100644
--- a/tests/test_workspace_cli.py
+++ b/tests/test_workspace_cli.py
@@ -72,8 +72,9 @@ def test_workspace_create_success_json(self, tmp_path: Path) -> None:
             "schema": "WORKSPACE_42",
             "user": "KEBOOLA_WORKSPACE_42",
             "password": "s3cret!Passw0rd",
+            "private_key": "-----BEGIN PRIVATE KEY-----\ntest\n-----END PRIVATE KEY-----\n",
             "read_only": True,
-            "message": "Workspace 'my-workspace' (42) created in project 'prod'. Save the password!",
+            "message": "Workspace 'my-workspace' (42) created in project 'prod'. Save the private key!",
         }
 
         with (
@@ -99,6 +100,7 @@ def test_workspace_create_success_json(self, tmp_path: Path) -> None:
         assert output["status"] == "ok"
         assert output["data"]["workspace_id"] == 42
         assert output["data"]["password"] == "s3cret!Passw0rd"
+        assert output["data"]["private_key"].startswith("-----BEGIN PRIVATE KEY-----")
         assert output["data"]["backend"] == "snowflake"
 
     def test_workspace_create_human_outputs_private_key_when_present(self, tmp_path: Path) -> None: