Storage Guide

This guide explains how to use the SDK's Storage module to upload and download files to Azure Blob Storage / Amazon S3 / Google Cloud Storage. A unified API is provided across all 3 providers.

Overview

The Storage module provides the following features:

Upload / Download — Upload and download local files
Signed URLs — Generate temporary download links (get_object_url)
Bulk Download — Parallel download of objects under a prefix
Path Management — Convention-based path generation with StoragePaths
Metadata — Attach custom metadata to files

The storage client's upload_file / download_file take a local file path as input/output (they do not accept raw bytes directly). To upload generated content, write it to a local file first, then call upload_file.

Provider Selection

Azure Blob Storage
Amazon S3
Google Cloud Storage

config.tomlToml
[storage.azure]
bucket_name = "agent-files"
connection_string = "${AZURE_STORAGE_CONNECTION_STRING}"

Python
from agenticstar_platform.storage import AzureBlobStorageClient, AzureBlobConfig

config = AzureBlobConfig(
    bucket_name="agent-files",  # container name on Azure
    connection_string="DefaultEndpointsProtocol=https;...",
)
client = AzureBlobStorageClient(config)

# Loading from a dict / TOML is also supported
config = AzureBlobConfig.from_dict({
    "bucket_name": "agent-files",
    "connection_string": "DefaultEndpointsProtocol=https;...",
})

config.tomlToml
[storage.s3]
bucket_name = "agent-files"
region_name = "ap-northeast-1"
aws_access_key_id = "${AWS_ACCESS_KEY_ID}"
aws_secret_access_key = "${AWS_SECRET_ACCESS_KEY}"

Python
from agenticstar_platform.storage import S3StorageClient, S3Config

config = S3Config(
    bucket_name="agent-files",
    aws_access_key_id="AKIA...",
    aws_secret_access_key="...",
    region_name="ap-northeast-1",
)
client = S3StorageClient(config)

config.tomlToml
[storage.gcs]
bucket_name = "agent-files"
project_id = "your-project"
credentials_path = "${GOOGLE_APPLICATION_CREDENTIALS}"

Python
from agenticstar_platform.storage import GCSStorageClient, GCSConfig

config = GCSConfig(
    bucket_name="agent-files",
    project_id="your-project",
    credentials_path="/path/to/service-account.json",
)
client = GCSStorageClient(config)

Basic Operations

The same API can be used across all providers. Using the client with async with closes it automatically.

Python
async with AzureBlobStorageClient(config) as client:
    await client.ensure_bucket_exists()
    # ... operations ...

Upload

Python
# Upload a local file
result = await client.upload_file(
    file_path="/tmp/summary.md",                 # source local path
    object_name="reports/2025/summary.md",       # destination object name (defaults to file name)
    metadata={"author": "agent-001", "version": "1.0"},
)

if result.success:
    print(result.object_name)   # "reports/2025/summary.md"
    print(result.object_url)    # destination URL
    print(result.file_size)     # bytes
    print(result.content_type)  # auto-detected Content-Type
else:
    print(result.error)

upload_file takes a local file path. To store a string or bytes, write it to a file first.

Python
from pathlib import Path

Path("/tmp/summary.md").write_text("# Monthly Summary\n...")
result = await client.upload_file(
    file_path="/tmp/summary.md",
    object_name="reports/2025/summary.md",
)

Download

Python
# Download an object to a local file
result = await client.download_file(
    object_name="reports/2025/summary.md",
    download_path="/tmp/summary.md",  # destination local path
)

if result.success:
    print(result.local_path)  # "/tmp/summary.md"
    print(result.file_size)   # bytes
    content = open(result.local_path, encoding="utf-8").read()

List Objects

Python
# List objects under a prefix
result = await client.list_objects(prefix="reports/2025/", max_results=100)
for obj in result.objects:
    print(f"{obj.name} ({obj.size} bytes, {obj.last_modified})")
print(result.count)  # number of objects

Existence Check

Python
exists = await client.object_exists(object_name="reports/2025/summary.md")

Delete

Python
# Returns bool (True on success)
ok = await client.delete_object(object_name="reports/2025/summary.md")

Signed URLs

Python
# Generate a temporary download link (expires after expires_in seconds)
url = await client.get_object_url(
    object_name="outputs/output.pdf",
    expires_in=3600,  # 1 hour. When omitted, returns an unsigned URL
)
print(url)  # https://...signed URL

S3 / GCS honor expires_in and return a signed (expiring) URL. Azure Blob currently ignores expires_in and always returns a non-expiring public URL (SAS signing is a future enhancement). If you need expiring URLs, use S3 / GCS.

Bulk Download (by prefix)

Python
# Download everything under a prefix in parallel
summary = await client.download_objects_by_prefix(
    prefix="reports/2025/",
    download_dir="/tmp/reports",
    max_concurrency=8,  # also overridable via the BLOB_DL_CONCURRENCY env var
)
print(summary["success_count"], summary["failed_count"], summary["skipped_count"])
for path in summary["downloaded"]:
    print(path)

StoragePaths: Path Conventions

StoragePaths is a set of static methods that generate object names following platform conventions (no instantiation needed).

Python
from agenticstar_platform.storage import StoragePaths

# Plan artifacts: plans/{plan_id}/{subdir}/{filename}
prefix = StoragePaths.plan_prefix("exec-123", "files")
# -> "plans/exec-123/files"
path = StoragePaths.plan_path("exec-123", "files", "report.pdf")
# -> "plans/exec-123/files/report.pdf"

# User uploads: uploads/{conversation_id}/{message_id}/{filename}
prefix = StoragePaths.upload_prefix("conv-456", "msg-789")
# -> "uploads/conv-456/msg-789"
path = StoragePaths.upload_path("conv-456", "msg-789", "user-document.pdf")
# -> "uploads/conv-456/msg-789/user-document.pdf"

# Owner-scoped (multi-tenant generic): users/{owner_id}/{inner}
prefix = StoragePaths.owner_prefix("tenant-1", "plans/exec-1/files")
# -> "users/tenant-1/plans/exec-1/files"

Subdirectory constants are also provided.

Constant	Value
`StoragePaths.SUBDIR_FILES`	files
`StoragePaths.SUBDIR_DOWNLOADS`	downloads
`StoragePaths.SUBDIR_SCREENSHOTS`	screenshots
`StoragePaths.SUBDIR_DELIVERABLES`	deliverables

Usage Patterns in Agents

File Generation -> Download Link Delivery

Python
from pathlib import Path
from agenticstar_platform.storage import AzureBlobStorageClient, StoragePaths
from agenticstar_platform.events import EventEmitter, EventType

async def generate_and_upload_report(emitter: EventEmitter, client: AzureBlobStorageClient):
    # Generate the report and write it to a local file first
    report_bytes = await generate_report()  # assumed to return bytes
    local_path = "/tmp/monthly-report.pdf"
    Path(local_path).write_bytes(report_bytes)

    # Upload to storage
    object_name = StoragePaths.plan_path("exec-123", StoragePaths.SUBDIR_DELIVERABLES, "monthly-report.pdf")
    result = await client.upload_file(file_path=local_path, object_name=object_name)

    # The front-end (chat UI) does NOT read file_name / download_url. It renders:
    #   - FILE_CREATED       → metadata.files[]            (ProgressFile: id/filename/fileType/size/url)
    #   - COMPLETION_SUCCESS → metadata.deliverable_files[] (the "deliverables" download in the reply body, url required)
    # Pass object_url (or object_name) for url so the front-end can relativize it.
    file_url = result.object_url or result.object_name

    await emitter.emit_event(
        event_type=EventType.FILE_CREATED,
        message="Monthly report generated",
        metadata={
            "title": "Monthly report",
            "status": "completed",
            "files": [{
                "id": result.object_name,
                "filename": "monthly-report.pdf",
                "fileType": "pdf",
                "size": result.file_size,
                "url": file_url,
                "source": "agent",
            }],
        },
    )

    # To surface the file as a "deliverable" in the completion reply body, add deliverable_files[]
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message="Done",
        metadata={
            "deliverable_files": [{
                "filename": "monthly-report.pdf",
                "url": file_url,
                "relative_path": result.object_name,
                "size_bytes": result.file_size,
                "content_type": "application/pdf",
                "source": "agent",
                "is_primary": True,
            }],
        },
    )

Mind the keys the front-end reads

Putting file_name / download_url in the FILE_CREATED metadata will not render in the UI (the file stays invisible even though the upload succeeded). Use metadata.files[] (ProgressFile shape) for in-progress files, and metadata.deliverable_files[] for deliverables in the completion reply body. The front-end relativizes url, so pass object_url (same container) or object_name.

Receiving Input Attachments (Chat UI)

Files a user attaches in the chat UI are not included in the request body (execution_data.messages); they are stored in Blob storage. The front-end saves attachments under the following convention prefix, and the platform (the execution runtime) injects the matching identifiers as environment variables when the agent Pod starts (USER_ID / CONVERSATION_ID / MESSAGE_ID).

users/{USER_ID}/uploads/{CONVERSATION_ID}/{MESSAGE_ID}/{filename}

Build the prefix with StoragePaths.input_uploads_prefix(...) and fetch the files locally with download_objects_by_prefix.

Python
import os
from agenticstar_platform.storage import AzureBlobStorageClient, StoragePaths

async def fetch_input_attachments(client: AzureBlobStorageClient) -> list[str]:
    prefix = StoragePaths.input_uploads_prefix(
        os.environ["USER_ID"],
        os.environ["CONVERSATION_ID"],
        os.environ["MESSAGE_ID"],
    )  # → users/{USER_ID}/uploads/{conv}/{msg}
    summary = await client.download_objects_by_prefix(prefix, download_dir="/tmp/inputs")
    # summary["downloaded"] = [{"object_name", "local_path", "file_size"}, ...]
    return [d["local_path"] for d in summary["downloaded"]]

note

USER_ID / CONVERSATION_ID / MESSAGE_ID are injected by the platform as environment variables when the agent Pod starts (like the DB connection settings — you do not set them in your app). execution_data.messages contains only text (role / content) — no attachments or their blob_path. Always fetch attachments from the Blob prefix above. If MESSAGE_ID is absent in a given startup path, treat the turn as having no attachments.

Path convention for front-served deliverables

Conversely, to let the chat UI fetch a file your agent generated, note that the front-end proxy validates ownership by the leading path segment (e.g. plans/{conversation_id}/... or users/{owner}/...). Save front-served deliverables under a conversation/owner-scoped prefix (e.g. StoragePaths.plan_path(conversation_id, subdir, filename)); an arbitrary prefix such as an execution-id will be rejected by the ownership check.

Path Traversal Protection

StoragePaths.owner_prefix rejects .. segments, leading slashes, and NUL characters in owner_id / inner by raising ValueError. download_objects_by_prefix also validates the download destination with resolve() + relative_to() to prevent writes outside the base directory.

Python
# Invalid owner_id / inner raises ValueError
StoragePaths.owner_prefix("tenant-1", "../../etc/passwd")  # -> ValueError

Next Steps

SDK API Reference — Storage Module

Complete specifications for StorageClient / UploadResult / DownloadResult

ガイドを見る

Deploy Guide

Setting up storage connection environment variables

ガイドを見る

Overview​

Provider Selection​

Basic Operations​

Upload​

Download​

List Objects​

Existence Check​

Delete​

Signed URLs​

Bulk Download (by prefix)​

StoragePaths: Path Conventions​

Usage Patterns in Agents​

File Generation -> Download Link Delivery​

Receiving Input Attachments (Chat UI)​

Path Traversal Protection​

Next Steps​