ComfyUI MCP Server

Instructions

Implementation Reference

• comfy_mcp/mcp/server.py:547-575 (handler)

  MCP tool handler for 'sync_official_templates'. This is the entry point registered with @mcp.tool. It logs progress using the context and delegates to TemplateManager.sync_official_templates().
  @mcp.tool async def sync_official_templates(ctx: Context) -> dict: """Sync official ComfyUI templates from GitHub. Downloads and processes official workflow templates from the Comfy-Org/workflow_templates repository. Converts them to DSL format for use with the template system. Returns: Sync status with count of successfully processed templates Examples: sync_official_templates() """ try: await ctx.info("🔄 Starting sync of official ComfyUI templates...") result = await template_manager.sync_official_templates() if result["status"] == "success": await ctx.info(f"✅ Successfully synced {result['synced_count']} official templates") else: await ctx.info(f"❌ Sync failed: {result['error']}") return result except Exception as e: raise ToolError(f"Error syncing official templates: {e}")
• comfy_mcp/templates/manager.py:49-64 (helper)

  TemplateManager.sync_official_templates(): Intermediate helper method that calls OfficialTemplateManager.sync_official_templates() and formats the response.
  async def sync_official_templates(self) -> Dict[str, Any]: """Sync official ComfyUI templates.""" try: templates = await official_manager.sync_official_templates() self.official_templates_synced = True return { "status": "success", "synced_count": len(templates), "templates": list(templates.keys()) } except Exception as e: return { "status": "error", "error": str(e) }
• comfy_mcp/templates/official.py:72-144 (helper)

  OfficialTemplateManager.sync_official_templates(): Core implementation that fetches templates from GitHub API, downloads workflows, converts JSON to DSL, handles caching and errors with concurrency control.
  async def sync_official_templates(self) -> Dict[str, OfficialTemplate]: """Sync all official templates and convert to DSL.""" sync_start_time = time.time() print("🔄 Syncing official ComfyUI templates...") print(f"📋 Config: max_concurrent={self.config.max_concurrent_downloads}, " f"timeout={self.config.request_timeout}s, retries={self.config.max_retries}") # Reset stats self.sync_stats = { "total_attempted": 0, "successful": 0, "failed": 0, "skipped": 0, "conversion_failures": 0 } try: # Fetch template list (files in /templates directory) template_list = await self.fetch_template_list() # Filter for .json files only and apply configuration filters json_files = [] for item in template_list: if item["type"] == "file" and item["name"].endswith(".json"): file_size = item.get("size", 0) if self.config.should_sync_template(item["name"], file_size): json_files.append(item) else: self.sync_stats["skipped"] += 1 print(f"⏭️ Skipped {item['name']} (filtered by config)") print(f"📊 Found {len(json_files)} templates to sync ({self.sync_stats['skipped']} skipped by filters)") # Process templates with concurrency control semaphore = asyncio.Semaphore(self.config.max_concurrent_downloads) tasks = [self._process_template(semaphore, json_file, template_list) for json_file in json_files] results = await asyncio.gather(*tasks, return_exceptions=True) # Collect successful templates synced_templates = {} for result in results: if isinstance(result, Exception): self.sync_stats["failed"] += 1 print(f"❌ Template processing failed: {result}") elif result is not None: template_name, template = result synced_templates[template_name] = template self.sync_stats["successful"] += 1 # Cache results await self._cache_templates(synced_templates) self.templates = synced_templates self.last_sync_time = sync_start_time sync_duration = time.time() - sync_start_time print(f"🎉 Sync completed in {sync_duration:.1f}s") print(f"📊 Stats: {self.sync_stats['successful']} successful, " f"{self.sync_stats['failed']} failed, " f"{self.sync_stats['skipped']} skipped, " f"{self.sync_stats['conversion_failures']} conversion failures") return synced_templates except Exception as e: print(f"❌ Failed to sync official templates: {e}") # Try to load from cache cached_templates = await self._load_cached_templates() if cached_templates: print(f"📁 Loaded {len(cached_templates)} templates from cache") return cached_templates
• comfy_mcp/mcp/server.py:547-575 (registration)

  The @mcp.tool decorator registers this function as an MCP tool named 'sync_official_templates'.
  @mcp.tool async def sync_official_templates(ctx: Context) -> dict: """Sync official ComfyUI templates from GitHub. Downloads and processes official workflow templates from the Comfy-Org/workflow_templates repository. Converts them to DSL format for use with the template system. Returns: Sync status with count of successfully processed templates Examples: sync_official_templates() """ try: await ctx.info("🔄 Starting sync of official ComfyUI templates...") result = await template_manager.sync_official_templates() if result["status"] == "success": await ctx.info(f"✅ Successfully synced {result['synced_count']} official templates") else: await ctx.info(f"❌ Sync failed: {result['error']}") return result except Exception as e: raise ToolError(f"Error syncing official templates: {e}")