get_queue_status
Check the execution status of workflows in ComfyUI by retrieving running and pending job details from the queue.
Instructions
Get current queue: running and pending jobs.
Returns:
- queue_running: List of currently executing workflows
- queue_pending: List of queued workflows waiting to run
- running_count: Number of running jobs
- pending_count: Number of pending jobs
Use this to check if workflows are executing or queued.
Tool call template
This tool declares no required parameters. Add optional parameters from the input schema as needed.
{
"id": 1,
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"arguments": {},
"name": "get_queue_status"
}
}Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The handler function for the 'get_queue_status' tool. It fetches the queue status from ComfyUI's /queue endpoint, validates it with QueueStatus model, adds computed fields, and returns the result or an error.
def get_queue_status(ctx: Context) -> dict: """Get current queue: running and pending jobs. Returns: - queue_running: List of currently executing workflows - queue_pending: List of queued workflows waiting to run - running_count: Number of running jobs - pending_count: Number of pending jobs Use this to check if workflows are executing or queued. """ ctx.info("Fetching queue status...") try: data = comfy_get("/queue") status = QueueStatus(**data) result = status.model_dump() result["running_count"] = status.running_count result["pending_count"] = status.pending_count result["is_empty"] = status.is_empty return result except Exception as e: return ErrorResponse.unavailable(str(e)).model_dump() - src/comfy_mcp_server/models.py:53-70 (schema)Pydantic model defining the structure and validation for the queue status response used in get_queue_status.
class QueueStatus(BaseModel): """Queue status from /queue endpoint.""" queue_running: list[list[Any]] = Field(default_factory=list) queue_pending: list[list[Any]] = Field(default_factory=list) @property def running_count(self) -> int: return len(self.queue_running) @property def pending_count(self) -> int: return len(self.queue_pending) @property def is_empty(self) -> bool: return self.running_count == 0 and self.pending_count == 0 - src/comfy_mcp_server/__init__.py:92-92 (registration)Top-level registration call that includes get_queue_status by invoking register_all_tools(mcp), which chains to register_system_tools.
register_all_tools(mcp) - src/comfy_mcp_server/tools/__init__.py:25-25 (registration)Specific registration for system tools, including get_queue_status, called within register_all_tools.
register_system_tools(mcp)