Tool_Agent/tools/local/midjourney/midjourney_client.py

"""将请求转发到自建 Midjourney HTTP 服务（与 tools/local 约定一致）。"""

from __future__ import annotations

import os
from typing import Any

import httpx
from dotenv import load_dotenv

_ = load_dotenv()


def _base_url() -> str:
    base = os.environ.get("MIDJOURNEY_API_BASE", "").strip().rstrip("/")
    if not base:
        raise ValueError("缺少环境变量 MIDJOURNEY_API_BASE（上游根 URL，不含尾路径）")
    return base


def forward_post(path: str, json_body: dict[str, Any]) -> Any:
    """POST {MIDJOURNEY_API_BASE}{path}，Content-Type: application/json。"""
    url = f"{_base_url()}{path if path.startswith('/') else '/' + path}"
    with httpx.Client(timeout=300.0) as client:
        r = client.post(
            url,
            json=json_body,
            headers={"accept": "application/json", "Content-Type": "application/json"},
        )
        ct = (r.headers.get("content-type") or "").lower()
        if "application/json" not in ct:
            r.raise_for_status()
            raise RuntimeError(f"非 JSON 响应 ({r.status_code}): {r.text[:1500]}")
        try:
            data = r.json()
        except Exception:
            raise RuntimeError(f"无效 JSON ({r.status_code}): {r.text[:1500]}") from None

    if r.status_code >= 400:
        if isinstance(data, dict):
            msg = data.get("detail", data.get("message", data.get("error", str(data))))
        else:
            msg = str(data)
        raise RuntimeError(f"上游 HTTP {r.status_code}: {msg}")

    return data