이 페이지에서는 간단한 HTTP 제출부터 실시간 이미지 출력과의 완전한 WebSocket 통합까지, ComfyUI 서버 API와 상호작용하는 세 가지 방법을 보여줍니다.
모든 예제는 설명을 위해 기본 SD1.5 워크플로우 를 사용합니다. API를 사용하기 전에 API 형식 으로 워크플로우를 내보내야 합니다.
이 예제들은 표준 라이브러리와 websocket-client 패키지를 사용한 파이썬 코드입니다(pip install websocket-client). 언어에 관계없이 기본 API 프로토콜은 동일합니다—TypeScript 및 curl equivalent에 대한 클라우드 API 참조 를 확인하세요.
방법 1: 제출 후 잊어버리기 (HTTP 전용)
소스: basic_api_example.py
가장 간단한 방식: 워크플로우를 제출하고 결과를 기다리지 않습니다. 나중에 출력물을 확인하는 일회성 작업에 유용합니다.
"""basic_api_example.py — HTTP만을 통해 워크플로우를 제출합니다."""
import json
from urllib import request
SERVER_ADDRESS = "127.0.0.1:8188"
def queue_prompt ( prompt ):
p = { "prompt" : prompt}
data = json.dumps(p).encode( "utf-8" )
req = request.Request(
f "http:// { SERVER_ADDRESS } /prompt" , data = data
)
request.urlopen(req)
if __name__ == "__main__" :
# API 형식으로 내보낸 워크플로우를 로드합니다
prompt_text = """{
"3": {
"class_type": "KSampler",
"inputs": {
"cfg": 8, "denoise": 1,
"latent_image": ["5", 0],
"model": ["4", 0],
"negative": ["7", 0],
"positive": ["6", 0],
"sampler_name": "euler",
"scheduler": "normal",
"seed": 8566257, "steps": 20
}
},
"4": {
"class_type": "CheckpointLoaderSimple",
"inputs": {"ckpt_name": "v1-5-pruned-emaonly.safetensors"}
},
"5": {
"class_type": "EmptyLatentImage",
"inputs": {"batch_size": 1, "height": 512, "width": 512}
},
"6": {
"class_type": "CLIPTextEncode",
"inputs": {"clip": ["4", 1], "text": "masterpiece best quality man"}
},
"7": {
"class_type": "CLIPTextEncode",
"inputs": {"clip": ["4", 1], "text": "bad hands"}
},
"8": {
"class_type": "VAEDecode",
"inputs": {"samples": ["3", 0], "vae": ["4", 2]}
},
"9": {
"class_type": "SaveImage",
"inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]}
}
}"""
prompt = json.loads(prompt_text)
# 제출 전 입력값 수정
prompt[ "3" ][ "inputs" ][ "seed" ] = 5
prompt[ "6" ][ "inputs" ][ "text" ] = "masterpiece best quality man"
queue_prompt(prompt)
print ( "프롬프트가 성공적으로 제출되었습니다." )
이 방법은 SaveImage 노드를 사용하며, 이 노드는 이미지를 서버의 디스크에 저장합니다. 이를 가져오려면 /view?filename=... 호출을 추가로 실행해야 합니다.
방법 2: WebSocket + 히스토리 (완료 모니터링)
소스: websockets_api_example.py
WebSocket을 사용해 실행이 완료될 때까지 기다린 후, /history 엔드포인트를 통해 출력물을 가져옵니다. 대부분의 사용 사례에 권장되는 방식입니다.
"""websockets_api_example.py — WebSocket을 통해 실행 상태를 모니터링하고, /history를 통해 다운로드합니다."""
import websocket # pip install websocket-client
import uuid
import json
import urllib.request
import urllib.parse
SERVER_ADDRESS = "127.0.0.1:8188"
client_id = str (uuid.uuid4())
def queue_prompt ( prompt , prompt_id ):
p = { "prompt" : prompt, "client_id" : client_id, "prompt_id" : prompt_id}
data = json.dumps(p).encode( "utf-8" )
req = urllib.request.Request(
f "http:// { SERVER_ADDRESS } /prompt" , data = data
)
urllib.request.urlopen(req)
def get_image ( filename , subfolder , folder_type ):
params = urllib.parse.urlencode({
"filename" : filename,
"subfolder" : subfolder,
"type" : folder_type,
})
with urllib.request.urlopen(
f "http:// { SERVER_ADDRESS } /view? { params } "
) as response:
return response.read()
def get_history ( prompt_id ):
with urllib.request.urlopen(
f "http:// { SERVER_ADDRESS } /history/ { prompt_id } "
) as response:
return json.loads(response.read())
def get_images ( ws , prompt ):
prompt_id = str (uuid.uuid4())
queue_prompt(prompt, prompt_id)
while True :
out = ws.recv()
if isinstance (out, str ):
message = json.loads(out)
if message[ "type" ] == "executing" :
data = message[ "data" ]
if data[ "node" ] is None and data[ "prompt_id" ] == prompt_id:
break # 실행 완료
# 이진 프레임은 미리보기 이미지이므로 여기서 건너뜁니다
continue
history = get_history(prompt_id)[prompt_id]
output_images = {}
for node_id in history[ "outputs" ]:
node_output = history[ "outputs" ][node_id]
images_output = []
if "images" in node_output:
for image in node_output[ "images" ]:
image_data = get_image(
image[ "filename" ], image[ "subfolder" ], image[ "type" ]
)
images_output.append(image_data)
output_images[node_id] = images_output
return output_images
if __name__ == "__main__" :
prompt_text = """{
"3": { ... }, "4": { ... }, "5": { ... },
"6": { ... }, "7": { ... }, "8": { ... },
"9": { "class_type": "SaveImage", "inputs": { ... } }
}"""
prompt = json.loads(prompt_text)
prompt[ "3" ][ "inputs" ][ "seed" ] = 5
prompt[ "6" ][ "inputs" ][ "text" ] = "masterpiece best quality man"
ws = websocket.WebSocket()
ws.connect( f "ws:// { SERVER_ADDRESS } /ws?clientId= { client_id } " )
images = get_images(ws, prompt)
ws.close()
print ( f " { len (images) } 개의 출력 노드에서 이미지를 받았습니다." )
# 이미지 표시(필요한 경우 Pillow 사용):
# for node_id in images:
# for image_data in images[node_id]:
# from PIL import Image
# import io
# img = Image.open(io.BytesIO(image_data))
# img.show()
WebSocket 이진 프레임에는 생성 중인 미리보기 이미지가 포함됩니다. 이를 디코딩하여 실시간 미리보기를 볼 수 있습니다(서버 메시지 페이지에서 이진 형식을 확인하세요).
방법 3: SaveImageWebsocket을 사용한 WebSocket (실시간 이미지)
소스: websockets_api_example_ws_images.py
이미지를 디스크에 저장하지 않고 싶은 경우, SaveImageWebsocket 노드를 사용하세요. 이미지는 WebSocket 이진 프레임을 통해 바로 전송됩니다.
"""websockets_api_example_ws_images.py — WebSocket을 통해 직접 이미지를 수신합니다."""
import websocket # pip install websocket-client
import uuid
import json
import urllib.request
import urllib.parse
SERVER_ADDRESS = "127.0.0.1:8188"
client_id = str (uuid.uuid4())
def queue_prompt ( prompt ):
p = { "prompt" : prompt, "client_id" : client_id}
data = json.dumps(p).encode( "utf-8" )
req = urllib.request.Request(
f "http:// { SERVER_ADDRESS } /prompt" , data = data
)
return json.loads(urllib.request.urlopen(req).read())
def get_images ( ws , prompt ):
prompt_id = queue_prompt(prompt)[ "prompt_id" ]
output_images = {}
current_node = ""
while True :
out = ws.recv()
if isinstance (out, str ):
message = json.loads(out)
if message[ "type" ] == "executing" :
data = message[ "data" ]
if data[ "prompt_id" ] == prompt_id:
if data[ "node" ] is None :
break # 실행 완료
current_node = data[ "node" ]
else :
# 이진 프레임 — SaveImageWebsocket에서 온 이미지 데이터
if current_node == "save_image_websocket_node" :
images_output = output_images.get(current_node, [])
# 처음 8바이트는 타입/메타데이터, 나머지는 이미지 데이터
images_output.append(out[ 8 :])
output_images[current_node] = images_output
return output_images
if __name__ == "__main__" :
prompt_text = """{
"3": { "class_type": "KSampler", "inputs": { ... } },
...
"save_image_websocket_node": {
"class_type": "SaveImageWebsocket",
"inputs": {"images": ["8", 0]}
}
}"""
prompt = json.loads(prompt_text)
prompt[ "3" ][ "inputs" ][ "seed" ] = 5
ws = websocket.WebSocket()
ws.connect( f "ws:// { SERVER_ADDRESS } /ws?clientId= { client_id } " )
images = get_images(ws, prompt)
ws.close()
print ( f " { len (images) } 개의 이미지를 WebSocket을 통해 받았습니다." )
# 표시(필요한 경우 Pillow 사용):
# for image_data in images.get("save_image_websocket_node", []):
# from PIL import Image
# import io
# img = Image.open(io.BytesIO(image_data))
# img.show()
워크플로우는 일반 SaveImage 노드 대신 class_type: "SaveImageWebsocket"를 가진 노드를 사용해야 합니다.
어떤 방법을 사용해야 할까요?
방법 1: HTTP 전용 즉시 처리 후 잊어버리기. 즉각적인 출력이 필요하지 않거나, 나중에 결과를 가져오는 것이 괜찮을 때 사용하세요.
방법 2: WebSocket + 히스토리 권장. 실행이 완료될 때까지 기다렸다가 출력물을 다운로드하세요. 간편함과 신뢰성의 최적 조화입니다.
방법 3: SaveImageWebsocket 실시간 이미지. 디스크 쓰기 없이 이미지를 전달받고자 하는 인터랙티브 앱에 가장 적합합니다.
전체 API 참조(엔드포인트, 페이로드 형식, 오류 처리)는 서버 라우트 및 서버 메시지 페이지를 참고하세요.