ストレージガイド

SDK の Storage モジュールを使って、Azure Blob Storage / Amazon S3 / Google Cloud Storage にファイルをアップロード・ダウンロードする方法を解説します。3 つのプロバイダーで統一された API を提供します。

概要

Storage モジュールは以下の機能を提供します。

アップロード / ダウンロード — ローカルファイルのアップロード・ダウンロード
署名付き URL — 一時的なダウンロードリンクの生成（get_object_url）
一括ダウンロード — プレフィックス指定による複数ファイルの並列ダウンロード
パス管理 — StoragePaths による規約ベースのパス生成
メタデータ — ファイルにカスタムメタデータを付与

Storage クライアントの upload_file / download_file はローカルファイルパスを入出力します（バイト列を直接渡す API ではありません）。生成したコンテンツをアップロードする場合は、いったんローカルファイルに書き出してから upload_file を呼び出してください。

プロバイダーの選択

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",  # Azure では container 名
    connection_string="DefaultEndpointsProtocol=https;...",
)
client = AzureBlobStorageClient(config)

# 辞書 / TOML からの読み込みも可能
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)

基本操作

すべてのプロバイダーで同一の API を使用できます。async with でクライアントを利用するとクローズが自動化されます。

Python
async with AzureBlobStorageClient(config) as client:
    await client.ensure_bucket_exists()
    # ... 各種操作 ...

アップロード

Python
# ローカルファイルをアップロード
result = await client.upload_file(
    file_path="/tmp/summary.md",                 # アップロード元のローカルパス
    object_name="reports/2025/summary.md",       # 保存先オブジェクト名（省略時はファイル名）
    metadata={"author": "agent-001", "version": "1.0"},
)

if result.success:
    print(result.object_name)   # "reports/2025/summary.md"
    print(result.object_url)    # 保存先 URL
    print(result.file_size)     # バイト数
    print(result.content_type)  # 自動判定された Content-Type
else:
    print(result.error)

upload_file はローカルファイルパスを受け取ります。文字列・バイト列を保存したい場合は、先にファイルへ書き出してください。

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",
)

ダウンロード

Python
# オブジェクトをローカルファイルにダウンロード
result = await client.download_file(
    object_name="reports/2025/summary.md",
    download_path="/tmp/summary.md",  # 保存先ローカルパス
)

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

一覧取得

Python
# プレフィックス配下のオブジェクトを一覧
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)  # 件数

存在確認

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

削除

Python
# 戻り値は bool（成功で True）
ok = await client.delete_object(object_name="reports/2025/summary.md")

署名付き URL

Python
# 一時的なダウンロードリンクを生成（expires_in 秒で失効）
url = await client.get_object_url(
    object_name="outputs/output.pdf",
    expires_in=3600,  # 1 時間。省略時は署名なしの URL を返す
)
print(url)  # https://...signed URL

expires_in を尊重して署名付き（期限付き）URL を返すのは S3 / GCS です。Azure Blob は現状 expires_in を無視し、常に非期限の公開 URL を返します（SAS 署名は将来拡張）。期限付きURLが必須の場合は S3 / GCS を使用してください。

一括ダウンロード（プレフィックス指定）

Python
# プレフィックス配下を並列ダウンロード
summary = await client.download_objects_by_prefix(
    prefix="reports/2025/",
    download_dir="/tmp/reports",
    max_concurrency=8,  # 環境変数 BLOB_DL_CONCURRENCY でも上書き可
)
print(summary["success_count"], summary["failed_count"], summary["skipped_count"])
for path in summary["downloaded"]:
    print(path)

StoragePaths: パス規約

StoragePaths はプラットフォーム規約に従ったオブジェクト名を生成する静的メソッド群です（インスタンス化は不要）。

Python
from agenticstar_platform.storage import StoragePaths

# プラン成果物: 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"

# ユーザーアップロード: 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"

# 所有者スコープ（マルチテナント汎用）: users/{owner_id}/{inner}
prefix = StoragePaths.owner_prefix("tenant-1", "plans/exec-1/files")
# → "users/tenant-1/plans/exec-1/files"

サブディレクトリ定数も提供されています。

定数	値
`StoragePaths.SUBDIR_FILES`	files
`StoragePaths.SUBDIR_DOWNLOADS`	downloads
`StoragePaths.SUBDIR_SCREENSHOTS`	screenshots
`StoragePaths.SUBDIR_DELIVERABLES`	deliverables

エージェントでの活用パターン

ファイル生成 → ダウンロードリンク配信

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):
    # レポートを生成し、いったんローカルに書き出す
    report_bytes = await generate_report()  # bytes を返す想定
    local_path = "/tmp/monthly-report.pdf"
    Path(local_path).write_bytes(report_bytes)

    # ストレージにアップロード
    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)

    # フロント（チャットUI）は file_name / download_url を読みません。描画されるのは:
    #   - FILE_CREATED       → metadata.files[]（ProgressFile 形: id/filename/fileType/size/url）
    #   - COMPLETION_SUCCESS → metadata.deliverable_files[]（本文の「成果物」DL、url 必須）
    # url は相対化のため object_url（無ければ object_name）を渡す
    file_url = result.object_url or result.object_name

    await emitter.emit_event(
        event_type=EventType.FILE_CREATED,
        message="月次レポートを生成しました",
        metadata={
            "title": "月次レポート",
            "status": "completed",
            "files": [{
                "id": result.object_name,
                "filename": "monthly-report.pdf",
                "fileType": "pdf",
                "size": result.file_size,
                "url": file_url,
                "source": "agent",
            }],
        },
    )

    # 完了応答の本文に「成果物」として出すには deliverable_files[] を付ける
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message="完了しました",
        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,
            }],
        },
    )

フロントが読むキーに注意

FILE_CREATED の metadata に file_name / download_url を入れても フロントは描画しません（アップロード自体が成功していてもファイルが UI に出ません）。進捗中のファイルは metadata.files[]（ProgressFile 形）、完了応答本文の成果物は metadata.deliverable_files[] を使ってください。url はフロント側で相対化されるため、object_url（同一コンテナ）または object_name を渡します。

入力添付ファイルの受け取り（チャットUI）

チャットUIからユーザーが添付したファイルは、リクエスト本文（execution_data.messages）には含まれず Blob ストレージ に保存されます。フロントエンドは添付を次の規約 prefix に置き、プラットフォーム（実行基盤）が対応する識別子を、エージェント Pod 起動時に環境変数（USER_ID / CONVERSATION_ID / MESSAGE_ID）として注入します。

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

エージェントは StoragePaths.input_uploads_prefix(...) で prefix を組み立て、 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"]]

注記

USER_ID / CONVERSATION_ID / MESSAGE_ID は、エージェント Pod 起動時にプラットフォームが環境変数として注入します（DB 接続情報などと同様、アプリ側で設定する必要はありません）。 execution_data.messages はテキスト（role / content）のみで、添付ファイルやその blob_path は含まれません。添付は必ず上記 Blob prefix から取得してください。MESSAGE_ID が無い起動経路では添付なしとして扱います。

フロント配信を前提とする成果物のパス規約

逆に、エージェントが生成したファイルをフロントのチャットUIから取得させる場合、フロントの proxy は plans/{conversation_id}/... や users/{owner}/... のようにパス先頭で会話/所有者を検証します。StoragePaths.plan_path(conversation_id, subdir, filename) のように、配信を前提とする成果物は会話/所有者スコープの prefix で保存してください（execution-id 等の任意 prefix だと所有者チェックで弾かれます）。

パストラバーサルの防止

StoragePaths.owner_prefix は owner_id / inner に含まれる .. セグメントや先頭スラッシュ・NUL を検出して ValueError を送出します。また download_objects_by_prefix はダウンロード先を resolve() + relative_to() で二重に検証し、ベースディレクトリ外への書き込みを防ぎます。

Python
# 不正な owner_id / inner は ValueError
StoragePaths.owner_prefix("tenant-1", "../../etc/passwd")  # → ValueError

次のステップ

SDK API リファレンス — Storage Module

StorageClient / UploadResult / DownloadResult の完全仕様

ガイドを見る

デプロイガイド

ストレージ接続情報の環境変数設定

ガイドを見る

概要​

プロバイダーの選択​

基本操作​

アップロード​

ダウンロード​

一覧取得​

存在確認​

削除​

署名付き URL​

一括ダウンロード（プレフィックス指定）​

StoragePaths: パス規約​

エージェントでの活用パターン​

ファイル生成 → ダウンロードリンク配信​

入力添付ファイルの受け取り（チャットUI）​

パストラバーサルの防止​

次のステップ​

概要