メインコンテンツまでスキップ

ストレージガイド

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

概要​

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

  • アップロード / ダウンロード — バイナリ・テキスト・ストリーム対応
  • プリサインド URL — 一時的なダウンロードリンクの生成
  • 一括ダウンロード — 複数ファイルの並列ダウンロード
  • パス管理 — StoragePaths による規約ベースのパス生成
  • メタデータ — ファイルにカスタムメタデータを付与

プロバイダーの選択​

config.tomlToml
[storage]
provider = "azure"

[storage.azure]
connection_string = "${AZURE_STORAGE_CONNECTION_STRING}"
container_name = "agent-files"
Python
from agenticstar_platform.storage import AzureBlobStorageClient, AzureBlobConfig

config = AzureBlobConfig(
    connection_string="DefaultEndpointsProtocol=https;...",
    container_name="agent-files",
)
client = AzureBlobStorageClient(config)

基本操作​

すべてのプロバイダーで同一の API を使用できます。

アップロード​

Python
# テキストファイルのアップロード
result = await client.upload(
    path="reports/2025/summary.md",
    data=b"# Monthly Summary\n...",
    content_type="text/markdown",
    metadata={"author": "agent-001", "version": "1.0"},
)
print(result.path)  # "reports/2025/summary.md"
print(result.size)  # バイト数

# ローカルファイルのアップロード
with open("output.pdf", "rb") as f:
    result = await client.upload(
        path="outputs/output.pdf",
        data=f.read(),
        content_type="application/pdf",
    )

ダウンロード​

Python
# ファイルのダウンロード
result = await client.download(path="reports/2025/summary.md")
print(result.data.decode("utf-8"))  # テキスト内容
print(result.metadata)              # カスタムメタデータ

# ローカルファイルに保存
with open("downloaded.pdf", "wb") as f:
    result = await client.download(path="outputs/output.pdf")
    f.write(result.data)

一覧取得​

Python
# ディレクトリ内のファイルを一覧
result = await client.list_objects(prefix="reports/2025/")
for obj in result.objects:
    print(f"{obj.path} ({obj.size} bytes, {obj.last_modified})")

削除​

Python
await client.delete(path="reports/2025/summary.md")

プリサインド URL​

Python
# 一時的なダウンロードリンクを生成（有効期限: 1時間）
url = await client.generate_presigned_url(
    path="outputs/output.pdf",
    expires_in=3600,
)
print(url)  # https://...signed URL

一括ダウンロード​

Python
# 複数ファイルを並列ダウンロード
paths = ["file1.pdf", "file2.pdf", "file3.pdf"]
results = await client.bulk_download(paths)
for path, result in zip(paths, results):
    print(f"{path}: {len(result.data)} bytes")

StoragePaths: パス規約​

StoragePaths クラスを使うと、プラットフォーム規約に従ったパスを生成できます。

Python
from agenticstar_platform.storage import StoragePaths

paths = StoragePaths(conversation_id="conv-001")

# プラン保存パス
plan_path = paths.plan_path("intent-plan.json")
# → "plans/conv-001/intent-plan.json"

# ファイルアップロードパス
upload_path = paths.upload_path("user-document.pdf")
# → "uploads/conv-001/user-document.pdf"

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

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

Python
from agenticstar_platform.storage import AzureBlobStorageClient
from agenticstar_platform.events import EventEmitter, EventType

async def generate_and_upload_report(emitter: EventEmitter, client: AzureBlobStorageClient):
    # レポートを生成
    report_content = await generate_report()

    # ストレージにアップロード
    result = await client.upload(
        path="reports/monthly-report.pdf",
        data=report_content,
        content_type="application/pdf",
    )

    # プリサインド URL を生成
    url = await client.generate_presigned_url(result.path, expires_in=3600)

    # ファイル生成イベントを発行
    await emitter.emit_event(
        EventType.FILE_CREATED,
        {"file_name": "monthly-report.pdf", "download_url": url},
    )

パスのサニタイズ​

Storage モジュールはパストラバーサル攻撃を自動で防止します。

Python
# 危険なパスは自動的にサニタイズされる
result = await client.upload(
    path="../../etc/passwd",  # → "etc/passwd" に正規化
    data=b"safe content",
)

次のステップ​

SDK API リファレンス — Storage Module

StorageClient / UploadResult / DownloadResult の完全仕様

ガイドを見る

デプロイガイド

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

ガイドを見る