bria-fibo-image

Bria FIBO による画像生成

ユーザーがローカルの重みを明示的に求めている場合を除き、Bria のファーストパーティがホストする v2 API を使用してください。ローカルの重みに関するライセンスと、ホスト型 API の利用規約は分けて扱ってください。ファーストパーティへのリクエストを fal.ai、Replicate、その他のゲートウェイ経由で送信してはいけません。

根拠を示す用語

重要な事実を提示するときは、次のラベルを使用してください。

• 文書化された事実 - 引用したファーストパーティの API スキーマ、モデルカード、論文、価格ページ、または法的文書に記載されている内容です。
• プロバイダーの主張 - Bria による性能、トレーニングデータ、安全性、権利、コンプライアンス、または補償に関する表明です。独立して検証された事実として言い換えてはいけません。
• 運用上の目安 - このスキルの保守的な推奨事項であり、Bria の制限や保証ではありません。
• 不明 - 確認した公開ファーストパーティ資料では確立されていない内容です。それを前提にする前に、Bria に問い合わせるか、ユーザーが締結した契約を確認してください。

このスキルに記載された変動し得る事実と出典は、2026-07-10 に確認しました。本番環境への導入や調達を決定する前に、スキーマ、価格、プランの制限、モデルカード、および適用される規約を再確認してください。

まず判断記録を作成する

コードを書く前、または作業を提出する前に、次の内容を記録してください。

1. 目的：generate、1 つの参照画像からの inspire、保存済みの構造化プロンプトを使った refine、または構造化プロンプトとシードを使った recreate。
2. 提供形態：ホスト型 Standard、ホスト型 Lite、またはローカル／オンプレミスでの評価。
3. 入力：プロンプト、参照画像 1 枚、保存済みの structured_prompt、シード、権利／同意の出所情報。
4. 出力：アスペクト比、png または jpeg、Standard のみで指定可能な 1MP または 4MP、生成数、期限、受け入れ基準。
5. ガバナンス：アカウント／プラン、許可される用途、個人データ／機密データの分類、地域、保持要件、現在の契約／DPA。
6. 費用：無料枠の消費量または予想料金、呼び出しの最大数、最大 USD 額、送信を承認する担当者。

権利、同意、またはデータ分類が明確でない場合は、作業をオフラインのままにして確認してください。公開画像の URL は、使用許可の根拠にはなりません。

正確なルートを選択する

文書化された事実（v2 OpenAPI）： ベース URL は https://engine.prod.bria-api.com/v2 です。認証には api_token リクエストヘッダーを使用します。Bria の LLM ドキュメントでは、エージェントのクライアントに User-Agent: BriaPlatform/APIdocs/LLMsAgent の送信も求めていますが、この要件はエンドポイントの OpenAPI には記載されていません。このヘッダーを含めてください。

ニーズ	ルート	ブリッジ／モデルの動作	重要なリクエストフィールド
最高品質の画像	POST /image/generate	Bria の説明によると、VLM ブリッジとして Gemini 2.5 Flash を使用し、その後フル FIBO を使用します	resolution、negative_prompt、steps_num を利用できます
高速／低コストの反復	POST /image/generate/lite	Bria の説明によると、オープンソースの FIBO-VLM ブリッジと FIBO Lite を使用します	現行スキーマには resolution、negative_prompt、steps_num がありません
構造化プロンプトのみ	POST /structured_prompt/generate	Standard VLM ブリッジを使用します。画像は返しません	画像出力の制御項目はありません
Lite の構造化プロンプトのみ	POST /structured_prompt/generate/lite	Lite VLM ブリッジを使用します。画像は返しません	画像出力の制御項目はありません

プロバイダーの主張： Standard は解釈品質を最大化する用途、FIBO Lite は速度と低コストを重視する用途として位置付けられています。FIBO Lite のモデルカードでは、品質とのトレードオフがある、蒸留された少ステップのバリアントと説明されています。本番環境の既定値を選ぶ前に、ユーザー独自のプロンプトセットでベンチマークを実施してください。

プライバシーへの影響： Standard のドキュメントでは、プロンプト／画像から JSON への変換に Gemini 2.5 Flash が関与すると記載されています。確認した公開ページでは、ホスト環境における正確なデータ経路、Google のサブプロセッサー／規約、リージョン、保持期間、またはゼロデータ保持が適用されるかどうかは明らかになっていません。締結済みの契約／DPA とアーキテクチャによってこれらの疑問が解消されるまでは、機密情報、規制対象情報、個人情報、または秘密情報を送信しないでください。Lite のホスト型ルートでもデータは Bria に送信されます。「ローカル／オンプレミス」での利用には、別途ライセンスされたデプロイが必要です。

現在のリクエスト仕様

2 つの画像生成ルートでは、以下の相互排他的な組み合わせから必ず 1 つを指定してください。

• prompt
• images（配列、最大 1 件）
• images と prompt
• structured_prompt
• structured_prompt と prompt

2 つの構造化プロンプトルートについて、現在のファーストパーティスキーマで許可されているのは、prompt、images、images と prompt、またはブラッシュアップ用の prompt を伴う structured_prompt です。structured_prompt 単独は許可されていません。4 つのエンドポイントを 1 つの仕様にまとめず、選択したルートのスキーマに従ってください。

再現またはブラッシュアップする場合は、保存した構造化プロンプトとともに保存した seed を指定してください。structured_prompt はネストされたリクエストオブジェクトではなく、JSON を含む文字列です。JSON オブジェクトとして解析できることを検証しつつ、来歴情報として返された文字列をそのまま保持してください。

共通の画像生成フィールド（ドキュメント記載）：

• model_version：現在の列挙値は FIBO のみです。フィールドが暗黙に省略されないよう送信してください。ただし、変更されないバックエンドのリビジョンを固定するものではありません。
• aspect_ratio：1:1、2:3、3:2、3:4、4:3、4:5、5:4、9:16、または 16:9。
• seed：整数です。公開 OpenAPI には最小値・最大値が記載されていません。範囲を独自に設定しないでください。
• sync：デフォルトは false です。本番環境では非同期処理を使用してください。
• webhook_url：非同期処理用の任意の HTTPS コールバックです。
• output_type：png または jpeg。デフォルトは png です。
• ip_signal：デフォルトは false です。ワークフローで IP に関する警告が役立つ場合は true に設定してください。
• prompt_content_moderation、visual_input_content_moderation、visual_output_content_moderation：現在のスキーマのデフォルトは true です。このスキルでは常に true を送信し、無効化する切り替え機能は公開しません。

標準画像専用フィールド：

• resolution：1MP または 4MP。デフォルトは 1MP です。Bria によると、4MP では約 30 秒追加されます。レイテンシはプロバイダーの推定値として扱ってください。
• steps_num：35～50 の整数。デフォルトは 50 です。
• negative_prompt：任意の文字列です。

現在の Lite 画像スキーマは、これら 3 つの Standard 専用フィールドを受け付けません。現在の構造化プロンプトルートは画像出力フィールドを受け付けません。サービスが無視することを期待せず、未知のフィールドはローカルで拒否してください。

OpenAPI では、画像は公開 URL または data:image/...;base64, の接頭辞を付けない生の base64 として受け付けられ、JPEG/JPG/PNG/WEBP が列挙されています。FIBO 固有のバイト数、ピクセル数、URL 取得、アニメーション、メタデータ、または展開に関する制限は記載されていません。クライアント側の上限は運用上の目安であり、プロバイダーの制限ではありません。

制御可能な入力を組み立てる

テキストから最初の画像を生成する場合は、監査可能なクリエイティブ仕様としてプロンプトを作成してください。

Purpose: ecommerce hero image for a ceramic pour-over set.
Subject: one matte cobalt dripper and matching carafe; accurate circular rim; no people.
Composition: centered three-quarter view, object fills 65% of frame, clear negative space at upper right.
Environment: warm gray seamless studio sweep.
Lighting: large softbox camera-left, gentle rim light, grounded soft shadow.
Camera: eye-level, 70 mm product-photography perspective, all product edges sharp.
Palette/material: cobalt #164B8C ceramic, neutral background, physically plausible glaze.
Text: none. Logos/brands: none.
Acceptance: one product set, no duplicate handles, no floating parts, no clipped silhouette.

タイポグラフィでは、画像に表示する文字を引用符で囲んで正確に指定し、受け入れ基準でも同じ文字を繰り返してください。生成されたテキストは、必ず人が校正してください。

参照画像を使う場合：

• その画像を送信する権利をユーザーが所有しているか、使用許諾を得ていることを確認してください。人物が含まれる場合は、プライバシーおよび肖像権に関する許可も確認してください。
• 画像は 1 枚だけ使用してください。ポリシー上問題がない場合は、不要な EXIF/GPS 情報を削除しつつ、未加工の原本を別途保存してください。
• 引き継いでよい属性（色調、照明、大まかな構図）と、引き継いではならない属性（個人の識別情報、ロゴ、特徴的な保護対象キャラクター、透かし）を明示してください。
• 「着想を得る」ことを、著作権上の許可や類似しないことの保証として説明しないでください。

反復生成では、返された構造化プロンプトとシードを保存してください。「カメラアングルだけをローアングルの 3/4 ビューに変更し、被写体の数、色調、照明、レイアウトは維持する」のように、変更を 1 つだけ指定してください。確認のために解析済み JSON の差分を取りますが、元の文字列もバイト単位でそのまま保持してください。プロバイダーの主張： Bria は、構造化プロンプトとシードによって決定論的かつ正確に再現し、変更を 1 つに限定してブラッシュアップできると説明しています。運用上の目安： ホスト型ルートでは変更されないモデルリビジョンが公開されていないため、ビット単位の同一性ではなく、再現性を意図したものとして扱ってください。独自の回帰テストセットで知覚的・意味的な差異を測定してください。

承認とコストのポリシー

プロンプト、ペイロード、ドライラン、バリデーター、ローカルテスト用フィクスチャの準備は許可されています。ホスト型の生成または構造化プロンプトの送信ではクォータを消費し、料金が発生する場合があります。

実際のリクエストを送信する前に、次の4項目をすべて満たす必要があります。

1. SEND_BRIA_REQUEST=1
2. APPROVE_BRIA_PAID_CALL=YES（無料トライアルのクォータを使用する場合も必要です）
3. 日付付きの1回あたりの見積もり額以上の数値を指定した BRIA_MAX_COST_USD
4. レビュー済みのドライランからコピーした BRIA_APPROVED_REQUEST_SHA256。これにより、承認対象を正確なルート、ペイロード／参照データのバイト列、出力ホストの許可リスト、管理対象の出力ディレクトリ、コスト、上限、クライアントジョブキーに紐付けます

ルート、呼び出し回数、推定最大額、リトライによって2つ目の課金対象ジョブが作成される可能性があるかどうかも明示してください。推測で複数のバリエーションを試してはいけません。クライアントの接続失敗を、Bria がジョブを作成しなかった証拠とみなしてはいけません。

日付付きの料金ページ情報（2026-07-10）： Development では Fibo が $0.03/image、Fibo Lite が $0.02/image、Fibo Structured Prompt が $0.02/call と記載されています。Business と Enterprise の料金は個別見積もりです。また、このページでは無料生成100回も案内されています。税金、クレジット、交渉済み料金、失敗したジョブへの課金、4MP の扱い、Structured Prompt Lite の料金は、公開されているスキーマ／ページでは完全には明確になっていません。アカウントコンソールと締結済みの契約が優先されます。

この例では、アカウントまたは契約で確認した BRIA_ESTIMATED_CALL_USD を設定するまで、4MP Standard の実呼び出しまたは Lite の構造化プロンプト呼び出しをブロックします。このオペレーター指定の見積もり額は BRIA_MAX_COST_USD 以下でなければなりません。これは Bria が公開している料金として示されたものではありません。

公開されているレート制限には不一致があります。API overview では、Free Trial が毎分 10 件、Starter が 60 件、Pro/Enterprise が毎分 1000 件のリクエストと記載されています（エンドポイントごと）。一方、料金比較ページでは毎分 10 件、60 件、150 件のアクションと、Enterprise の個別設定のスループットが示されています。どちらも契約上の確定値としてハードコードしないでください。レスポンスヘッダーがある場合は確認し、429 に従い、アカウントで現在有効な制限を確認してください。

ホスト型の実行例：まずはドライラン

この単一ファイルの Python 3.11 以降のクライアントは、API 呼び出しに標準ライブラリのみを使用し、ローカルの画像検証には Pillow を使用します（python -m pip install Pillow）。ファーストパーティの4つのルートすべて、プロンプト／参照画像／リファイン入力、非同期ポーリング、厳格な URL ポリシー、読み取りサイズの上限、アトミック書き込み、来歴マニフェストに対応しています。デフォルトでは送信しません。

bria_fibo.py にコピーしてください。

from __future__ import annotations

import base64
import hashlib
import ipaddress
import json
import math
import os
import random
import socket
import tempfile
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

BASE = "https://engine.prod.bria-api.com/v2"
API_HOST = "engine.prod.bria-api.com"
USER_AGENT = "BriaPlatform/APIdocs/LLMsAgent"
ROUTES: dict[str, float | None] = {
    "/image/generate": 0.03,
    "/image/generate/lite": 0.02,
    "/structured_prompt/generate": 0.02,
    "/structured_prompt/generate/lite": None,  # public price page is not explicit
}
RATIOS = {"1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9"}
MAX_JSON_BYTES = 16 * 1024 * 1024       # client heuristic, not a Bria limit
MAX_INPUT_BYTES = 12 * 1024 * 1024      # client heuristic, not a FIBO limit
MAX_OUTPUT_BYTES = 64 * 1024 * 1024     # client heuristic
MAX_PIXELS = 40_000_000                  # client heuristic
POLL_TIMEOUT_S = 600


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RuntimeError(f"redirect refused ({code}); validate the new URL out of band")


class BriaHTTPError(RuntimeError):
    def __init__(self, status_code: int, message: str, retry_after: str | None = None):
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after


class AmbiguousTransportError(RuntimeError):
    pass


OPENER = build_opener(NoRedirect)


def env_bool(name: str) -> bool:
    return os.getenv(name, "") == "1"


def exact_https(url: str, allowed_hosts: set[str], path_prefix: str | None = None) -> str:
    p = urlsplit(url)
    if p.scheme != "https" or not p.hostname or p.username or p.password:
        raise ValueError("URL must be credential-free HTTPS")
    if p.port not in (None, 443) or p.hostname.lower() not in allowed_hosts:
        raise ValueError("URL host/port is not explicitly allowed")
    if path_prefix is not None and not p.path.startswith(path_prefix):
        raise ValueError("URL path is outside the allowed prefix")
    for item in socket.getaddrinfo(p.hostname, 443, type=socket.SOCK_STREAM):
        addr = ipaddress.ip_address(item[4][0])
        if not addr.is_global:
            raise ValueError("URL resolved to a non-public address")
    return url


def read_json_response(resp, cap: int = MAX_JSON_BYTES) -> dict[str, Any]:
    length = resp.headers.get("Content-Length")
    if length and int(length) > cap:
        raise RuntimeError("JSON response exceeds client cap")
    raw = resp.read(cap + 1)
    if len(raw) > cap:
        raise RuntimeError("JSON response exceeds client cap")
    ctype = resp.headers.get_content_type()
    if ctype != "application/json":
        raise RuntimeError(f"unexpected response Content-Type: {ctype}")
    value = json.loads(raw)
    if not isinstance(value, dict):
        raise RuntimeError("expected a JSON object")
    return value


def api_json(
    method: str,
    url: str,
    token: str,
    body: dict[str, Any] | None = None,
    timeout: float = 60,
) -> dict[str, Any]:
    exact_https(url, {API_HOST}, "/v2/")
    raw = None if body is None else json.dumps(body, separators=(",", ":")).encode()
    if raw is not None and len(raw) > MAX_JSON_BYTES:
        raise ValueError("request exceeds client cap")
    headers = {"api_token": token, "Accept": "application/json", "User-Agent": USER_AGENT}
    if raw is not None:
        headers["Content-Type"] = "application/json"
    req = Request(url, data=raw, headers=headers, method=method)
    try:
        with OPENER.open(req, timeout=timeout) as resp:
            return read_json_response(resp)
    except HTTPError as exc:
        retry_after = exc.headers.get("Retry-After")
        try:
            raw_error = exc.read(16_385)[:16_384]
        finally:
            exc.close()
        try:
            error_data = json.loads(raw_error)
        except (UnicodeDecodeError, json.JSONDecodeError):
            error_data = {}
        if not isinstance(error_data, dict):
            error_data = {}
        error_obj = error_data.get("error")
        error_obj = error_obj if isinstance(error_obj, dict) else {}
        message = str(error_obj.get("message", ""))
        raise BriaHTTPError(
            exc.code,
            f"Bria HTTP {exc.code}; request_id={error_data.get('request_id')!r}; "
            f"error_code={error_obj.get('code')!r}; "
            f"message_sha256={hashlib.sha256(message.encode()).hexdigest() if message else None}",
            retry_after,
        ) from None
    except (TimeoutError, URLError) as exc:
        raise AmbiguousTransportError(
            "transport outcome is ambiguous; do not resubmit a create request automatically; "
            "reconcile by request_id/account logs first"
        ) from exc


def retry_after_seconds(value: str | None) -> float:
    if not value:
        return 0.0
    try:
        seconds = float(value)
    except ValueError:
        try:
            moment = parsedate_to_datetime(value)
            if moment.tzinfo is None:
                moment = moment.replace(tzinfo=timezone.utc)
            seconds = (moment - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            return 0.0
    return seconds if math.isfinite(seconds) and seconds > 0 else 0.0


def validate_local_image(path: Path, byte_cap: int = MAX_INPUT_BYTES) -> bytes:
    if not path.is_file() or path.stat().st_size > byte_cap:
        raise ValueError("reference is missing or exceeds the client cap")
    raw = path.read_bytes()
    if not (raw.startswith(b"\x89PNG\r\n\x1a\n") or raw.startswith(b"\xff\xd8\xff") or raw.startswith(b"RIFF") and raw[8:12] == b"WEBP"):
        raise ValueError("reference magic is not PNG/JPEG/WEBP")
    try:
        from PIL import Image, UnidentifiedImageError
        Image.MAX_IMAGE_PIXELS = MAX_PIXELS
        with Image.open(path) as im:
            if im.width * im.height > MAX_PIXELS or getattr(im, "n_frames", 1) != 1:
                raise ValueError("reference pixels/frames exceed client policy")
            im.verify()
        with Image.open(path) as im:
            im.load()
    except ImportError as exc:
        raise RuntimeError("install Pillow to validate image inputs") from exc
    except (UnidentifiedImageError, OSError) as exc:
        raise ValueError("reference image failed full decode") from exc
    return raw


def load_structured_prompt(path: Path) -> str:
    text = path.read_text(encoding="utf-8")
    parsed = json.loads(text)
    if not isinstance(parsed, dict):
        raise ValueError("structured_prompt must encode a JSON object")
    return text


def build_payload(route: str) -> dict[str, Any]:
    prompt = os.getenv("BRIA_PROMPT", "").strip()
    ref_path = os.getenv("BRIA_REFERENCE_PATH", "").strip()
    struct_path = os.getenv("BRIA_STRUCTURED_PROMPT_PATH", "").strip()
    if ref_path and struct_path:
        raise ValueError("images and structured_prompt are mutually exclusive")
    payload: dict[str, Any] = {}
    if prompt:
        payload["prompt"] = prompt
    if ref_path:
        payload["images"] = [base64.b64encode(validate_local_image(Path(ref_path))).decode("ascii")]
    if struct_path:
        payload["structured_prompt"] = load_structured_prompt(Path(struct_path))
    if not any(k in payload for k in ("prompt", "images", "structured_prompt")):
        raise ValueError("set BRIA_PROMPT, BRIA_REFERENCE_PATH, or BRIA_STRUCTURED_PROMPT_PATH")
    if route.startswith("/structured_prompt/") and "structured_prompt" in payload and "prompt" not in payload:
        raise ValueError("structured-prompt routes require a prompt when refining a structured_prompt")

    seed_text = os.getenv("BRIA_SEED", "").strip()
    if struct_path and not seed_text:
        raise ValueError("recreation/refinement requires the saved BRIA_SEED")
    if seed_text:
        payload["seed"] = int(seed_text)

    payload.update({
        "sync": False,
        "ip_signal": True,
        "prompt_content_moderation": True,
        "visual_input_content_moderation": True,
    })
    if route.startswith("/image/"):
        ratio = os.getenv("BRIA_ASPECT_RATIO", "1:1")
        if ratio not in RATIOS:
            raise ValueError("unsupported aspect ratio")
        payload.update({
            "model_version": "FIBO",
            "aspect_ratio": ratio,
            "output_type": os.getenv("BRIA_OUTPUT_TYPE", "png"),
            "visual_output_content_moderation": True,
        })
        if payload["output_type"] not in {"png", "jpeg"}:
            raise ValueError("output type must be png or jpeg")
    if route == "/image/generate":
        resolution = os.getenv("BRIA_RESOLUTION", "1MP")
        steps = int(os.getenv("BRIA_STEPS", "50"))
        if resolution not in {"1MP", "4MP"} or not 35 <= steps <= 50:
            raise ValueError("invalid Standard resolution/steps")
        payload.update({"resolution": resolution, "steps_num": steps})
        if os.getenv("BRIA_NEGATIVE_PROMPT"):
            payload["negative_prompt"] = os.environ["BRIA_NEGATIVE_PROMPT"]
    return payload


def poll(request_id: str, token: str) -> dict[str, Any]:
    # Construct from the trusted request_id; never forward api_token to returned arbitrary URLs.
    if not request_id or "/" in request_id or len(request_id) > 256:
        raise RuntimeError("invalid request_id")
    url = f"{BASE}/status/{request_id}"
    deadline = time.monotonic() + POLL_TIMEOUT_S
    delay = 1.0
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError(
                f"poll timeout for {request_id}; this does not cancel the job or prove it was not billed"
            )
        try:
            data = api_json("GET", url, token, timeout=max(0.001, min(60.0, remaining)))
        except BriaHTTPError as exc:
            if exc.status_code == 404:
                raise RuntimeError(f"status record is missing or expired for {request_id}") from exc
            if exc.status_code != 429 and exc.status_code < 500:
                raise
            print(f"status retry after HTTP {exc.status_code}")
            delay = max(min(delay * 1.7, 15.0), retry_after_seconds(exc.retry_after))
        except AmbiguousTransportError:
            # GET status is safe to retry; still bound the attempts by the deadline.
            print("status retry after transport failure")
            delay = min(delay * 1.7, 15.0)
        else:
            if data.get("request_id") not in (None, request_id):
                raise RuntimeError("status request_id mismatch")
            status = data.get("status")
            if status == "IN_PROGRESS":
                pass
            elif status == "COMPLETED":
                return data
            elif status in {"ERROR", "UNKNOWN"}:
                err = data.get("error") or {}
                if not isinstance(err, dict):
                    raise RuntimeError(f"Bria job {status} returned an invalid error object")
                message = str(err.get("message", ""))
                raise RuntimeError(
                    f"Bria job {status}: code={err.get('code')!r} "
                    f"message_sha256={hashlib.sha256(message.encode()).hexdigest() if message else None}"
                )
            else:
                raise RuntimeError(f"unknown Bria job status: {status!r}")
            delay = min(delay * 1.7, 15.0)
        remaining = deadline - time.monotonic()
        sleep_for = delay + random.uniform(0, min(delay * 0.2, 1.0))
        if sleep_for >= remaining:
            raise TimeoutError(
                f"poll deadline cannot accommodate retry delay for {request_id}; "
                "this does not cancel the job or prove it was not billed"
            )
        time.sleep(sleep_for)


def atomic_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f:
            fd = None
            f.write(text)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp_name, path)
    finally:
        if fd is not None:
            os.close(fd)
        if os.path.exists(tmp_name):
            os.unlink(tmp_name)


def download_image(
    url: str,
    destination: Path,
    allowed_hosts: set[str],
    expected_content_type: str,
) -> tuple[str, int, str]:
    exact_https(url, allowed_hosts)
    req = Request(url, headers={"Accept": "image/png,image/jpeg", "User-Agent": USER_AGENT}, method="GET")
    destination.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(prefix=destination.name + ".", dir=destination.parent)
    digest = hashlib.sha256()
    size = 0
    ctype = ""
    try:
        with os.fdopen(fd, "wb") as out:
            fd = None
            with OPENER.open(req, timeout=60) as resp:
                ctype = resp.headers.get_content_type()
                if ctype != expected_content_type:
                    raise RuntimeError(f"unexpected image Content-Type: {ctype}")
                length = resp.headers.get("Content-Length")
                if length and int(length) > MAX_OUTPUT_BYTES:
                    raise RuntimeError("image exceeds client cap")
                while True:
                    chunk = resp.read(1024 * 1024)
                    if not chunk:
                        break
                    size += len(chunk)
                    if size > MAX_OUTPUT_BYTES:
                        raise RuntimeError("image exceeds client cap")
                    digest.update(chunk)
                    out.write(chunk)
            out.flush()
            os.fsync(out.fileno())
        with Path(tmp_name).open("rb") as check:
            raw_head = check.read(12)
        if ctype == "image/png" and not raw_head.startswith(b"\x89PNG\r\n\x1a\n"):
            raise RuntimeError("PNG magic mismatch")
        if ctype == "image/jpeg" and not raw_head.startswith(b"\xff\xd8\xff"):
            raise RuntimeError("JPEG magic mismatch")
        validate_local_image(Path(tmp_name), MAX_OUTPUT_BYTES)
        os.replace(tmp_name, destination)
        return digest.hexdigest(), size, ctype
    finally:
        if fd is not None:
            os.close(fd)
        if os.path.exists(tmp_name):
            os.unlink(tmp_name)


def read_local_json(path: Path) -> dict[str, Any]:
    if path.stat().st_size > MAX_JSON_BYTES:
        raise RuntimeError("local job record exceeds JSON cap")
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise RuntimeError("local job record is not a JSON object")
    return value


def claim_job(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as output:
            fd = None
            json.dump(value, output, indent=2, sort_keys=True)
            output.flush(); os.fsync(output.fileno())
    finally:
        if fd is not None: os.close(fd)


def save_job(path: Path, value: dict[str, Any]) -> None:
    atomic_text(path, json.dumps(value, indent=2, sort_keys=True) + "\n")


def main() -> None:
    route = os.getenv("BRIA_ROUTE", "/image/generate")
    if route not in ROUTES:
        raise ValueError("BRIA_ROUTE is not one of the four first-party FIBO routes")
    payload = build_payload(route)
    public_cost = ROUTES[route]
    override = os.getenv("BRIA_ESTIMATED_CALL_USD", "").strip()
    try: cost = float(override) if override else public_cost
    except ValueError as exc: raise RuntimeError("BRIA_ESTIMATED_CALL_USD must be numeric") from exc
    if route == "/image/generate" and payload.get("resolution") == "4MP" and not override:
        cost = None  # public page does not state whether 4MP changes billing
    if cost is not None and (not math.isfinite(cost) or cost <= 0):
        raise RuntimeError("estimated call cost must be finite and positive")
    if cost is not None and public_cost is not None and cost < public_cost:
        raise RuntimeError("operator estimate cannot undercut the dated public route price")
    try: max_cost = float(os.getenv("BRIA_MAX_COST_USD", "0"))
    except ValueError as exc: raise RuntimeError("BRIA_MAX_COST_USD must be numeric") from exc
    if not math.isfinite(max_cost) or max_cost < 0:
        raise RuntimeError("BRIA_MAX_COST_USD must be finite and non-negative")
    request_sha256 = hashlib.sha256(json.dumps(
        payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    client_job_key = os.getenv("BRIA_CLIENT_JOB_KEY", f"bria-{request_sha256[:20]}")
    output_hosts = {h.strip().lower() for h in os.getenv("BRIA_OUTPUT_HOSTS", "").split(",") if h.strip()}
    out_dir = Path(os.getenv("BRIA_OUTPUT_DIR", "bria-output")).resolve()
    approval_value = {"route": route, "request_sha256": request_sha256,
                      "estimated_call_usd": cost, "max_cost_usd": max_cost,
                      "client_job_key": client_job_key,
                      "output_hosts": sorted(output_hosts), "output_dir": str(out_dir)}
    approval_sha256 = hashlib.sha256(json.dumps(
        approval_value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    input_mode = "+".join(k for k in ("prompt", "images", "structured_prompt") if k in payload)
    print(json.dumps({
        "dry_run": not env_bool("SEND_BRIA_REQUEST"),
        "url": BASE + route,
        "input_mode": input_mode,
        "aspect_ratio": payload.get("aspect_ratio"),
        "resolution": payload.get("resolution"),
        "estimated_max_usd_2026_07_10": cost,
        "maximum_approved_usd": max_cost,
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "approval_sha256": approval_sha256,
        "output_hosts": sorted(output_hosts),
        "output_dir": str(out_dir),
        "moderation": "all applicable stages enabled",
    }, indent=2))
    if not env_bool("SEND_BRIA_REQUEST"):
        print("DRY RUN ONLY: no network request made")
        return
    if os.getenv("APPROVE_BRIA_PAID_CALL") != "YES":
        raise RuntimeError("set APPROVE_BRIA_PAID_CALL=YES after reviewing this request")
    if cost is None:
        raise RuntimeError("set BRIA_ESTIMATED_CALL_USD from the live account/contract before approval")
    if max_cost <= 0 or max_cost < cost:
        raise RuntimeError("BRIA_MAX_COST_USD is below the dated per-call estimate")
    if os.getenv("BRIA_APPROVED_REQUEST_SHA256") != approval_sha256:
        raise RuntimeError("approval hash does not match this exact request, cost, and client job key")
    token = os.getenv("BRIA_API_TOKEN", "")
    if not token:
        raise RuntimeError("BRIA_API_TOKEN is required only for a live call")
    if route.startswith("/image/") and not output_hosts:
        raise RuntimeError("preconfigure exact BRIA_OUTPUT_HOSTS from your Bria deployment/egress policy")

    job_token = hashlib.sha256(client_job_key.encode()).hexdigest()[:20]
    job_dir = out_dir / job_token
    ledger_path = job_dir / "job.json"
    manifest_path = job_dir / "manifest.json"
    ledger_base = {"client_job_key": client_job_key, "route": route,
                   "request_sha256": request_sha256, "approval_sha256": approval_sha256,
                   "estimated_call_usd": cost, "approved_max_cost_usd": max_cost}
    if manifest_path.exists():
        existing_manifest = read_local_json(manifest_path)
        if any(existing_manifest.get(key) != value for key, value in ledger_base.items()):
            raise RuntimeError("existing manifest does not match this approved request")
        print(json.dumps(existing_manifest, indent=2))
        return
    warnings: list[str] = []
    warning_hashes: list[str] = []
    warning_present = False
    if ledger_path.exists():
        existing = read_local_json(ledger_path)
        if any(existing.get(key) != value for key, value in ledger_base.items()):
            raise RuntimeError("existing job ledger does not match this approved request")
        request_id = existing.get("request_id")
        if not isinstance(request_id, str):
            raise RuntimeError(
                "existing create has no confirmed request_id; treat it as ambiguous and do not replay. "
                "Use a new explicit BRIA_CLIENT_JOB_KEY only after operator reconciliation and approval."
            )
        saved_hashes = existing.get("provider_warning_sha256", [])
        if not isinstance(saved_hashes, list) or not all(isinstance(value, str) for value in saved_hashes):
            raise RuntimeError("existing job ledger has invalid warning hashes")
        warning_hashes = saved_hashes
        warning_present = bool(existing.get("provider_warning_present"))
        final = poll(request_id, token)
    else:
        try:
            claim_job(ledger_path, {**ledger_base, "state": "pre_submit",
                                   "request_id": None, "updated_unix": time.time()})
        except FileExistsError as exc:
            raise RuntimeError(
                "another worker claimed this client job key; do not submit; retry later to resume"
            ) from exc
        try:
            initial = api_json("POST", BASE + route, token, payload)  # never auto-retry this create
        except AmbiguousTransportError:
            save_job(ledger_path, {**ledger_base, "state": "ambiguous_create",
                                   "request_id": None, "updated_unix": time.time()})
            raise
        except BriaHTTPError as exc:
            state = "ambiguous_create" if exc.status_code >= 500 else "create_rejected"
            save_job(ledger_path, {**ledger_base, "state": state,
                                   "request_id": None, "http_status": exc.status_code,
                                   "updated_unix": time.time()})
            raise
        request_id = initial.get("request_id")
        if not isinstance(request_id, str) or not request_id or "/" in request_id or len(request_id) > 256:
            save_job(ledger_path, {**ledger_base, "state": "invalid_create_response",
                                   "request_id": None, "updated_unix": time.time()})
            raise RuntimeError("missing or invalid request_id")
        warnings = [initial["warning"]] if isinstance(initial.get("warning"), str) else []
        warning_hashes = [hashlib.sha256(value.encode()).hexdigest() for value in warnings]
        warning_present = bool(warnings)
        save_job(ledger_path, {**ledger_base, "state": "accepted",
                               "request_id": request_id,
                               "provider_warning_present": warning_present,
                               "provider_warning_sha256": warning_hashes,
                               "updated_unix": time.time()})
        if "status_url" in initial:
            expected = f"{BASE}/status/{request_id}"
            if initial["status_url"] != expected:
                raise RuntimeError("unexpected status_url; token was not forwarded")
            final = poll(request_id, token)
        else:
            # Defensive support for a synchronous-shaped response, though this client sends sync=false.
            final = initial
    if isinstance(final.get("warning"), str):
        final_warning_hash = hashlib.sha256(final["warning"].encode()).hexdigest()
        if final_warning_hash not in warning_hashes: warning_hashes.append(final_warning_hash)
        warning_present = True
    if warning_present:
        print(f"Bria returned {len(warning_hashes)} distinct warning(s) for {request_id}; review required")
    result = final.get("result")
    if not isinstance(result, dict):
        raise RuntimeError("completed response has no result object")

    structured = result.get("structured_prompt")
    if isinstance(structured, str):
        parsed = json.loads(structured)
        if not isinstance(parsed, dict):
            raise RuntimeError("returned structured_prompt is not a JSON object")
        atomic_text(job_dir / "structured_prompt.json", structured)
    manifest: dict[str, Any] = {
        "request_id": request_id,
        "route": route,
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "approval_sha256": approval_sha256,
        "seed": result.get("seed"),
        "structured_prompt_sha256": hashlib.sha256(structured.encode()).hexdigest() if isinstance(structured, str) else None,
        "source_url_recorded": False,
        "provider_warning_present": warning_present,
        "provider_warning_sha256": warning_hashes,
        "moderation_flags": {
            "prompt": True,
            "visual_input": True,
            "visual_output": True if route.startswith("/image/") else None,
        },
        "estimated_call_usd_2026_07_10": cost,
        "estimated_call_usd": cost,
        "approved_max_cost_usd": max_cost,
    }
    if route.startswith("/image/"):
        image_url = result.get("image_url")
        if not isinstance(image_url, str):
            raise RuntimeError("completed image response has no image_url")
        suffix = ".png" if payload["output_type"] == "png" else ".jpg"
        expected_type = "image/png" if payload["output_type"] == "png" else "image/jpeg"
        sha256, size, ctype = download_image(
            image_url,
            job_dir / ("image" + suffix),
            output_hosts,
            expected_type,
        )
        manifest.update({"image_sha256": sha256, "image_bytes": size, "content_type": ctype})
    atomic_text(manifest_path, json.dumps(manifest, indent=2, sort_keys=True) + "\n")
    save_job(ledger_path, {**ledger_base, "state": "completed",
                           "request_id": request_id,
                           "provider_warning_present": warning_present,
                           "provider_warning_sha256": warning_hashes,
                           "updated_unix": time.time()})
    print(json.dumps(manifest, indent=2))


if __name__ == "__main__":
    main()

ドライランの例（PowerShell。リクエストは送信されません）：

$env:BRIA_PROMPT='A cobalt ceramic pour-over set on a warm gray studio sweep; no text or logos'
python .\bria_fibo.py

$env:BRIA_ROUTE='/image/generate/lite'
$env:BRIA_PROMPT='Use only the reference palette and soft side lighting; create a distinct unbranded product'
$env:BRIA_REFERENCE_PATH='C:\approved-inputs\reference.png'
python .\bria_fibo.py

$env:BRIA_ROUTE='/image/generate'
$env:BRIA_REFERENCE_PATH=''
$env:BRIA_STRUCTURED_PROMPT_PATH='C:\approved-inputs\structured_prompt.json'
$env:BRIA_SEED='123456'
$env:BRIA_PROMPT='Change only the background to pale warm gray; preserve every object and camera property'
python .\bria_fibo.py

ドライランの概要を人間が確認し、現在のアカウント料金と利用条件を確かめ、1回の呼び出しを承認した後に限り、次を実行してください。

$env:BRIA_API_TOKEN='<set securely; never paste into logs or source>'
$env:BRIA_OUTPUT_HOSTS='<exact host approved for your Bria result URLs>'
$env:BRIA_MAX_COST_USD='0.03'
$env:BRIA_APPROVED_REQUEST_SHA256='<copy approval_sha256 from the dry run with this same cost ceiling>'
$env:APPROVE_BRIA_PAID_CALL='YES'
$env:SEND_BRIA_REQUEST='1'
python .\bria_fibo.py

承認済みの呼び出しが完了したら、SEND_BRIA_REQUEST、APPROVE_BRIA_PAID_CALL、BRIA_APPROVED_REQUEST_SHA256、API トークンの設定を直ちに解除してください。.env、機密情報を含むプロンプト、Base64 形式の参照データ、署名付きの結果 URL、API レスポンスをコミットしないでください。この例では、トークン、Base64 データ、プロンプト本文、構造化プロンプト、署名付き URL は意図的にログへ出力しません。

非同期処理と Webhook の正確性

文書に記載された事実： sync=false は、request_id と status_url を含む HTTP 202 を返します。api_token を指定して GET /status/{request_id} をポーリングします。終端ステータスは COMPLETED、ERROR、UNKNOWN で、IN_PROGRESS の場合は処理が続行中です。ジョブレベルのエラーでも、通常は HTTP 200 が返されます。リクエストが見つからない場合や期限切れの場合は、404 が返されることがあります。

ステータスの OpenAPI の成功スキーマには、説明文で必須とされている IN_PROGRESS が現時点で列挙されていません。説明文で定義された状態を処理してください。検証していない返却 URL に api_token を決して送信しないでください。例にあるように、信頼できるベース URL とリクエスト ID からステータス URL を構築してください。

タイムアウト、接続リセット、または曖昧な 5xx の後に、POST を自動的に再試行しないでください。冪等性キーのフィールドも、クライアントキーによるジョブ検索機能も文書化されていません。取得済みの request_id、アカウントログ、またはサポートを通じて状況を確認してください。ステータスの GET には、上限付きバックオフを使用できます。ポーリングのタイムアウトによってジョブがキャンセルされることはありません。公開されているキャンセル用エンドポイントは見つかりませんでした。

本番環境のコールバックには、Bria は Webhook を推奨しています。文書に記載されている仕様は次のとおりです。

• 生の POST 本文と、Bria-Webhook-Id、Bria-Webhook-Timestamp、Bria-Webhook-Signature（v1=<base64>）。
• HMAC-SHA256 署名キー：HMAC-SHA256(api_token, "bria-webhook-signing-v1")。
• 署名対象メッセージ：{webhook_id}.{timestamp}.{raw_body}。
• 10 秒以内に応答すること。
• 45 分間に最大 5 回の配信試行があり、重複する可能性があります。ジョブ ID／リクエスト ID で重複を排除してください。

JSON を解析する前に生バイト列を検証し、成功応答を返す前に、hmac.compare_digest または同等の定時間プリミティブを使って期待される署名と受信した署名を比較してください。検証と永続的な重複排除の後、重い処理をキューに登録し、10 秒以内に応答してください。アーティファクトの取得と後続処理はバックグラウンドで実行してください。Bria の公開ページには、許容されるタイムスタンプのずれ／リプレイウィンドウが記載されていません。運用上のヒューリスティックとして、短く監視可能な鮮度ウィンドウ（たとえば 5 分）を設け、処理済み ID は 45 分の再試行期間より長く永続的に保存し、署名キーの元となるトークンは慎重にローテーションし、時刻同期を確認してください。任意のサイズのペイロードをコールバックに公開しないでください。

アーティファクトと URL の安全性

クエリが表示されていない場合でも、image_url と status_url は認証情報を含む、または機密性の高い情報として扱ってください。FIBO の出力 URL の有効期間は、公開ドキュメントに記載されていません。すぐに管理下のストレージにダウンロードし、その後、永続化したコピーをハッシュ化してインベントリに記録してください。

最低限、次の対策を行ってください。

• HTTPS、デフォルトポート、完全一致のホスト許可リスト、パブリック DNS を必須にしてください。
• リダイレクトを無効にし、アセットホストに api_token を決して転送しないでください。
• より強力な SSRF／DNS リバインディング対策として、ネットワーク層で送信トラフィックと DNS のピン留めを設定してください。
• ヘッダー、本文、時間に上限を設け、MIME タイプとファイルマジックナンバーを確認し、完全にデコードして、ピクセル数／フレーム数の上限を適用し、アーカイブ／HTML／SVG を拒否してください。
• 同じディレクトリ内の一時ファイルに書き込み、fsync を実行してから、アトミックに置き換えてください。
• C2PA やその他の来歴メタデータが誤って削除されないよう、元のバイト列を保持してください。必要に応じて、来歴を別途検証してください。
• リクエスト ID、ルート、シード、構造化プロンプトのハッシュ、出力ハッシュ、モデレーション／警告の処置、承認、価格のスナップショット、コード／設定のリビジョンを保存してください。
• 通常のログには、署名付き URL、トークン、base64、個人データ、機密性の高いプロンプトの生データを記録しないでください。

品質とリリースの確認

見た目の魅力だけでなく、記載された受け入れ基準に照らして評価してください。

• 被写体の数、解剖学的・幾何学的な整合性、空間的な関係、トリミング、テキストのスペル、ブランド／知的財産の漏出。
• 参照画像との過度な類似性、なりすましや肖像権に関するリスク。
• 出力の寸法／アスペクト比、デコードの完全性、カラーモード、アルファの挙動、ハッシュ。
• 下流で最適化した後の、必須メタデータ／来歴情報の挙動。
• 保存したプロンプト／シード／モデルのルートでの再現性、バックエンド更新後のドリフト。
• 影響の大きい用途、一般公開用途、規制対象の用途、政治関連の用途、または個人のアイデンティティに関わる用途では、人によるレビューを実施してください。

モデレーションに成功しても、許可が得られたこと、事実の正確性、権利侵害がないこと、またはブランドの承認を意味するわけではありません。プロバイダーのモデレーションを有効にしたまま、アプリケーション側のポリシーチェックも追加してください。

権利、プライバシー、ライセンスの台帳

以下は個別に記録してください。「商用利用しても安全」にまとめないでください。

• トレーニングデータ ― プロバイダーの主張： Bria は、FIBO がライセンスされたデータのみでトレーニングされ、トレーサビリティ／GDPR／EU AI Act への準拠を実現していると説明しています。論文／モデルカードは Bria が作成した証拠であり、独立した監査ではありません。
• ホスト型出力の権利／補償 ― 契約によって異なります： 料金ページでは、プランに応じた標準または完全な知的財産／プライバシーに関する補償をうたっています。一方、Legal Lobby では、オンライン／無料の利用規約は保証が限定され、エンタープライズ向け規約ではより広範な保護が提供されると説明されています。準拠する規約、適用除外、上限、地域、利用制限、ユーザーのプロンプト／参照画像によって結果が変わることがあります。これを信頼する場合は、法律の専門家による確認を受けてください。
• 入力コンテンツの権利： ユーザーは、プロンプト／参照画像のコンテンツと、それに関するプライバシー権／パブリシティ権を所有するか、使用許諾を得ている必要があります。Bria のコンテンツガイドラインでは、権利侵害、欺瞞、なりすまし、違法、ヘイト、ポルノ、悪意のある行為、詐欺的な行為などを禁止しています。
• 出力の独占性： 過去に公開された規約では、AI の出力は独自のものとは限らないとされています。現行の契約を確認してください。独占性や登録可能性を約束してはいけません。
• 重み： briaai/FIBO はアクセス制限付きで、bria-fibo と表示されています。モデルカードのメタデータには CC BY-NC 4.0 へのリンクがあり、非商用利用のみと記載されています。briaai/Fibo-lite も CC BY-NC 4.0 と表示され、非商用利用のみと記載されています。商用のローカル／オンプレミス利用には、別途 Bria の許可／規約が必要です。「オープンソース」は Bria が使用している表現です。これを非商用制限やアクセス制限の条件を覆すものとして扱わないでください。ダウンロード／デプロイの前に、実際に同意する規約とすべての依存関係を確認してください。
• 顧客データを使ったトレーニング： Bria の Legal Lobby には、明示的な書面合意がない限り、パイロット顧客のコンテンツ／出力を生成モデルのトレーニングや改善に使用しないと記載されています。2025年3月のエンタープライズ向け文書にも同様の記載がありますが、ファインチューニングは例外です。現行のセルフサービスプランすべてに適用されるかは、締結済みの現行規約を確認しない限り不明です。
• 保持／削除： 2024年7月のプライバシーポリシーでは、明記された目的と法的な必要性に応じて、個人データを保持するとされています。以前の公開規約では、文書で合意されていない限りデータは保存せず、契約終了後30日以内に削除すると説明されています。ただし、現行の FIBO ジョブ／結果の保持期間と URL の有効期限（TTL）は不明です。ゼロ保持を約束しないでください。
• 個人データ： 以前の規約では、Customer Content に個人データを含めてはならないとされています。現行のプラン別の取り扱いは不明です。現行の契約と DPA で使用が明示的に許可されていない限り、個人データの使用は禁止するのが安全です。
• データ所在地／サブプロセッサー／セキュリティ： 確認した公開ページからは、FIBO の正確な処理地域、顧客が選択できるデータ所在地、サブプロセッサーの一覧、暗号化の詳細、バックアップ、インシデント対応に関する約束、ステータスログの保持期間は不明です。DPA／セキュリティ資料一式と、アーキテクチャに関する確認を入手してください。

これは運用上のガイダンスであり、法律上の助言ではありません。

現在の公式ソース

確認日：2026-07-10

• Image Generation v2 API とダウンロード可能な OpenAPI
• 画像生成エンドポイント
• ステータスサービス
• Webhook の配信と署名
• API の概要：認証、レート制限、非同期フロー、入力エンコーディング
• Bria の料金
• Bria Legal Lobby
• Bria の Legal Lobby にリンクされている最新のコンテンツガイドライン（2026 年 1 月）
• Bria の Legal Lobby にリンクされている最新のオンライン規約（2026 年 3 月）
• Bria コンテンツガイドライン（一般公開されている公式 PDF。2024 年 6 月のインデックス版）
• Bria プライバシーポリシー、最終更新日 2024-07-08
• FIBO モデルカード
• FIBO Lite モデルカード
• FIBO 論文、arXiv:2511.06876

文書間で内容に相違がある場合、通信形式についてはエンドポイントの OpenAPI を優先し、商用および運用上の利用権についてはユーザーのアカウントまたは契約を優先してください。また、Bria が書面で矛盾を解消するまでは、より厳格な安全性およびプライバシー保護の方針に従ってください。