cbpark84
1c919f1106
- run_batch(): sequential|concurrent|auto - chain(): 파이프라인 체이닝 - schema(): 스키마 자동 추출 - ResultCache: LRU+TTL 캐시 - preview_callback: 실시간 프리뷰 - upload_file/audio/video() - BridgeInputSEED: random/increment/fixed - MCPServer: MCP Tool 자동 생성
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""MCPServer — 워크플로우를 MCP Tool로 자동 변환."""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from .bridge import ComfyBridge
|
|
from .schema import extract_schema
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MCPServer:
|
|
"""
|
|
ComfyUI 워크플로우를 Model Context Protocol (MCP) Tool로 변환한다.
|
|
|
|
클라우드 AI 플랫폼(Claude, ChatGPT 등)이 ComfyUI 워크플로우를 직접 호출할 수 있게 한다.
|
|
|
|
사용 예::
|
|
server = MCPServer(bridge)
|
|
server.add_workflow("sdxl", "sdxl.json", "SDXL 텍스트-이미지")
|
|
server.write_mcp_json("tools.json")
|
|
"""
|
|
|
|
def __init__(self, bridge: ComfyBridge):
|
|
self.bridge = bridge
|
|
self.workflows: Dict[str, Dict[str, Any]] = {}
|
|
|
|
def add_workflow(
|
|
self, tool_name: str, workflow_path: str, description: str = ""
|
|
) -> None:
|
|
"""워크플로우를 MCP Tool로 등록한다."""
|
|
from .workflow import load_workflow
|
|
|
|
wf = load_workflow(workflow_path)
|
|
schema = extract_schema(wf)
|
|
|
|
self.workflows[tool_name] = {
|
|
"workflow_path": workflow_path,
|
|
"workflow": wf,
|
|
"schema": schema,
|
|
"description": description,
|
|
}
|
|
|
|
def to_mcp_tools(self) -> List[Dict[str, Any]]:
|
|
"""MCP Tool 포맷으로 변환한다."""
|
|
tools = []
|
|
|
|
for tool_name, wf_info in self.workflows.items():
|
|
schema = wf_info["schema"]
|
|
inputs_spec = schema.get("inputs", {})
|
|
|
|
# MCP inputSchema 생성
|
|
properties: Dict[str, Any] = {}
|
|
required = []
|
|
|
|
for input_name, input_spec in inputs_spec.items():
|
|
type_name = input_spec.get("type", "STRING")
|
|
if input_spec.get("required"):
|
|
required.append(input_name)
|
|
|
|
if type_name in ("STRING", "INT", "FLOAT"):
|
|
properties[input_name] = {
|
|
"type": _mcp_type(type_name),
|
|
"description": f"{input_name} ({type_name})",
|
|
}
|
|
else:
|
|
properties[input_name] = {
|
|
"type": "string",
|
|
"description": f"{input_name} ({type_name})",
|
|
}
|
|
|
|
tool = {
|
|
"name": tool_name,
|
|
"description": wf_info["description"] or f"Run {tool_name} workflow",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": properties,
|
|
"required": required if required else [],
|
|
},
|
|
}
|
|
tools.append(tool)
|
|
|
|
return tools
|
|
|
|
async def call_tool(self, tool_name: str, input_dict: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""MCP Tool을 실행한다."""
|
|
if tool_name not in self.workflows:
|
|
raise ValueError(f"Unknown tool: {tool_name}")
|
|
|
|
wf_info = self.workflows[tool_name]
|
|
workflow = wf_info["workflow"]
|
|
|
|
result = await self.bridge.run(workflow, inputs=input_dict)
|
|
return {
|
|
"outputs": dict(zip(result.keys(), [result.get(k) for k in result.keys()]))
|
|
}
|
|
|
|
def write_mcp_json(self, path: str) -> None:
|
|
"""MCP Tool 목록을 JSON 파일로 저장한다."""
|
|
tools = self.to_mcp_tools()
|
|
mcp_config = {
|
|
"mcpServers": {
|
|
"comfyui": {
|
|
"command": "python",
|
|
"args": ["-m", "comfybridgeio.mcp_client"],
|
|
"tools": tools,
|
|
}
|
|
}
|
|
}
|
|
with open(path, "w") as f:
|
|
json.dump(mcp_config, f, indent=2)
|
|
logger.info(f"[ComfyBridgeIO] Wrote MCP config to {path}")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"MCPServer({len(self.workflows)} workflows)"
|
|
|
|
|
|
def _mcp_type(type_name: str) -> str:
|
|
"""ComfyUI 타입을 JSON Schema 타입으로 변환."""
|
|
if type_name == "STRING":
|
|
return "string"
|
|
if type_name == "INT":
|
|
return "integer"
|
|
if type_name == "FLOAT":
|
|
return "number"
|
|
if type_name == "BOOLEAN":
|
|
return "boolean"
|
|
return "string"
|