feat(databricks-volume): add Python token provider authentication

docs/python/resource/databricks_volume.mdx

@@ -17,14 +17,19 @@ from mirage import MountMode, Workspace
 from mirage.resource.databricks_volume import (
     DatabricksVolumeConfig,
     DatabricksVolumeResource,
+    StaticTokenProvider,
 )
 
-resource = DatabricksVolumeResource(DatabricksVolumeConfig(
-    catalog="main",
-    schema="default",
-    volume="agent_files",
-    root_path="/reports",
-))
+resource = DatabricksVolumeResource(
+    DatabricksVolumeConfig(
+        host="https://example.cloud.databricks.com",
+        catalog="main",
+        schema="default",
+        volume="agent_files",
+        root_path="/reports",
+    ),
+    token_provider=StaticTokenProvider("token"),
+)
 ws = Workspace({"/dbx": resource}, mode=MountMode.READ)
 ```
 
@@ -33,12 +38,15 @@ ws = Workspace({"/dbx": resource}, mode=MountMode.READ)
 | `catalog` | required | Unity Catalog catalog name. |
 | `schema` | required | Unity Catalog schema name. |
 | `volume` | required | Unity Catalog volume name. |
+| `host` | required | Databricks workspace host. |
 | `root_path` | `/` | Subdirectory inside the volume to expose. |
-| `host` | `None` | Optional workspace host override. |
-| `token` | `None` | Optional PAT override. Redacted in snapshots. |
-| `profile` | `None` | Optional Databricks SDK profile name. |
 | `timeout` | `30` | Request timeout in seconds. |
 
+Credentials are supplied separately through a token provider. Mirage includes
+`StaticTokenProvider` for fixed tokens and `DatabricksProfileTokenProvider` for
+Databricks CLI profiles. Applications can implement `get_token()` to supply
+rotating or request-scoped tokens.
+
 ## Mount mode
 
 `read` or `write`.
@@ -51,6 +59,7 @@ Given:
 
 ```python
 DatabricksVolumeConfig(
+    host="https://example.cloud.databricks.com",
     catalog="main",
     schema="default",
     volume="agent_files",
@@ -134,21 +143,38 @@ or local-only utilities).
 
 ## Snapshot behavior
 
-`token` is redacted in resource state. Loading a snapshot back requires an
-override config that provides fresh credentials if the runtime auth chain does
-not already supply them.
+Snapshots contain the volume location config but never the token provider.
+Loading a snapshot requires a full resource override for the mount:
+
+```python
+workspace = Workspace.load(
+    snapshot,
+    resources={"/dbx/": resource},
+)
+```
 
 ## Databricks Apps
 
-For Databricks Apps, prefer SDK-default auth and keep Mirage in-process:
+For Databricks Apps, pass an application-owned provider. Its `get_token()`
+method can read the current request's OBO token or return a service-principal
+token managed by the application:
 
 ```python
+class AppTokenProvider:
+    def get_token(self) -> str:
+        return current_request_token.get()
+
+
 config = DatabricksVolumeConfig(
+    host="https://example.cloud.databricks.com",
     catalog="main",
     schema="default",
     volume="agent_files",
 )
-resource = DatabricksVolumeResource(config)
+resource = DatabricksVolumeResource(
+    config,
+    token_provider=AppTokenProvider(),
+)
 ws = Workspace({"/dbx/": resource}, mode=MountMode.READ)
 ```
 
@@ -162,16 +188,22 @@ import asyncio
 
 from mirage import MountMode, Workspace
 from mirage.resource.databricks_volume import (
+    DatabricksProfileTokenProvider,
     DatabricksVolumeConfig,
     DatabricksVolumeResource,
 )
 
-resource = DatabricksVolumeResource(DatabricksVolumeConfig(
-    catalog="main",
-    schema="default",
-    volume="agent_files",
-    root_path="/reports",
-))
+host = "https://example.cloud.databricks.com"
+resource = DatabricksVolumeResource(
+    DatabricksVolumeConfig(
+        host=host,
+        catalog="main",
+        schema="default",
+        volume="agent_files",
+        root_path="/reports",
+    ),
+    token_provider=DatabricksProfileTokenProvider(host),
+)
 
 
 async def main() -> None:

docs/python/setup/databricks.mdx

@@ -15,69 +15,94 @@ guide.
 
 ## Configuration
 
-### Databricks Apps or SDK-default auth
+### Static token
 
 ```python
+import os
+
 from mirage import Workspace, MountMode
 from mirage.resource.databricks_volume import (
     DatabricksVolumeConfig,
     DatabricksVolumeResource,
+    StaticTokenProvider,
 )
 
 config = DatabricksVolumeConfig(
+    host=os.environ["DATABRICKS_HOST"],
     catalog="main",
     schema="default",
     volume="agent_files",
+    root_path="/reports",
+)
+resource = DatabricksVolumeResource(
+    config,
+    token_provider=StaticTokenProvider(os.environ["DATABRICKS_TOKEN"]),
 )
-resource = DatabricksVolumeResource(config=config)
 ws = Workspace({"/dbx/": resource}, mode=MountMode.READ)
 ```
 
-### Explicit host and token
+### Profile-based auth
 
 ```python
 import os
 
+from mirage.resource.databricks_volume import DatabricksProfileTokenProvider
+
+host = os.environ["DATABRICKS_HOST"]
 config = DatabricksVolumeConfig(
+    host=host,
     catalog="main",
     schema="default",
     volume="agent_files",
-    host=os.environ["DATABRICKS_HOST"],
-    token=os.environ["DATABRICKS_TOKEN"],
-    root_path="/reports",
 )
-resource = DatabricksVolumeResource(config=config)
+resource = DatabricksVolumeResource(
+    config,
+    token_provider=DatabricksProfileTokenProvider(host, profile="DEV"),
+)
 ws = Workspace({"/dbx/": resource}, mode=MountMode.READ)
 ```
 
-### Profile-based auth
+### Custom token strategy
+
+Implement `get_token()` when the application owns token caching, refresh, or
+request-scoped OBO credentials. It may return a string or an awaitable string.
 
 ```python
+class AppTokenProvider:
+    async def get_token(self) -> str:
+        return await application_credentials.current_token()
+
+
 config = DatabricksVolumeConfig(
+    host="https://example.cloud.databricks.com",
     catalog="main",
     schema="default",
     volume="agent_files",
-    profile="DEV",
 )
-resource = DatabricksVolumeResource(config=config)
+resource = DatabricksVolumeResource(
+    config,
+    token_provider=AppTokenProvider(),
+)
 ws = Workspace({"/dbx/": resource}, mode=MountMode.READ)
 ```
 
+Mirage asks the provider for a token at each Files API operation. The provider
+decides whether to return a cached token or refresh it. Mirage does not retry a
+rejected token.
+
 ## Config Reference
 
 | Field | Required | Default | Description |
 | --- | --- | --- | --- |
 | `catalog` | Yes |  | Unity Catalog catalog name. |
 | `schema` | Yes |  | Unity Catalog schema name. |
 | `volume` | Yes |  | Unity Catalog volume name. |
+| `host` | Yes |  | Databricks workspace host. |
 | `root_path` | No | `/` | Subdirectory inside the volume to expose. |
-| `host` | No |  | Databricks workspace host. |
-| `token` | No |  | Databricks personal access token. Redacted in snapshots. |
-| `profile` | No |  | Databricks SDK profile name. |
 | `timeout` | No | `30` | Request timeout in seconds. |
 
 ## Notes
 
 - Supports both read and write mount modes (`MountMode.READ` / `MountMode.WRITE`).
-- Auth falls through to the Databricks SDK defaults when `host`, `token`, and `profile` are omitted.
+- Credentials are supplied separately through a token provider and are never serialized.
 - `root_path` is normalized and cannot contain `..`.

examples/python/agents/langchain/databricks_volume_deepagent.py

@@ -21,21 +21,30 @@
 from mirage import MountMode, Workspace
 from mirage.agents.langchain import (LangchainWorkspace, build_system_prompt,
                                      extract_text)
-from mirage.resource.databricks_volume import (DatabricksVolumeConfig,
-                                               DatabricksVolumeResource)
+from mirage.resource.databricks_volume import (DatabricksProfileTokenProvider,
+                                               DatabricksVolumeConfig,
+                                               DatabricksVolumeResource,
+                                               StaticTokenProvider)
 
 load_dotenv(".env.development")
 
+host = os.environ["DATABRICKS_HOST"]
+token = os.environ.get("DATABRICKS_TOKEN")
+token_provider = (StaticTokenProvider(token)
+                  if token else DatabricksProfileTokenProvider(
+                      host,
+                      os.environ.get("DATABRICKS_CONFIG_PROFILE", "DEFAULT"),
+                  ))
 resource = DatabricksVolumeResource(
     DatabricksVolumeConfig(
+        host=host,
         catalog=os.environ["DATABRICKS_VOLUME_CATALOG"],
         schema=os.environ["DATABRICKS_VOLUME_SCHEMA"],
         volume=os.environ["DATABRICKS_VOLUME_NAME"],
         root_path=os.environ.get("DATABRICKS_VOLUME_ROOT_PATH", "/"),
-        host=os.environ.get("DATABRICKS_HOST"),
-        token=os.environ.get("DATABRICKS_TOKEN"),
-        profile=os.environ.get("DATABRICKS_CONFIG_PROFILE"),
-    ))
+    ),
+    token_provider=token_provider,
+)
 
 ws = Workspace({"/dbx/": resource}, mode=MountMode.READ)
 

examples/python/databricks_volume/databricks_volume.py

@@ -18,21 +18,31 @@
 from dotenv import load_dotenv
 
 from mirage import MountMode, Workspace
-from mirage.resource.databricks_volume import (DatabricksVolumeConfig,
-                                               DatabricksVolumeResource)
+from mirage.resource.databricks_volume import (DatabricksProfileTokenProvider,
+                                               DatabricksVolumeConfig,
+                                               DatabricksVolumeResource,
+                                               StaticTokenProvider)
 
 load_dotenv(".env.development")
 
+host = os.environ["DATABRICKS_HOST"]
+token = os.environ.get("DATABRICKS_TOKEN")
+token_provider = (StaticTokenProvider(token)
+                  if token else DatabricksProfileTokenProvider(
+                      host,
+                      os.environ.get("DATABRICKS_CONFIG_PROFILE", "DEFAULT"),
+                  ))
 config = DatabricksVolumeConfig(
+    host=host,
     catalog=os.environ["DATABRICKS_VOLUME_CATALOG"],
     schema=os.environ["DATABRICKS_VOLUME_SCHEMA"],
     volume=os.environ["DATABRICKS_VOLUME_NAME"],
     root_path=os.environ.get("DATABRICKS_VOLUME_ROOT_PATH", "/"),
-    host=os.environ.get("DATABRICKS_HOST"),
-    token=os.environ.get("DATABRICKS_TOKEN"),
-    profile=os.environ.get("DATABRICKS_CONFIG_PROFILE"),
 )
-resource = DatabricksVolumeResource(config=config)
+resource = DatabricksVolumeResource(
+    config=config,
+    token_provider=token_provider,
+)
 
 
 async def _run(ws, cmd):

python/mirage/accessor/databricks_volume.py

@@ -12,49 +12,17 @@
 # limitations under the License.
 # ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
 
-from typing import Any
-
 from mirage.accessor.base import Accessor
+from mirage.core.databricks_volume.client import DatabricksFilesClient
 from mirage.resource.databricks_volume.config import DatabricksVolumeConfig
 
-try:
-    from databricks.sdk import WorkspaceClient
-    from databricks.sdk.config import Config as WorkspaceConfig
-except ImportError:
-    WorkspaceConfig = None
-    WorkspaceClient = None
-
 
 class DatabricksVolumeAccessor(Accessor):
 
     def __init__(
         self,
         config: DatabricksVolumeConfig,
-        client: Any | None = None,
+        client: DatabricksFilesClient,
     ) -> None:
         self.config = config
-        self._client = client
-
-    @property
-    def client(self) -> Any:
-        if self._client is None:
-            if WorkspaceClient is None or WorkspaceConfig is None:
-                raise ImportError("DatabricksVolumeResource requires the "
-                                  "'databricks' extra. Install with: "
-                                  "pip install mirage-ai[databricks]")
-            kwargs = {
-                "host": self.config.host,
-                "token": self.config.token,
-                "profile": self.config.profile,
-                "http_timeout_seconds": self.config.timeout,
-            }
-            sdk_config = WorkspaceConfig(**{
-                k: v
-                for k, v in kwargs.items() if v is not None
-            })
-            self._client = WorkspaceClient(config=sdk_config)
-        return self._client
-
-    @property
-    def files(self) -> Any:
-        return self.client.files
+        self.client = client