Add file upload/download support via TemporalAsyncFiles and TemporalAsyncClient

File operations (upload, download) now run entirely inside Temporal
activities using the real genai.Client on the worker side, avoiding
filesystem and OS module access in the workflow sandbox.

- Add TemporalAsyncFiles (_temporal_files.py): subclasses AsyncFiles,
  overriding upload() and download() to dispatch through activities.
  Supports str paths, os.PathLike, and io.IOBase inputs. Validates
  http_options for non-serializable fields before crossing the activity
  boundary.
- Add TemporalAsyncClient (_temporal_async_client.py): subclasses
  AsyncClient, wiring in TemporalAsyncFiles. gemini_client() now
  returns this instead of plain AsyncClient.
- Add gemini_files_upload and gemini_files_download activities.
- Include HTTP method in activity summaries (e.g. "POST files").

Author: JasonSteving99

temporalio/contrib/google_gemini_sdk/_gemini_activity.py

@@ -11,7 +11,7 @@
 from collections.abc import Sequence
 from typing import Any, Callable
 
-from google.genai import Client as GeminiClient
+from google.genai import Client as GeminiClient, types
 from google.genai.types import HttpOptions
 from google.genai.types import HttpResponse as SdkHttpResponse
 
@@ -20,6 +20,8 @@
     _GeminiApiRequest,
     _GeminiApiResponse,
     _GeminiApiStreamedResponse,
+    _GeminiDownloadFileRequest,
+    _GeminiUploadFileRequest,
 )
 
 
@@ -85,7 +87,34 @@ async def gemini_api_client_async_request_streamed(
                 )
             return _GeminiApiStreamedResponse(chunks=chunks)
 
+        @activity.defn(name="gemini_files_upload")
+        async def gemini_files_upload(
+            req: _GeminiUploadFileRequest,
+        ) -> types.File:
+            """Upload a file using the real genai.Client on the worker."""
+            if req.file_bytes is not None:
+                import io
+
+                file_arg: Any = io.BytesIO(req.file_bytes)
+            else:
+                file_arg = req.file_path
+
+            return await self._client.aio.files.upload(
+                file=file_arg, config=req.config
+            )
+
+        @activity.defn(name="gemini_files_download")
+        async def gemini_files_download(
+            req: _GeminiDownloadFileRequest,
+        ) -> bytes:
+            """Download a file using the real genai.Client on the worker."""
+            return await self._client.aio.files.download(
+                file=req.file, config=req.config
+            )
+
         return [
             gemini_api_client_async_request,
             gemini_api_client_async_request_streamed,
+            gemini_files_upload,
+            gemini_files_download,
         ]

temporalio/contrib/google_gemini_sdk/_temporal_api_client.py

@@ -1,9 +1,9 @@
 """Temporal-aware BaseApiClient that routes SDK calls through activities.
 
-This module provides ``TemporalApiClient``, a minimal ``BaseApiClient``
-subclass whose ``async_request()`` dispatches through a Temporal activity
-instead of making direct HTTP calls.  The real ``genai.Client`` with real
-credentials only exists on the worker side inside the activity.
+This module provides ``TemporalApiClient``, a ``BaseApiClient`` subclass
+whose HTTP methods dispatch through Temporal activities instead of making
+direct calls.  The real ``genai.Client`` with real credentials only exists
+on the worker side inside the activity.
 
 This ensures:
 - No credential fetching or refreshing happens in the workflow.
@@ -17,6 +17,7 @@
 from datetime import timedelta
 from typing import Any, Optional
 
+from google.genai import types
 from google.genai._api_client import BaseApiClient
 from google.genai.types import HttpOptions, HttpOptionsOrDict
 from google.genai.types import HttpResponse as SdkHttpResponse
@@ -55,6 +56,22 @@ class _SerializableHttpOptions(BaseModel):
 )
 
 
+def _validate_http_options(http_options: Optional[HttpOptions]) -> None:
+    """Raise if http_options contains non-serializable fields."""
+    if http_options is None:
+        return
+    bad_fields = [
+        f
+        for f in _REJECTED_HTTP_OPTION_FIELDS
+        if getattr(http_options, f, None) is not None
+    ]
+    if bad_fields:
+        raise ValueError(
+            f"http_options cannot include {bad_fields}. "
+            f"Configure custom HTTP clients at GeminiPlugin init instead."
+        )
+
+
 # ── async_request models ──────────────────────────────────────────────────
 
 class _GeminiApiRequest(BaseModel):
@@ -83,6 +100,28 @@ class _GeminiApiStreamedResponse(BaseModel):
     chunks: list[_GeminiApiResponse]
 
 
+# ── files upload/download models ──────────────────────────────────────────
+
+
+class _GeminiUploadFileRequest(BaseModel):
+    """Serializable activity input for a file upload.
+
+    For file path uploads the path is resolved on the worker.  For
+    in-memory uploads the raw bytes are sent across the activity boundary.
+    """
+
+    file_bytes: bytes | None = None
+    file_path: str | None = None
+    config: types.UploadFileConfig | None = None
+
+
+class _GeminiDownloadFileRequest(BaseModel):
+    """Serializable activity input for a file download."""
+
+    file: str
+    config: types.DownloadFileConfig | None = None
+
+
 class TemporalApiClient(BaseApiClient):
     """A ``BaseApiClient`` that routes all API calls through Temporal activities.
 
@@ -157,16 +196,7 @@ def _process_http_options(
         else:
             opts = HttpOptions.model_validate(http_options)
 
-        bad_fields = [
-            f
-            for f in _REJECTED_HTTP_OPTION_FIELDS
-            if getattr(opts, f, None) is not None
-        ]
-        if bad_fields:
-            raise ValueError(
-                f"Per-request http_options cannot include {bad_fields}. "
-                f"Configure custom HTTP clients at GeminiPlugin init instead."
-            )
+        _validate_http_options(opts)
 
         # timeout is owned by Temporal — apply it to the activity config
         # rather than forwarding to the underlying HTTP client.
@@ -201,7 +231,7 @@ async def async_request(
         config: ActivityConfig = {**self._activity_config}
         if "summary" not in config:
             # Default summary is the API path (e.g. "models/gemini-2.5-flash:generateContent").
-            config["summary"] = path
+            config["summary"] = f"{http_method.upper()} {path}"
         overrides = self._process_http_options(http_options, config)
 
         resp = await temporal_workflow.execute_activity(
@@ -252,7 +282,7 @@ async def async_request_streamed(
     ) -> Any:
         config: ActivityConfig = {**self._activity_config}
         if "summary" not in config:
-            config["summary"] = path
+            config["summary"] = f"{http_method.upper()} {path}"
         overrides = self._process_http_options(http_options, config)
 
         resp = await temporal_workflow.execute_activity(
@@ -273,29 +303,29 @@ async def _yield_chunks():
 
         return _yield_chunks()
 
-    # ── File upload/download (not supported) ─────────────────────────────
-    # These methods are only used by the SDK's file_search_stores module,
-    # which is not workflow-safe because it uses the OS module for file
-    # I/O.  Until we explicitly add support for the file_search_stores module
-    # operations as activities, raise early so callers get a clear error rather
-    # than a crash from missing HTTP clients.
+    # ── File upload/download ─────────────────────────────────────────────
+    # File operations are handled at a higher level by TemporalAsyncFiles
+    # (in _temporal_files.py), which dispatches the entire upload/download
+    # as a Temporal activity using the real client on the worker side.
+    # These internal BaseApiClient methods are not called in that path,
+    # so we raise here to catch any unexpected direct usage.
 
     def upload_file(self, *args: Any, **kwargs: Any) -> Any:
         raise NotImplementedError(
-            "File uploads are not yet supported in the Temporal Gemini integration."
+            "Use client.files.upload() instead of the internal upload_file() method."
         )
 
     async def async_upload_file(self, *args: Any, **kwargs: Any) -> Any:
         raise NotImplementedError(
-            "File uploads are not yet supported in the Temporal Gemini integration."
+            "Use client.files.upload() instead of the internal async_upload_file() method."
         )
 
     def download_file(self, *args: Any, **kwargs: Any) -> Any:
         raise NotImplementedError(
-            "File downloads are not yet supported in the Temporal Gemini integration."
+            "Use client.files.download() instead of the internal download_file() method."
         )
 
     async def async_download_file(self, *args: Any, **kwargs: Any) -> Any:
         raise NotImplementedError(
-            "File downloads are not yet supported in the Temporal Gemini integration."
+            "Use client.files.download() instead of the internal async_download_file() method."
         )

temporalio/contrib/google_gemini_sdk/_temporal_async_client.py

@@ -0,0 +1,36 @@
+"""Temporal-aware AsyncClient shim.
+
+``TemporalAsyncClient`` is an ``AsyncClient`` subclass that wires up
+``TemporalAsyncFiles`` in place of the default ``AsyncFiles``.
+"""
+
+from __future__ import annotations
+
+from google.genai.client import AsyncClient
+
+from temporalio.workflow import ActivityConfig
+
+from temporalio.contrib.google_gemini_sdk._temporal_api_client import (
+    TemporalApiClient,
+)
+from temporalio.contrib.google_gemini_sdk._temporal_files import (
+    TemporalAsyncFiles,
+)
+
+
+class TemporalAsyncClient(AsyncClient):
+    """``AsyncClient`` subclass that uses Temporal-aware modules.
+
+    Replaces ``AsyncFiles`` with ``TemporalAsyncFiles`` so that file
+    upload/download operations run entirely inside Temporal activities.
+    Other modules (models, caches, etc.) are inherited unchanged and
+    work through ``TemporalApiClient``'s activity-backed HTTP methods.
+    """
+
+    def __init__(
+        self,
+        api_client: TemporalApiClient,
+        activity_config: ActivityConfig | None = None,
+    ) -> None:
+        super().__init__(api_client)
+        self._files = TemporalAsyncFiles(api_client, activity_config)

temporalio/contrib/google_gemini_sdk/_temporal_files.py

@@ -0,0 +1,125 @@
+"""Temporal-aware AsyncFiles shim.
+
+``TemporalAsyncFiles`` is an ``AsyncFiles`` subclass whose ``upload``
+and ``download`` methods dispatch through Temporal activities so the
+entire file operation (including filesystem access) runs on the
+activity worker.
+"""
+
+from __future__ import annotations
+
+import io
+from datetime import timedelta
+from typing import Optional, Union
+
+from google.genai import types
+from google.genai.files import AsyncFiles
+from google.genai.types import HttpOptions
+
+from temporalio import workflow as temporal_workflow
+from temporalio.workflow import ActivityConfig
+
+from temporalio.contrib.google_gemini_sdk._temporal_api_client import (
+    TemporalApiClient,
+    _GeminiDownloadFileRequest,
+    _GeminiUploadFileRequest,
+    _validate_http_options,
+)
+
+
+class TemporalAsyncFiles(AsyncFiles):
+    """``AsyncFiles`` subclass that routes ``upload`` and ``download`` through activities.
+
+    The entire file operation — including filesystem access, resumable
+    upload negotiation, and chunked transfer — runs inside a Temporal
+    activity on the worker.  ``get``, ``delete``, and ``list`` are
+    inherited from ``AsyncFiles`` and already work through the
+    ``TemporalApiClient``'s ``async_request`` activity.
+    """
+
+    def __init__(
+        self,
+        api_client: TemporalApiClient,
+        activity_config: ActivityConfig | None = None,
+    ) -> None:
+        super().__init__(api_client)
+        self._activity_config = activity_config or ActivityConfig(
+            start_to_close_timeout=timedelta(seconds=60),
+        )
+
+    async def upload(
+        self,
+        *,
+        file: Union[str, "os.PathLike[str]", io.IOBase],
+        config: Optional[types.UploadFileConfigOrDict] = None,
+    ) -> types.File:
+        """Upload a file via a Temporal activity.
+
+        Accepts a file path (resolved on the worker), ``os.PathLike``, or
+        an ``io.IOBase`` (bytes sent across the activity boundary).
+        """
+        act_config: ActivityConfig = {**self._activity_config}
+        if "summary" not in act_config:
+            act_config["summary"] = "files.upload"
+
+        upload_config = None
+        if config is not None:
+            if isinstance(config, dict):
+                upload_config = types.UploadFileConfig.model_validate(config)
+            else:
+                upload_config = config
+            _validate_http_options(upload_config.http_options)
+
+        if isinstance(file, io.IOBase):
+            req = _GeminiUploadFileRequest(
+                file_bytes=file.read(), config=upload_config
+            )
+        elif isinstance(file, str):
+            req = _GeminiUploadFileRequest(
+                file_path=file, config=upload_config
+            )
+        else:
+            # os.PathLike — convert via __fspath__() to avoid importing os
+            req = _GeminiUploadFileRequest(
+                file_path=file.__fspath__(), config=upload_config
+            )
+
+        return await temporal_workflow.execute_activity(
+            "gemini_files_upload",
+            req,
+            result_type=types.File,
+            **act_config,
+        )
+
+    async def download(
+        self,
+        *,
+        file: Union[str, types.File],
+        config: Optional[types.DownloadFileConfigOrDict] = None,
+    ) -> bytes:
+        """Download a file via a Temporal activity."""
+        act_config: ActivityConfig = {**self._activity_config}
+        if "summary" not in act_config:
+            act_config["summary"] = "files.download"
+
+        download_config = None
+        if config is not None:
+            if isinstance(config, dict):
+                download_config = types.DownloadFileConfig.model_validate(config)
+            else:
+                download_config = config
+            _validate_http_options(download_config.http_options)
+
+        if isinstance(file, types.File):
+            if not file.name:
+                raise ValueError("File object must have a name to download.")
+            file_name = file.name
+        else:
+            file_name = file
+
+        return await temporal_workflow.execute_activity(
+            "gemini_files_download",
+            _GeminiDownloadFileRequest(file=file_name, config=download_config),
+            result_type=bytes,
+            **act_config,
+        )

temporalio/contrib/google_gemini_sdk/workflow.py

@@ -24,6 +24,9 @@
 from temporalio.contrib.google_gemini_sdk._temporal_api_client import (
     TemporalApiClient,
 )
+from temporalio.contrib.google_gemini_sdk._temporal_async_client import (
+    TemporalAsyncClient,
+)
 from temporalio.exceptions import ApplicationError
 from temporalio.workflow import ActivityConfig
 
@@ -181,4 +184,4 @@ async def run(self, query: str) -> str:
         location=location,
         activity_config=activity_config,
     )
-    return AsyncClient(temporal_api_client)
+    return TemporalAsyncClient(temporal_api_client, activity_config)