Client API¶

Complete reference for the main InvokeAI client interface, covering connection management, repository access, HTTP operations, and Socket.IO event handling. Key source locations include the main InvokeAIClient class, from_url() constructor, _make_request() HTTP helper, health_check() connectivity testing, and Socket.IO methods connect_socketio()/socketio_session().

InvokeAIClient¶

Main client class for interacting with an InvokeAI server instance. This class represents a connection to an InvokeAI server and provides high-level operations for workflow execution, asset management, and job tracking.

Constructor¶

def __init__(
    self,
    host: str = "localhost",
    port: int = 9090,
    api_key: str | None = None,
    timeout: float = 30.0,
    base_path: str = "/api/v1",
    use_https: bool = False,
    verify_ssl: bool = True,
    max_retries: int = 3,
    **kwargs: Any,
) -> None:

Initialize the InvokeAI client with connection parameters.

Parameters: - host (str): The hostname or IP address of the InvokeAI server - port (int): The port number of the InvokeAI server
- api_key (str, optional): API key for authentication, if required - timeout (float): Request timeout in seconds - base_path (str): Base path for API endpoints - use_https (bool): Whether to use HTTPS for connections - verify_ssl (bool): Whether to verify SSL certificates - max_retries (int): Maximum number of retry attempts for failed requests

Example:

# Direct initialization
client = InvokeAIClient(
    host="192.168.1.100",
    port=9090,
    timeout=60.0,
    use_https=True
)

`from_url()` - URL Helper Constructor¶

@classmethod
def from_url(cls, url: str, **kwargs: Any) -> InvokeAIClient:

Create an InvokeAI client from a URL. Parses the URL into host/port/base_path automatically.

Parameters: - url (str): The URL of the InvokeAI instance (e.g., "http://localhost:9090") - **kwargs: Additional keyword arguments passed to the constructor

Returns: - InvokeAIClient: A configured client instance

Examples:

# Recommended approach - automatic URL parsing
client = InvokeAIClient.from_url("http://localhost:9090")

# With custom base path
client = InvokeAIClient.from_url("https://my-invoke.ai:8080/api/v1")

# With additional parameters
client = InvokeAIClient.from_url(
    "http://localhost:9090", 
    timeout=120.0,
    api_key="my-key"
)

Source: InvokeAIClient.from_url()

Repository Properties¶

The client provides repository instances for different operations, lazily constructed and cached on first access.

`board_repo` Property¶

@property
def board_repo(self) -> BoardRepository:

Get the board repository instance for board-related operations.

Returns: - BoardRepository: Repository for managing boards and images

Example:

# Access board operations
boards = client.board_repo.list_boards()
handle = client.board_repo.get_board_handle("my-board-id")

Source: InvokeAIClient.board_repo

`workflow_repo` Property¶

@property
def workflow_repo(self) -> WorkflowRepository:

Get the workflow repository instance for workflow-related operations.

Returns: - WorkflowRepository: Repository for creating and managing workflows

Example:

# Create workflow from definition
definition = WorkflowDefinition.from_file("workflow.json")
workflow = client.workflow_repo.create_workflow(definition)

Source: InvokeAIClient.workflow_repo

`dnn_model_repo` Property¶

@property
def dnn_model_repo(self) -> DnnModelRepository:

Get the DNN model repository instance for model operations.

Returns: - DnnModelRepository: Repository for listing and accessing models

Example:

# List available models
models = client.dnn_model_repo.list_models()
sdxl_models = [m for m in models if m.base == "sdxl"]

Source: InvokeAIClient.dnn_model_repo

Core Methods¶

`_make_request()` - HTTP Request Helper¶

def _make_request(self, method: str, endpoint: str, **kwargs: Any) -> requests.Response:

Make an HTTP request to the API. This is the core method used by all repositories and handles for HTTP communication.

Parameters: - method (str): HTTP method ("GET", "POST", "PUT", "DELETE", etc.) - endpoint (str): API endpoint path (will be appended to base_url) - **kwargs: Additional arguments passed to requests (json, data, params, etc.)

Returns: - requests.Response: HTTP response object

Behavior: - Adds timeout if not provided in kwargs - Raises requests.HTTPError for HTTP error status codes
- Full URL constructed as: base_url + endpoint - Includes authentication headers if API key is configured

Source: InvokeAIClient._make_request()

`health_check()` - Connection Verification¶

def health_check(self) -> bool:

Check if the InvokeAI instance is healthy and reachable.

Returns: - bool: True if server responds successfully, False otherwise

Behavior: - Sends GET request to {base_url}/health - Returns True on successful response, False on any error - Does not raise exceptions - safe for connectivity testing

Example:

client = InvokeAIClient.from_url("http://localhost:9090")
if client.health_check():
    print("InvokeAI server is reachable")
else:
    print("Cannot connect to InvokeAI server")

Source: InvokeAIClient.health_check()

Socket.IO Operations (Async)¶

The client provides async Socket.IO functionality for real-time event monitoring during workflow execution.

`connect_socketio()` - Establish Socket.IO Connection¶

async def connect_socketio(self) -> socketio.AsyncClient:

Connect to the InvokeAI Socket.IO server for real-time events.

Returns: - socketio.AsyncClient: Connected Socket.IO client instance

Connection Details: - Connects to ws(s)://{host}:{port} with path /ws/socket.io - Uses same host/port as HTTP client - Enables real-time workflow progress and status updates

Source: InvokeAIClient.connect_socketio()

`disconnect_socketio()` - Close Socket.IO Connection¶

async def disconnect_socketio(self) -> None:

Disconnect from the Socket.IO server and clean up the connection.

`socketio_session()` - Managed Socket.IO Context¶

@asynccontextmanager
async def socketio_session(self) -> AsyncGenerator[socketio.AsyncClient, None]:

Context manager for Socket.IO connections. Automatically connects and disconnects.

Yields: - socketio.AsyncClient: Connected Socket.IO client for event handling

Example:

async def monitor_workflow():
    client = InvokeAIClient.from_url("http://localhost:9090")

    async with client.socketio_session() as sio:
        # Subscribe to queue events
        await sio.emit("subscribe_queue", {"queue_id": "default"})

        @sio.on("queue_item_status_changed")
        async def on_status_change(data):
            print(f"Status: {data.get('status')}")

        # Your async workflow operations...
        await asyncio.sleep(10)

        # Unsubscribe before context exits
        await sio.emit("unsubscribe_queue", {"queue_id": "default"})

Context Management¶

`close()` - Resource Cleanup¶

def close(self) -> None:

Close the client connection and clean up all resources.

Behavior: - Closes HTTP session and connection pools - Attempts to disconnect Socket.IO if connected - Should be called when done with the client

Context Manager Support¶

def __enter__(self) -> InvokeAIClient:
def __exit__(self, exc_type, exc_val, exc_tb) -> None:

Enables usage as a context manager for automatic resource cleanup.

Example:

with InvokeAIClient.from_url("http://localhost:9090") as client:
    boards = client.board_repo.list_boards()
    # client.close() called automatically on exit

Source: InvokeAIClient.close()

Note on non-existent methods - Older docs mentioned connect(), disconnect(), ping(), get_server_info(), connect_websocket(), subscribe_to_queue(). These are not present. Use: - health_check() for a quick probe - _make_request("GET", "/app/version") if you need version info - connect_socketio()/socketio_session() for WebSocket(SIO) usage

Usage Examples¶

Basic connection and quick probe

from invokeai_py_client import InvokeAIClient

# 1) Using URL helper (recommended)
client = InvokeAIClient.from_url("http://localhost:9090")

# 2) Or explicit host/port (base_path defaults to /api/v1)
client = InvokeAIClient(host="localhost", port=9090)

# Quick probe
if client.health_check():
    print("InvokeAI reachable")

# Read version (raw request)
resp = client._make_request("GET", "/app/version")
print(resp.json())

Create a workflow and run (blocking)

from invokeai_py_client.workflow import WorkflowDefinition

client = InvokeAIClient.from_url("http://localhost:9090")
wf = client.workflow_repo.create_workflow(
    WorkflowDefinition.from_file("data/workflows/sdxl-text-to-image.json")
)

# Discover inputs and set values (indices are the stable handle)
for inp in wf.list_inputs():
    print(f"[{inp.input_index:02d}] {inp.label or inp.field_name}")

field = wf.get_input_value(0)
if hasattr(field, "value"):
    field.value = "A cinematic sunset over snowy mountains"

submission = wf.submit_sync()
queue_item = wf.wait_for_completion_sync(timeout=180)
for m in wf.map_outputs_to_images(queue_item):
    print(m["node_id"], m.get("image_names"))

- Mirrors the example: sdxl-text-to-image.py

Boards and images

# List boards
boards = client.board_repo.list_boards(include_uncategorized=True)
for b in boards:
    print(getattr(b, "board_id", "?"), getattr(b, "board_name", ""))

# Get a handle and download an image (by name)
bh = client.board_repo.get_board_handle("none")  # uncategorized
names = bh.list_images(limit=10)
if names:
    data = bh.download_image(names[0], full_resolution=True)
    with open(names[0], "wb") as f:
        f.write(data)

- See handle methods: BoardHandle class

Socket.IO events (async)

import asyncio

async def main():
    client = InvokeAIClient.from_url("http://localhost:9090")
    async with client.socketio_session() as sio:
        await sio.emit("subscribe_queue", {"queue_id": "default"})

        @sio.on("queue_item_status_changed")
        async def on_status(evt):
            print("Status:", evt.get("status"))

        # Submit a workflow (sync submit), then wait using your own logic
        # The server will emit events while processing

        await asyncio.sleep(5)
        await sio.emit("unsubscribe_queue", {"queue_id": "default"})

asyncio.run(main())

DNN models (read-only repo)

# List models (fresh API call)
models = client.dnn_model_repo.list_models()
print(f"Total models: {len(models)}")

# Use sync_dnn_model during workflow prep to normalize identifiers
wf.sync_dnn_model(by_name=True, by_base=True)

Best practices¶

Use from_url() to avoid manual host/port parsing.
Treat indices as the public, stable input API.
Keep Socket.IO connections long-lived if you need frequent events; reuse via socketio_session().
For uploads, prefer BoardHandle.upload_image or upload_image_data; omit board_id for uncategorized (“none” is a sentinel for read ops).
Handle HTTP errors raised by _make_request() and use repository methods wherever available.

Cross-references - Workflows: docs/api-reference/workflow.md - Boards: docs/api-reference/boards.md - Fields: docs/api-reference/fields.md - Examples: docs/examples/index.md

Client API¶

InvokeAIClient¶

Constructor¶

from_url() - URL Helper Constructor¶

Repository Properties¶

board_repo Property¶

workflow_repo Property¶

dnn_model_repo Property¶

Core Methods¶

_make_request() - HTTP Request Helper¶

health_check() - Connection Verification¶

Socket.IO Operations (Async)¶

connect_socketio() - Establish Socket.IO Connection¶

disconnect_socketio() - Close Socket.IO Connection¶

socketio_session() - Managed Socket.IO Context¶

Context Management¶

close() - Resource Cleanup¶

Context Manager Support¶

Usage Examples¶

Best practices¶

`from_url()` - URL Helper Constructor¶

`board_repo` Property¶

`workflow_repo` Property¶

`dnn_model_repo` Property¶

`_make_request()` - HTTP Request Helper¶

`health_check()` - Connection Verification¶

`connect_socketio()` - Establish Socket.IO Connection¶

`disconnect_socketio()` - Close Socket.IO Connection¶

`socketio_session()` - Managed Socket.IO Context¶

`close()` - Resource Cleanup¶