ComfyUI MCP Server

Instructions

Implementation Reference

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

  MCP tool handler for 'sync_official_templates'. Decorated with @mcp.tool. Logs progress via context and delegates to TemplateManager.sync_official_templates(), returning the result or raising ToolError on failure.
  @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 method that calls OfficialTemplateManager.sync_official_templates() and formats the success/error response with synced count.
  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)

  Core implementation in OfficialTemplateManager. Fetches template list from GitHub API, downloads and converts JSON workflows to DSL concurrently, handles caching, retries, filtering, and statistics. Returns dict of synced templates.
  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