Hello,
I’m building a file upload feature using Temporal where the frontend uploads directly to Google Cloud Storage (GCS) via a signed URL. My backend, which uses the Temporal Python SDK, is responsible for generating this URL.
The process is initiated by an agent event, generate_upload_url, which then calls an activity, generate_upload_url_activity, to create the signed URL. To create the GCS client, I need to read a service-account-file.json from the local filesystem.
Initially, I was reading the file directly within my agent’s event handler, which is part of the workflow logic. This resulted in the following error:
{
"signed_url": "",
"object_name": "",
"error": "Cannot access __builtins__.open from inside a workflow. If this is code from a module not used in a workflow or known to only be used deterministically from a workflow, mark the import as pass through."
}
Here is the relevant (and problematic) code from my agent:
src/agents/main_agent.py
@agent.event
async def generate_upload_url(self, input: GenerateUploadUrlInput) -> GenerateUploadUrlOutput:
"""Generate a signed URL for direct client upload to GCS"""
log.info("--- generate_upload_url event handler started ---")
try:
# ... (user_id and logging)
from temporalio import workflow
from src.activities.cloud_storage import generate_upload_url_activity, GenerateUploadUrlActivityInput
from datetime import timedelta
# THIS IS THE PROBLEMATIC LINE
with open("service-account-file.json", "r") as f:
service_account_json = f.read()
activity_input = GenerateUploadUrlActivityInput(
# ... (other params)
service_account_json=service_account_json,
)
result = await workflow.execute_activity(
generate_upload_url_activity,
activity_input,
schedule_to_close_timeout=timedelta(seconds=60),
)
# ... (return logic)
And here is the activity that uses the credentials:
src/activities/cloud_storage.py
@activity.defn
async def generate_upload_url_activity(input: GenerateUploadUrlActivityInput) -> GenerateUploadUrlActivityOutput:
"""Activity to generate signed URL for direct client upload"""
try:
# ... (logging)
# Load credentials from the input string
credentials_info = json.loads(input.service_account_json)
storage_client = storage.Client.from_service_account_info(credentials_info)
# ... (URL generation logic)
My proposed solution is to create a new, separate activity whose only job is to read the service-account-file.json and return its contents as a string. The agent would then call this new activity first and pass the result to the generate_upload_url_activity.
Is this the standard and most secure way to handle this? Or is there a better approach, such as loading the credentials into an environment variable when the worker starts and accessing it from the activity context?
Thank you for your guidance!