ComfyUI MCP Server

Instructions

    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.
    

Implementation Reference

• src/comfy_mcp_server/tools/system.py:37-58 (handler)

  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)