ComfyUI API Python: تشغيل سير العمل ونقاط النهاية والأمثلة

يبدو ComfyUI كلوحة رسم لعقد سير العمل، لكنه تحت الواجهة خادم HTTP عادي. كل زر تضغطه في المتصفح يستدعي نقطة نهاية، ويستطيع سكربت Python استدعاء نقاط النهاية نفسها. هذه هي الفكرة الأساسية وراء ComfyUI API مع Python: صدّر سير العمل بصيغة JSON، وغيّر قيمتين أو ثلاثًا، ثم أرسله بطريقة POST إلى /prompt، واجمع الصور المكتملة بعد ذلك. لا حاجة إلى تبويب متصفح، ولا نقرات، ولا مراقبة مستمرة. يعرض هذا المقال نقاط النهاية الفعلية، ومستمع WebSocket لتتبع التقدم المباشر، وأمثلة عملية يمكنك لصقها في ملف وتشغيلها على جهازك.

لماذا تكتب سكربتًا يعمل مع ComfyUI أصلًا

صنع صورة واحدة يدويًا أمر مقبول. أما صنع مئتي صورة منتج، أو تشغيل مهمة مصغرات ليلية، أو السماح للعملاء بالضغط على زر داخل تطبيقك، فهذا أمر مختلف، ولا تستطيع اللوحة فعل أي منه. يستطيع خادم ComfyUI بدون واجهة رسومية (headless) ذلك، ويُعد Python أقصر طريق إليه. وكمكافأة، تنتهي أوامرك النصية وبذورك وإعداداتك في مستودع Git بدلًا من مجلد لقطات الشاشة.

تبرّر الواجهة البرمجية استخدامها في ثلاث حالات:

• العمل الدفعي: مئات الأوامر النصية، وقالب واحد، وصفر نقرات يدوية.
• المنتجات: يرسل تطبيقك طلبًا ويستلم صورة في المقابل.
• الأتمتة: مهمة cron أو روبوت محادثة أو خطوة CI تولّد الأصول وفق جدول زمني.

ما الذي يكشفه الخادم

ابدأ ComfyUI بالطريقة المعتادة (python main.py) فيستمع على 127.0.0.1:8188. أضف --listen 0.0.0.0 لقبول الاتصالات من أجهزة أخرى، واستخدم --port لتغيير المنفذ. هذه أكثر نقاط النهاية التي ستستخدمها:

الإيقاع لا يتغير أبدًا: أضِف إلى الطابور، ثم انتظر، ثم احضر النتائج. ترسل الرسم البياني بطريقة POST، ثم تنتظر (بالاستعلام المتكرر أو بالاستماع إلى WebSocket)، ثم تحمّل ما أنتجه الرسم البياني.

صدّر سير عملك بصيغة API

ابنِ الرسم البياني واختبره على اللوحة أولًا. عندما يُنتج الصورة التي تريدها، صدّره. لا يستطيع سكربتك تشغيل ملف سير العمل العادي، لأن هذه الصيغة تخزن مواضع العقد والألوان وتخطيط الأدوات. يحتاج السكربت إلى النسخة الخفيفة، حيث تُختزل كل عقدة إلى فئتها ومدخلاتها.

صيغة API مقابل JSON العادي

في الواجهات الحديثة، افتح قائمة Workflow واختر Export (API). في الإصدارات القديمة، فعّل Dev mode options في الإعدادات، ثم استخدم زر Save (API Format). احفظ الناتج باسم workflow_api.json بجانب سكربتك.

💡 نصيحة: احتفظ بالملفين معًا. يُعاد فتح JSON العادي على اللوحة للتحرير، بينما ملف JSON الخاص بالواجهة البرمجية هو ما يرسله كودك.

تشريح JSON المُصدَّر

افتح الملف وستجد قاموسًا مسطحًا. كل مدخل مسمّى بمعرّف العقدة (نص)، وتحتوي قيمته على class_type إضافة إلى inputs الخاصة بالعقدة:

{
  "3": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 421337,
      "steps": 20,
      "cfg": 1.0,
      "sampler_name": "euler",
      "scheduler": "simple",
      "denoise": 1.0,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["5", 0]
    }
  },
  "4": {
    "class_type": "CheckpointLoaderSimple",
    "inputs": { "ckpt_name": "flux1-dev-fp8.safetensors" }
  },
  "6": {
    "class_type": "CLIPTextEncode",
    "inputs": { "text": "a lighthouse at dawn", "clip": ["4", 1] },
    "_meta": { "title": "Positive Prompt" }
  }
}

نسخ الأمر النصي

يفترض هذا المثال استخدام checkpoint من Flux Dev، ولهذا تبقى cfg عند 1.0. أما checkpoints بأسلوب Stable Diffusion 3.5 Large فغالبًا تحتاج قيمة أعلى، تتراوح عادةً بين 4 و8، لذا انسخ الأرقام من ملف التصدير الخاص بك لا من أحد الدروس.

هناك تفصيلان مهمان. القيم العادية، مثل seed أو steps، هي المقابض التي تغيّرها من Python. أما القيم مثل ["4", 0] فهي روابط: العنصر الأول هو معرّف العقدة المصدر، والثاني هو فتحة المخرجات التي تُقرأ. لا تلمس الروابط إلا إذا كنت تعيد توصيل الرسم البياني عن قصد.

💡 نصيحة: أعد تسمية عقد الأوامر النصية على اللوحة ("Positive Prompt" و "Negative Prompt") قبل التصدير. يظهر الاسم في _meta.title، ويستطيع كودك العثور على العقد بعنوانها بدلًا من رقم هش.

أول استدعاء لك في Python

تكفي حزمتان لكل ما في هذا المقال: pip install requests websocket-client. احفظ التصدير باسم workflow_api.json، ثم شغّل ComfyUI، ونفّذ المقتطفات بالترتيب.

تثبيت الحزم وإضافة أمر إلى الطابور

import json
import requests

SERVER = "http://127.0.0.1:8188"

with open("workflow_api.json", "r", encoding="utf-8") as f:
    workflow = json.load(f)

# "6" is the positive CLIPTextEncode, "3" is the KSampler
workflow["6"]["inputs"]["text"] = "a lighthouse at dawn, 35mm photo, film grain"
workflow["3"]["inputs"]["seed"] = 421337

response = requests.post(f"{SERVER}/prompt", json={"prompt": workflow})
response.raise_for_status()
prompt_id = response.json()["prompt_id"]
print("Queued:", prompt_id)

نسخ الأمر النصي

يرد ComfyUI بجسم JSON يتضمن prompt_id إلى جانب number، وهو موضع الأمر في الطابور. لم يُنجز أي تصيير في هذه المرحلة. الأمر قُبل فحسب. احتفظ بالمعرّف، فكل استدعاء لاحق يحتاجه.

الاستعلام عن نقطة السجل

أبسط طريقة لمعرفة انتهاء الأمر هي سؤال نقطة السجل حتى تجيب. وما دام التشغيل مستمرًا، يُعيد /history/{prompt_id} كائنًا فارغًا.

import time

def wait_for_outputs(prompt_id, timeout=300):
    started = time.time()
    while time.time() - started < timeout:
        history = requests.get(f"{SERVER}/history/{prompt_id}").json()
        if prompt_id in history:
            return history[prompt_id]["outputs"]
        time.sleep(1)
    raise TimeoutError(f"Prompt {prompt_id} took longer than {timeout}s")

نسخ الأمر النصي

يُفهرس القاموس outputs حسب معرّف العقدة. تُبلغ كل عقدة SaveImage عن قائمة اسمها images، وكل صورة قاموس صغير يتضمن filename، subfolder، type.

تحميل الصورة المكتملة

import os

def download_images(outputs, folder="renders"):
    os.makedirs(folder, exist_ok=True)
    saved = []
    for node_id, node_output in outputs.items():
        for image in node_output.get("images", []):
            if image["type"] != "output":
                continue  # skip PreviewImage temp files
            data = requests.get(f"{SERVER}/view", params=image).content
            path = os.path.join(folder, image["filename"])
            with open(path, "wb") as f:
                f.write(data)
            saved.append(path)
    return saved

print(download_images(wait_for_outputs(prompt_id)))

نسخ الأمر النصي

يحتوي قاموس الصورة بالفعل على المعاملات الثلاثة التي تتوقعها /view، لذا يمكن تمريره مباشرة كسلسلة استعلام. تُبلغ عقد SaveImage عن type: "output"، بينما تُبلغ عقد PreviewImage عن temp، ولهذا تُصفّي الحلقة النتائج على هذا الأساس.

التقدم المباشر عبر WebSocket

الاستعلام المتكرر يعمل، لكنه يهدر الطلبات ولا يقول شيئًا حتى النهاية. يدعم ComfyUI أيضًا WebSocket، وهو ما يمنحك تغذية مباشرة: تغيّرات الطابور، والعقدة التي تعمل حاليًا، وعدّادًا لكل خطوة من خطوات أداة أخذ العينات. في تطبيق ويب، هذا ما يحرّك شريط التقدم.

الاتصال بمعرّف العميل

أنشئ UUID مرة واحدة واستخدمه في موضعين: سلسلة الاستعلام clientId الخاصة بالمقبس، وحقل client_id في طلب /prompt الخاص بك. يرسل ComfyUI أحداث الأمر فقط إلى العميل الذي أضافه إلى الطابور. إذا لم يتطابق المعرّفان، فسيبقى مقبسك صامتًا.

import json
import uuid
import requests
import websocket  # pip install websocket-client

HOST = "127.0.0.1:8188"
CLIENT_ID = str(uuid.uuid4())

def run_with_progress(workflow):
    ws = websocket.WebSocket()
    ws.connect(f"ws://{HOST}/ws?clientId={CLIENT_ID}")

    payload = {"prompt": workflow, "client_id": CLIENT_ID}
    r = requests.post(f"http://{HOST}/prompt", json=payload)
    r.raise_for_status()
    prompt_id = r.json()["prompt_id"]

    while True:
        message = ws.recv()
        if isinstance(message, bytes):
            continue  # binary frames are preview thumbnails
        event = json.loads(message)
        kind, data = event["type"], event["data"]

        if kind == "progress":
            print(f"step {data['value']}/{data['max']}")
        elif kind == "execution_error":
            raise RuntimeError(data.get("exception_message", "node failed"))
        elif data.get("prompt_id") == prompt_id and (
            kind == "execution_success"
            or (kind == "executing" and data["node"] is None)
        ):
            break

    ws.close()
    history = requests.get(f"http://{HOST}/history/{prompt_id}").json()
    return history[prompt_id]["outputs"]

نسخ الأمر النصي

تحمل الإطارات الثنائية صور المعاينة المصغّرة أثناء عمل أداة أخذ العينات، لذا تتجاهل الحلقة أي شيء ليس نصًا. إذا أردت عرض المعاينات المباشرة داخل واجهتك، فافكّ ترميز تلك الإطارات بدلًا من تجاهلها.

الرسائل التي ستستقبلها

تُعد رسالة executing التي يكون فيها node هو null إشارة انتهاء التشغيل الكلاسيكية. تضيف الإصدارات الأحدث execution_success، والتعامل مع الاثنتين يُبقي سكربتك يعمل عبر الإصدارات.

تغليفه في فئة عميل

الدوال المنفصلة مقبولة لأول اختبار. أما أي شيء يُنفَّذ أكثر من مرة فيستحق فئة صغيرة تخزّن المضيف ومعرّف العميل والعمليات التي تكررها.

import time
import uuid
import requests

class ComfyClient:
    def __init__(self, host="127.0.0.1:8188"):
        self.host = host
        self.client_id = str(uuid.uuid4())

    def queue(self, workflow):
        r = requests.post(
            f"http://{self.host}/prompt",
            json={"prompt": workflow, "client_id": self.client_id},
        )
        if r.status_code != 200:
            raise RuntimeError(r.text)  # includes node_errors
        return r.json()["prompt_id"]

    def result(self, prompt_id, timeout=300):
        deadline = time.time() + timeout
        while time.time() < deadline:
            history = requests.get(f"http://{self.host}/history/{prompt_id}").json()
            if prompt_id in history:
                return history[prompt_id]
            time.sleep(1)
        raise TimeoutError(prompt_id)

    def fetch(self, image):
        r = requests.get(f"http://{self.host}/view", params=image)
        r.raise_for_status()
        return r.content

    def upload(self, path):
        with open(path, "rb") as f:
            r = requests.post(
                f"http://{self.host}/upload/image",
                files={"image": f},
                data={"overwrite": "true"},
            )
        r.raise_for_status()
        return r.json()["name"]

نسخ الأمر النصي

تبديل الأوامر النصية والبذور بأمان

لا تكتب معرّفات العقد مثل "6" بشكل ثابت في مشروع حقيقي. أعد تصدير الرسم البياني فقد تتغير الأرقام. ابحث عن العقد حسب الفئة والعنوان بدلًا من ذلك، وعدّل دائمًا نسخة من القالب حتى لا ينتقل أي عمل إلى التالي.

import copy
import json
import random

def find_node(workflow, class_type, title=None):
    for node_id, node in workflow.items():
        if node["class_type"] != class_type:
            continue
        if title is None or node.get("_meta", {}).get("title") == title:
            return node_id
    raise LookupError(f"{class_type} {title or ''} not found")

def build(template, prompt, seed=None):
    wf = copy.deepcopy(template)
    wf[find_node(wf, "CLIPTextEncode", "Positive Prompt")]["inputs"]["text"] = prompt
    wf[find_node(wf, "KSampler")]["inputs"]["seed"] = (
        seed if seed is not None else random.randint(0, 2**32 - 1)
    )
    return wf

template = json.load(open("workflow_api.json", encoding="utf-8"))
client = ComfyClient()

prompts = [
    "ceramic teapot on a linen cloth, soft window light",
    "walnut desk with a fountain pen, low morning sun",
    "leather boots on wet cobblestones, overcast sky",
]

ids = [client.queue(build(template, p)) for p in prompts]  # queue everything first
for pid in ids:
    entry = client.result(pid)
    for out in entry["outputs"].values():
        for image in out.get("images", []):
            with open(image["filename"], "wb") as f:
                f.write(client.fetch(image))

نسخ الأمر النصي

ضع كل شيء في الطابور أولًا، ثم اجمع النتائج. تُشغّل ComfyUI الأوامر النصية واحدًا تلو الآخر حسب ترتيب وصولها، لذلك لا تبقى GPU في وضع الخمول أثناء تنزيل سكربتك لأي ملف.

💡 نصيحة: تحتاج إلى 200 أمر بدلًا من ثلاثة؟ اطلب من نموذج لغوي مثل Claude Sonnet 5 أو Gemini 3.5 Flash كتابتها على شكل قائمة JSON، ثم مرر تلك القائمة مباشرة إلى الحلقة أعلاه.

رفع الصور لتعديلها

تبدأ رسومات تحويل الصورة إلى صورة، والتعديل الموضعي، ورسومات ControlNet بعقدة LoadImage. تقرأ هذه العقدة من مجلد الإدخال في ComfyUI، لذا ارفع الملف أولًا ووجّه العقدة إلى الاسم الذي أعاده الخادم.

name = client.upload("portrait.png")
wf = copy.deepcopy(template)
wf[find_node(wf, "LoadImage")]["inputs"]["image"] = name
pid = client.queue(wf)

نسخ الأمر النصي

هنا يقترب السكربت من أعمال المؤثرات البصرية. إزالة الأجسام، وتبديل الخلفيات، وإعادة الإضاءة كلها هي الحلقة نفسها: ارفع صورة مصدرية، واضبط قناعًا وأمرًا نصيًا، ثم أضف إلى الطابور، ثم اجلب النتيجة. غلّفها في دالة، فيتحول مجلد من 500 صورة إلى أمر واحد.

أخطاء الإنتاج التي تجب تجنبها

تعمل السكربتات على حاسوبك المحمول، لكنها تفشل بطرق متوقعة بمجرد تشغيلها دون مراقبة. تسبب ثلاث مشكلات معظم أسئلة الدعم.

الأوامر المخزنة مؤقتًا تعود فورًا

يخزّن ComfyUI نتائج العقد مؤقتًا حسب مدخلاتها. إذا أضفت الرسم البياني نفسه إلى الطابور مرتين، فلن ينفذ التشغيل الثاني أي شيء، فتحصل على الصورة نفسها خلال أجزاء من الثانية. خيار "randomize seed after each run" موجود فقط في واجهة المتصفح. أما JSON الخاص بالواجهة البرمجية فيحمل رقمًا ثابتًا، لذا يجب على كودك اختيار بذرة جديدة كلما أراد صورة جديدة.

قراءة node_errors بشكل صحيح

عندما يفشل التحقق، يرد /prompt برمز HTTP 400 وجسم يتضمن error إلى جانب node_errors. يحدد الحقل الثاني معرّف العقدة والمدخل المحددين المسؤولين عن الخطأ، مثل اسم ملف checkpoint غير مثبت على هذا الجهاز. اطبع الجسم كاملًا، لا رمز الحالة وحده. وتذكّر أيضًا أن الأمر قد يجتاز التحقق ثم يفشل أثناء التنفيذ، وعندها يُظهر إدخال السجل الحالة status_str: "error".

لا تعرّض المنفذ 8188 أبدًا

يأتي ComfyUI بدون تسجيل دخول. أي شخص يستطيع الوصول إلى المنفذ يمكنه إضافة مهام إلى الطابور، وقراءة مجلد المخرجات، واستدعاء /object_info. العقد المخصصة هي كود Python عادي وتعمل بصلاحيات مستخدمك. اربط الخادم بالعنوان 127.0.0.1، أو ضعه خلف وكيل عكسي مع مصادقة أو VPN. إذا كان تطبيق متصفح على أصل آخر يجب أن يستدعيه، فمرر --enable-cors-header مع ذلك الأصل الواحد بدلًا من استخدام wildcard.

💡 نصيحة: تنفد VRAM لديك بعد عدد كبير من checkpoints مختلفة؟ أرسل {"unload_models": true, "free_memory": true} إلى /free بين الدفعات لتحرير الذاكرة دون إعادة تشغيل الخادم.

تخطَّ الخادم مع Picasso IA

لا يحتاج كل مشروع إلى صندوق GPU وبيئة Python وطابور للمراقبة. إذا كان هدفك ببساطة صورًا جيدة من نصوص أو صور مرجعية، فإن Picasso IA يشغّل نماذج مماثلة في المتصفح. إليك كيف تتوافق النماذج مع مهام ComfyUI الشائعة:

توليد الصورة في ست خطوات

إليك العملية كاملة باستخدام Flux 2 Pro، النموذج الأقرب إلى سير عمل ComfyUI بصور مرجعية:

1. افتح صفحة Flux 2 Pro على Picasso IA.
2. اكتب الأمر النصي. سمِّ الشخص والإضاءة والعدسة، بالطريقة نفسها التي تكتب بها في عقدة النص داخل ComfyUI.
3. اختر نسبة العرض إلى الارتفاع. القيمة الافتراضية هي 1:1. تناسب 16:9 اللافتات، بينما تحافظ match_input_image على شكل الصورة المرفوعة.
4. اختر الدقة. القيمة الافتراضية 1 MP، ويقبل النموذج حتى 4 MP، لكن يُوصى بدقة 2 MP أو أقل.
5. أضف حتى 8 صور إدخال إذا أردت أن تتبع النتيجة أسلوبًا أو وجهًا أو لقطة منتج.
6. اضبط صيغة الإخراج (WebP أو JPG أو PNG)، ثم اضغط توليد. أعد استخدام قيمة البذرة لاحقًا لإعادة إنتاج النتيجة نفسها.

💡 نصيحة: العادات التي اكتسبتها أعلاه تنتقل مباشرة. بذور ثابتة لنتائج قابلة للتكرار، وتغيير واحد في كل تشغيل، وأوامر نصية قصيرة تسمي الإضاءة والعدسة تعمل بالطريقة نفسها على المنصتين.

أنشئ صورك اليوم

لديك الآن الحلقة الكاملة: صدّر الرسم البياني، وأضفه إلى الطابور، واستمع إلى WebSocket، واجلب الملفات. شغّل المقتطف الأول الليلة وستحصل على صورة على القرص في وقت قصير جدًا. ثم حسّن فئة العميل، وأضف منطق البذور، ودع دفعة تعمل بينما تفعل شيئًا آخر.

وإذا كنت تفضّل تجاوز الإعداد، فافتح Picasso IA، واختر Flux Dev أو Flux 2 Pro، واكتب أول أمر نصي يخطر ببالك. غيّر إعدادًا واحدًا، وأعد التوليد، وقارن. خمس دقائق من التجريب ستعلّمك عن الأوامر النصية والبذور ونسب العرض إلى الارتفاع أكثر من أي قدر من القراءة.