sync_official_templates
Downloads and converts official ComfyUI workflow templates from GitHub to DSL format for use in the template system.
Instructions
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()
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
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) } - 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}")