feat: 图片改图 MCP 工具 image_edit
- imagegen.edit：POST multipart /v1/images/edits，输入图 必须是工作区内 plain basename 且存在（上限 20MB），产出 必须是新文件、绝不覆盖输入；响应解析与 generate 共用 _save_image_response（url 或 b64_json、下载落盘、格式探测）。 - imagegen_mcp 新增 image_edit 工具；image-workflow 技能 改图分支从“无专用工具”改为用 image_edit。 - 测试：multipart 体含提示词与原图字节、输入缺失/同名/ 路径逃逸拒绝、MCP 列表含 image_edit；全量相关 16 项过。 实测依据（2026-10-30）：中转站图片分组 key 调 gpt-image-2.5-flare edits 返回 200，URL 图为真实 PNG 1024x1024、909K，usage 含输入图 token（1659 image tokens）。
This commit is contained in:
host committed
1 parent
5b03e0a85d
commit
e13a95067c
4 files changed
+236
-30
No files matched your search
+135
-21
@@ -112,17 +112,7 @@ def _download(url, download=None):
|
||||
raise ImageGenError("image download failed: %s" % exc)
|
||||
|
||||
|
||||
def generate(prompt, output_filename=None, size="1024x1024",
|
||||
model=None, request=None, download=None, cwd=None):
|
||||
"""Generate one image and save it to the current working
|
||||
directory (the task worktree when run as a Codex MCP server).
|
||||
|
||||
``output_filename`` must be a plain basename (no separators,
|
||||
no ``..``); default is ``generated-<epoch>.png``. Returns a
|
||||
dict with path, format, dimensions (PNG only), byte size and
|
||||
the relay's usage block. Raises ImageGenError on any failure;
|
||||
the message never contains the API key.
|
||||
"""
|
||||
def _validate_common(prompt, size, output_filename):
|
||||
if not prompt or not str(prompt).strip():
|
||||
raise ImageGenError("prompt is required")
|
||||
if size not in ALLOWED_SIZES:
|
||||
@@ -138,20 +128,20 @@ def generate(prompt, output_filename=None, size="1024x1024",
|
||||
or "/" in name or "\\" in name):
|
||||
raise ImageGenError(
|
||||
"output_filename must be a plain basename, got %r" % name)
|
||||
use_model = model or _default_model()
|
||||
status, data = _post_json(
|
||||
_base_url() + "/v1/images/generations", key,
|
||||
{"model": use_model, "prompt": str(prompt), "size": size},
|
||||
request=request)
|
||||
return key, name
|
||||
|
||||
|
||||
def _save_image_response(status, data, name, use_model, size,
|
||||
download, cwd, verb):
|
||||
if status != 200:
|
||||
err = (data or {}).get("error") or {}
|
||||
msg = err.get("message") if isinstance(err, dict) else str(err)
|
||||
raise ImageGenError(
|
||||
"image generation failed: HTTP %s: %s"
|
||||
% (status, (msg or data or "unknown error")))
|
||||
"image %s failed: HTTP %s: %s"
|
||||
% (verb, status, (msg or data or "unknown error")))
|
||||
items = (data or {}).get("data") or []
|
||||
if not items or not isinstance(items[0], dict):
|
||||
raise ImageGenError("image generation returned no image data")
|
||||
raise ImageGenError("image %s returned no image data" % verb)
|
||||
first = items[0]
|
||||
b64 = first.get("b64_json")
|
||||
if b64:
|
||||
@@ -163,8 +153,8 @@ def generate(prompt, output_filename=None, size="1024x1024",
|
||||
raw = _download(first["url"], download=download)
|
||||
else:
|
||||
raise ImageGenError(
|
||||
"image generation returned neither url nor b64_json "
|
||||
"(keys: %s)" % ", ".join(sorted(first.keys())))
|
||||
"image %s returned neither url nor b64_json "
|
||||
"(keys: %s)" % (verb, ", ".join(sorted(first.keys()))))
|
||||
if not raw:
|
||||
raise ImageGenError("downloaded image is empty")
|
||||
fmt, width, height = _detect_format(raw)
|
||||
@@ -182,6 +172,130 @@ def generate(prompt, output_filename=None, size="1024x1024",
|
||||
return result
|
||||
|
||||
|
||||
def generate(prompt, output_filename=None, size="1024x1024",
|
||||
model=None, request=None, download=None, cwd=None):
|
||||
"""Generate one image and save it to the current working
|
||||
directory (the task worktree when run as a Codex MCP server).
|
||||
|
||||
``output_filename`` must be a plain basename (no separators,
|
||||
no ``..``); default is ``generated-<epoch>.png``. Returns a
|
||||
dict with path, format, dimensions (PNG only), byte size and
|
||||
the relay's usage block. Raises ImageGenError on any failure;
|
||||
the message never contains the API key.
|
||||
"""
|
||||
key, name = _validate_common(prompt, size, output_filename)
|
||||
use_model = model or _default_model()
|
||||
status, data = _post_json(
|
||||
_base_url() + "/v1/images/generations", key,
|
||||
{"model": use_model, "prompt": str(prompt), "size": size},
|
||||
request=request)
|
||||
return _save_image_response(status, data, name, use_model, size,
|
||||
download, cwd, "generation")
|
||||
|
||||
|
||||
def _content_type_for(filename):
|
||||
lower = filename.lower()
|
||||
if lower.endswith(".jpg") or lower.endswith(".jpeg"):
|
||||
return "image/jpeg"
|
||||
if lower.endswith(".webp"):
|
||||
return "image/webp"
|
||||
return "image/png"
|
||||
|
||||
|
||||
def _build_multipart(fields, file_field, filename, file_bytes):
|
||||
"""Build a multipart/form-data body. Returns (body, boundary).
|
||||
Pure function for testability: the real post uses it too."""
|
||||
import uuid
|
||||
|
||||
boundary = "----codeximagegen" + uuid.uuid4().hex
|
||||
body = b""
|
||||
for name, value in fields:
|
||||
body += ("--" + boundary + "\r\n").encode()
|
||||
body += ('Content-Disposition: form-data; '
|
||||
'name="%s"\r\n\r\n' % name).encode()
|
||||
body += value.encode() + b"\r\n"
|
||||
body += ("--" + boundary + "\r\n").encode()
|
||||
body += ('Content-Disposition: form-data; name="%s"; '
|
||||
'filename="%s"\r\n' % (file_field, filename)).encode()
|
||||
body += ("Content-Type: %s\r\n\r\n"
|
||||
% _content_type_for(filename)).encode()
|
||||
body += file_bytes + b"\r\n"
|
||||
body += ("--" + boundary + "--\r\n").encode()
|
||||
return body, boundary
|
||||
|
||||
|
||||
def _post_multipart(url, key, body, content_type,
|
||||
request_multipart=None):
|
||||
if request_multipart is not None:
|
||||
return request_multipart(url, key, body, content_type)
|
||||
req = urllib.request.Request(url, data=body, method="POST", headers={
|
||||
"Authorization": "Bearer " + key,
|
||||
"Content-Type": content_type})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=API_TIMEOUT) as resp:
|
||||
return resp.status, json.loads(resp.read().decode() or "{}")
|
||||
except urllib.error.HTTPError as exc:
|
||||
raw = exc.read().decode()
|
||||
try:
|
||||
payload = json.loads(raw or "{}")
|
||||
except ValueError:
|
||||
payload = {"raw": raw[:400]}
|
||||
return exc.code, payload
|
||||
except Exception as exc: # noqa: BLE001 - surfaced as tool error
|
||||
return 0, {"error": str(exc)[:300]}
|
||||
|
||||
|
||||
MAX_INPUT_BYTES = 20 * 1024 * 1024
|
||||
|
||||
|
||||
def edit(prompt, input_filename, output_filename=None,
|
||||
size="1024x1024", model=None, request_multipart=None,
|
||||
download=None, cwd=None):
|
||||
"""Edit an existing image in the working directory and save
|
||||
the result as a NEW file (the input is never overwritten).
|
||||
|
||||
``input_filename`` and ``output_filename`` must both be plain
|
||||
basenames in the working directory. The relay's
|
||||
``/v1/images/edits`` endpoint is multipart/form-data (verified
|
||||
2026-10-30: 200, URL PNG 1024x1024, input image tokens billed
|
||||
in usage). Returns the same dict shape as ``generate``.
|
||||
"""
|
||||
key, name = _validate_common(prompt, size, output_filename)
|
||||
if (os.path.basename(input_filename or "") != input_filename
|
||||
or input_filename in (".", "..")
|
||||
or "/" in (input_filename or "")
|
||||
or "\\" in (input_filename or "")):
|
||||
raise ImageGenError(
|
||||
"input_filename must be a plain basename, got %r"
|
||||
% input_filename)
|
||||
workdir = os.path.abspath(cwd or os.getcwd())
|
||||
input_path = os.path.join(workdir, input_filename)
|
||||
if not os.path.isfile(input_path):
|
||||
raise ImageGenError("input image not found: %r" % input_filename)
|
||||
with open(input_path, "rb") as fh:
|
||||
file_bytes = fh.read()
|
||||
if len(file_bytes) > MAX_INPUT_BYTES:
|
||||
raise ImageGenError(
|
||||
"input image too large: %d bytes (max %d)"
|
||||
% (len(file_bytes), MAX_INPUT_BYTES))
|
||||
if os.path.abspath(os.path.join(workdir, name)) == input_path:
|
||||
raise ImageGenError(
|
||||
"output_filename must differ from input_filename")
|
||||
use_model = model or _default_model()
|
||||
body, boundary = _build_multipart(
|
||||
[("model", use_model), ("prompt", str(prompt)),
|
||||
("size", size)],
|
||||
"image", input_filename, file_bytes)
|
||||
status, data = _post_multipart(
|
||||
_base_url() + "/v1/images/edits", key, body,
|
||||
"multipart/form-data; boundary=" + boundary,
|
||||
request_multipart=request_multipart)
|
||||
result = _save_image_response(status, data, name, use_model, size,
|
||||
download, cwd, "edit")
|
||||
result["input_filename"] = input_filename
|
||||
return result
|
||||
|
||||
|
||||
# ------------------------------------------------------- codex-home glue
|
||||
|
||||
def ensure_codex_home_imagegen(codex_home_dir, command_args=None,
|
||||
|
||||
Reference in new issue
Block a user