Phase 3: Server API Routes (/api/comfyui/*)

Add FastAPI routes for ComfyUI programmatic API:

Query endpoints:
- GET /api/comfyui/status - System stats (GPU, RAM, etc.)
- GET /api/comfyui/queue - Queue status (running/pending)
- DELETE /api/comfyui/queue - Clear the queue
- GET /api/comfyui/models - List available models
- GET /api/comfyui/history - List generation history
- GET /api/comfyui/history/{prompt_id} - Get specific result

Generation endpoints:
- POST /api/comfyui/generate - Text-to-image with full params
- POST /api/comfyui/workflow - Queue arbitrary API-format workflow

All endpoints protected with API key auth (X-API-Key header).
Pydantic models for request/response validation.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

TODO.md

@@ -27,17 +27,17 @@
   - Rich progress output via console
 
 ## Phase 3: Server API Routes (`tensors/server/comfyui_api_routes.py`)
-- [ ] Step 3.1: Create new router with query endpoints
+- [x] Step 3.1: Create new router with query endpoints
   - `GET /api/comfyui/status` - System stats
   - `GET /api/comfyui/queue` - Queue status
   - `DELETE /api/comfyui/queue` - Clear queue
   - `GET /api/comfyui/models` - List loaded models
   - `GET /api/comfyui/history` - List history
   - `GET /api/comfyui/history/{prompt_id}` - Get specific result
-- [ ] Step 3.2: Add generation endpoints
+- [x] Step 3.2: Add generation endpoints
   - `POST /api/comfyui/generate` - Text-to-image generation
   - `POST /api/comfyui/workflow` - Run arbitrary workflow
-- [ ] Step 3.3: Register router in server/__init__.py
+- [x] Step 3.3: Register router in server/__init__.py (with API key auth)
 
 ## Phase 4: Configuration (`tensors/config.py`)
 - [ ] Step 4.1: Add ComfyUI config functions

tensors/server/__init__.py

@@ -12,6 +12,7 @@ from scalar_fastapi import get_scalar_api_reference
 from tensors.config import get_server_api_key
 from tensors.server.auth_routes import create_auth_router
 from tensors.server.civitai_routes import create_civitai_router
+from tensors.server.comfyui_api_routes import create_comfyui_api_router
 from tensors.server.comfyui_routes import create_comfyui_router
 from tensors.server.db_routes import create_db_router
 from tensors.server.download_routes import create_download_router
@@ -77,6 +78,7 @@ def create_app() -> FastAPI:
     app.include_router(create_db_router(), dependencies=[Depends(verify_api_key)])
     app.include_router(create_gallery_router(), dependencies=[Depends(verify_api_key)])
     app.include_router(create_download_router(), dependencies=[Depends(verify_api_key)])
+    app.include_router(create_comfyui_api_router(), dependencies=[Depends(verify_api_key)])
     return app
 
 

tensors/server/comfyui_api_routes.py

@@ -0,0 +1,277 @@
+"""FastAPI route handlers for ComfyUI programmatic API endpoints."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query
+from pydantic import BaseModel as PydanticBaseModel
+from pydantic import Field
+
+from tensors.comfyui import (
+    clear_queue,
+    generate_image,
+    get_history,
+    get_loaded_models,
+    get_queue_status,
+    get_system_stats,
+    queue_prompt,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/comfyui", tags=["ComfyUI API"])
+
+
+# =============================================================================
+# Request/Response Models
+# =============================================================================
+
+
+class GenerateRequest(PydanticBaseModel):
+    """Request body for text-to-image generation."""
+
+    prompt: str = Field(..., description="Positive prompt text")
+    negative_prompt: str = Field(default="", description="Negative prompt text")
+    model: str | None = Field(default=None, description="Checkpoint model name")
+    width: int = Field(default=1024, ge=64, le=4096, description="Image width")
+    height: int = Field(default=1024, ge=64, le=4096, description="Image height")
+    steps: int = Field(default=20, ge=1, le=150, description="Sampling steps")
+    cfg: float = Field(default=7.0, ge=1.0, le=30.0, description="CFG scale")
+    seed: int = Field(default=-1, description="Random seed (-1 for random)")
+    sampler: str = Field(default="euler", description="Sampler name")
+    scheduler: str = Field(default="normal", description="Scheduler name")
+
+
+class GenerateResponse(PydanticBaseModel):
+    """Response from text-to-image generation."""
+
+    success: bool
+    prompt_id: str
+    images: list[str] = Field(default_factory=list)
+    errors: dict[str, Any] = Field(default_factory=dict)
+
+
+class WorkflowRequest(PydanticBaseModel):
+    """Request body for running arbitrary workflow."""
+
+    workflow: dict[str, Any] = Field(..., description="ComfyUI API-format workflow")
+
+
+class WorkflowResponse(PydanticBaseModel):
+    """Response from workflow execution."""
+
+    success: bool
+    prompt_id: str
+    number: int | None = None
+    error: str | None = None
+    node_errors: dict[str, Any] = Field(default_factory=dict)
+
+
+class QueueStatusResponse(PydanticBaseModel):
+    """Queue status response."""
+
+    queue_running: list[Any] = Field(default_factory=list)
+    queue_pending: list[Any] = Field(default_factory=list)
+
+
+class SystemStatsResponse(PydanticBaseModel):
+    """System stats response."""
+
+    system: dict[str, Any] = Field(default_factory=dict)
+    devices: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class ModelsResponse(PydanticBaseModel):
+    """Available models response."""
+
+    checkpoints: list[str] = Field(default_factory=list)
+    loras: list[str] = Field(default_factory=list)
+    vae: list[str] = Field(default_factory=list)
+    clip: list[str] = Field(default_factory=list)
+    controlnet: list[str] = Field(default_factory=list)
+    upscale_models: list[str] = Field(default_factory=list)
+
+
+class HistoryEntry(PydanticBaseModel):
+    """Single history entry."""
+
+    prompt_id: str
+    status: str
+    outputs: dict[str, Any] = Field(default_factory=dict)
+    images: list[str] = Field(default_factory=list)
+
+
+# =============================================================================
+# Query Endpoints
+# =============================================================================
+
+
+@router.get("/status", response_model=SystemStatsResponse)
+def comfyui_status() -> dict[str, Any]:
+    """Get ComfyUI system stats (GPU, RAM, etc.)."""
+    stats = get_system_stats()
+    if not stats:
+        raise HTTPException(status_code=502, detail="Could not connect to ComfyUI")
+    return stats
+
+
+@router.get("/queue", response_model=QueueStatusResponse)
+def comfyui_queue() -> dict[str, Any]:
+    """Get ComfyUI queue status."""
+    queue = get_queue_status()
+    if queue is None:
+        raise HTTPException(status_code=502, detail="Could not connect to ComfyUI")
+    return queue
+
+
+@router.delete("/queue")
+def comfyui_clear_queue() -> dict[str, Any]:
+    """Clear the ComfyUI queue."""
+    success = clear_queue()
+    if not success:
+        raise HTTPException(status_code=502, detail="Could not clear queue")
+    return {"cleared": True}
+
+
+@router.get("/models", response_model=ModelsResponse)
+def comfyui_models() -> dict[str, Any]:
+    """List available models in ComfyUI."""
+    models = get_loaded_models()
+    if models is None:
+        raise HTTPException(status_code=502, detail="Could not fetch models from ComfyUI")
+    return models
+
+
+@router.get("/history")
+def comfyui_history_list(
+    limit: int = Query(default=20, le=100, description="Max history items"),
+) -> dict[str, Any]:
+    """List ComfyUI generation history."""
+    history = get_history(max_items=limit)
+    if history is None:
+        raise HTTPException(status_code=502, detail="Could not fetch history from ComfyUI")
+
+    # Transform to list format with summary
+    items: list[dict[str, Any]] = []
+    for prompt_id, entry in history.items():
+        status = entry.get("status", {}).get("status_str", "unknown")
+        outputs = entry.get("outputs", {})
+
+        # Extract image filenames
+        images: list[str] = []
+        for _node_id, output in outputs.items():
+            if "images" in output:
+                for img in output["images"]:
+                    images.append(img.get("filename", ""))
+
+        items.append(
+            {
+                "prompt_id": prompt_id,
+                "status": status,
+                "image_count": len(images),
+                "images": images,
+            }
+        )
+
+    return {"items": items, "total": len(items)}
+
+
+@router.get("/history/{prompt_id}")
+def comfyui_history_detail(prompt_id: str) -> dict[str, Any]:
+    """Get details for a specific history entry."""
+    history = get_history(prompt_id=prompt_id)
+    if history is None:
+        raise HTTPException(status_code=502, detail="Could not fetch history from ComfyUI")
+
+    if prompt_id not in history:
+        raise HTTPException(status_code=404, detail="Prompt not found in history")
+
+    entry = history[prompt_id]
+    status = entry.get("status", {})
+    outputs = entry.get("outputs", {})
+
+    # Extract image filenames
+    images: list[str] = []
+    for _node_id, output in outputs.items():
+        if "images" in output:
+            for img in output["images"]:
+                images.append(img.get("filename", ""))
+
+    return {
+        "prompt_id": prompt_id,
+        "status": status.get("status_str", "unknown"),
+        "completed": status.get("completed", False),
+        "outputs": outputs,
+        "images": images,
+    }
+
+
+# =============================================================================
+# Generation Endpoints
+# =============================================================================
+
+
+@router.post("/generate", response_model=GenerateResponse)
+def comfyui_generate(request: GenerateRequest) -> dict[str, Any]:
+    """Generate an image using a simple text-to-image workflow.
+
+    This uses the built-in SDXL/Flux compatible workflow template.
+    For custom workflows, use the /workflow endpoint instead.
+    """
+    result = generate_image(
+        prompt=request.prompt,
+        negative_prompt=request.negative_prompt,
+        model=request.model,
+        width=request.width,
+        height=request.height,
+        steps=request.steps,
+        cfg=request.cfg,
+        seed=request.seed,
+        sampler=request.sampler,
+        scheduler=request.scheduler,
+    )
+
+    if not result:
+        raise HTTPException(status_code=502, detail="Failed to queue generation")
+
+    return {
+        "success": result.success,
+        "prompt_id": result.prompt_id,
+        "images": [str(img) for img in result.images],
+        "errors": result.node_errors,
+    }
+
+
+@router.post("/workflow", response_model=WorkflowResponse)
+def comfyui_workflow(request: WorkflowRequest) -> dict[str, Any]:
+    """Queue an arbitrary ComfyUI workflow for execution.
+
+    The workflow should be in ComfyUI API format (exported via "Save (API Format)").
+    This endpoint queues the workflow and returns immediately with the prompt_id.
+    Use /history/{prompt_id} to check the result.
+    """
+    result = queue_prompt(workflow=request.workflow)
+
+    if not result:
+        raise HTTPException(status_code=502, detail="Failed to queue workflow")
+
+    if "error" in result:
+        return {
+            "success": False,
+            "prompt_id": "",
+            "error": result.get("error"),
+            "node_errors": result.get("node_errors", {}),
+        }
+
+    return {
+        "success": True,
+        "prompt_id": result.get("prompt_id", ""),
+        "number": result.get("number"),
+    }
+
+
+def create_comfyui_api_router() -> APIRouter:
+    """Return the ComfyUI API router."""
+    return router