Skip to content

UpKeep Connector

The UpKeepConnector integrates Machina with UpKeep, a leading cloud-based CMMS platform, via its REST API v2.

Prerequisites

  • UpKeep account with API access enabled
  • API token (generated from Account Settings → API Tokens in the UpKeep web UI)

Installation

pip install machina-ai[cmms-rest]

Configuration

from machina.connectors import UpKeep

connector = UpKeep(api_key="your-upkeep-api-token")
await connector.connect()

connectors:
  cmms:
    type: upkeep
    api_key: ${UPKEEP_API_KEY}

Capabilities

Capability Description
read_assets Read all assets (/api/v2/assets)
read_work_orders Read work orders — filter by asset_id and/or status (accepts WorkOrderStatus enum or raw UpKeep string)
create_work_order Create a new work order
update_work_order Update status, assignee, or description via PATCH
read_spare_parts Read parts inventory (/api/v2/parts) — prefers partNumber / barcode as SKU
read_maintenance_plans Read preventive-maintenance schedules (/api/v2/preventive-maintenance)

Convenience methods

These methods are available but are not declared as agent-discoverable capabilities:

Method Description
get_work_order(id) Fetch a single work order by ID
close_work_order(id) Transition to CLOSED (maps to UpKeep complete) via update_work_order
cancel_work_order(id) Transition to CANCELLED (maps to UpKeep on hold) via update_work_order

Usage Examples

Read assets

assets = await connector.read_assets()
for asset in assets:
    print(f"{asset.id}: {asset.name}")

Read work orders with Machina enum filter

from machina.domain.work_order import WorkOrderStatus

wos = await connector.read_work_orders(
    asset_id="asset-123",
    status=WorkOrderStatus.IN_PROGRESS,  # auto-mapped to "in progress"
)

Get a single work order

wo = await connector.get_work_order("wo-123")

Create a work order

from datetime import datetime, timezone
from machina.domain import WorkOrder, WorkOrderType, Priority

wo = WorkOrder(
    id="",
    type=WorkOrderType.CORRECTIVE,
    priority=Priority.HIGH,
    asset_id="asset-123",
    description="Replace worn bearing",
    created_at=datetime.now(tz=timezone.utc),
    updated_at=datetime.now(tz=timezone.utc),
)
created = await connector.create_work_order(wo)
print(f"Created: {created.id}")

Update / close a work order

from machina.domain.work_order import WorkOrderStatus

updated = await connector.update_work_order(
    "wo-123",
    status=WorkOrderStatus.COMPLETED,
    assigned_to="tech-user-id",
)
await connector.close_work_order("wo-123")

Entity Mapping

UpKeep Field Machina Field
id Asset.id
name Asset.name
category Asset.type (mapped to closest AssetType)
location Asset.location
make Asset.manufacturer
model Asset.model
serialNumber Asset.serial_number
id (work order) WorkOrder.id
title WorkOrder.description
priority (0-3) WorkOrder.priority (0→Low, 1→Medium, 2→High, 3→Emergency)
status WorkOrder.status (open→Created, in progress→InProgress, on hold→Assigned, complete→Completed)
partNumber / barcode / id SparePart.sku (prefers physical identifier, falls back to record ID)
name (part) SparePart.name
quantity SparePart.stock_quantity
frequencyDays MaintenancePlan.interval.days

Resilience

All HTTP calls route through a shared retry helper with exponential backoff. See SAP PM Connector — Resilience for details.

Known Limitations

  • Asset criticality: UpKeep does not expose a native criticality field. All assets default to Criticality.C.
  • Work order types: UpKeep uses category ("preventive" / "reactive"). The connector maps these to PREVENTIVE and CORRECTIVE respectively. Predictive and improvement types are not natively supported by UpKeep; for custom categories, subclass the connector.
  • Spare part filtering by asset: The connector fetches all parts and filters client-side, since UpKeep's parts API does not support asset-level filtering. Filtering by sku is supported in-memory.
  • Failure data: UpKeep has no standard failure-mode fields. Failure-related data may be available in WorkOrder.metadata depending on your UpKeep configuration.

API Reference

UpKeepConnector 

UpKeepConnector(*, url: str = '', api_key: str = '')

Connector for UpKeep CMMS.

Provides integration with UpKeep's REST API v2 for reading and creating maintenance data.

Parameters:

Name Type Description Default
url str

UpKeep API base URL. Defaults to the production endpoint.

 ''
api_key str

API token from UpKeep (passed as Session-Token header).

 ''
Example 
from machina.connectors import UpKeep

connector = UpKeep(api_key="your-upkeep-api-token")
await connector.connect()
assets = await connector.read_assets()

connect async 

connect() -> None

Verify credentials against the UpKeep API.

Raises:

Type Description
ConnectorAuthError

If the API key is missing or invalid.
ConnectorError

If the API is unreachable.

disconnect async 

disconnect() -> None

Close the connector (no persistent connections to clean up).

health_check async 

health_check() -> ConnectorHealth

Return current health status.

read_assets async 

read_assets() -> list[Asset]

Return all assets from UpKeep.

get_asset async 

get_asset(asset_id: str) -> Asset | None

Look up a single asset by ID.

read_work_orders async 

read_work_orders(*, asset_id: str = '', status: WorkOrderStatus | str = '') -> list[WorkOrder]

Read work orders, optionally filtered by asset or status.

Parameters:

Name Type Description Default
asset_id str

Filter by UpKeep asset ID.

 ''
status WorkOrderStatus | str

Filter by status — accepts a :class:WorkOrderStatus enum (reverse-mapped to UpKeep's string) or a raw UpKeep status string for backward compatibility.

 ''

get_work_order async 

get_work_order(work_order_id: str) -> WorkOrder | None

Look up a single work order by ID.

create_work_order async 

create_work_order(work_order: WorkOrder) -> WorkOrder

Create a new work order in UpKeep.

Parameters:

Name Type Description Default
work_order WorkOrder

Machina :class:WorkOrder to create.

 required

Returns:

Type Description
WorkOrder

The created work order with the server-assigned ID.

update_work_order async 

update_work_order(work_order_id: str, *, status: WorkOrderStatus | None = None, assigned_to: str | None = None, description: str | None = None) -> WorkOrder

Update an existing work order in UpKeep via PATCH.

Only non-None fields are included in the PATCH payload.

Parameters:

Name Type Description Default
work_order_id str

UpKeep work order ID.

 required
status WorkOrderStatus | None

New status (reverse-mapped to UpKeep string).

 None
assigned_to str | None

New assignedToId.

 None
description str | None

New title.

 None

Returns:

Type Description
WorkOrder

The updated work order.

close_work_order async 

close_work_order(work_order_id: str) -> WorkOrder

Transition a work order to CLOSED (maps to 'complete' in UpKeep).

cancel_work_order async 

cancel_work_order(work_order_id: str) -> WorkOrder

Transition a work order to CANCELLED (maps to 'on hold' in UpKeep).

read_spare_parts async 

read_spare_parts(*, sku: str = '') -> list[SparePart]

Read spare parts (UpKeep calls them parts).

Parameters:

Name Type Description Default
sku str

Optional SKU / part number to filter the result in-memory after fetching. Matches the parsed :attr:SparePart.sku, which prefers the physical part identifier.

 ''
Note

UpKeep's /api/v2/parts endpoint does not expose an asset-compatibility relation, so filtering by asset is not supported here. Use work-order line items to discover parts associated with a specific asset.

read_maintenance_plans async 

read_maintenance_plans() -> list[MaintenancePlan]

Read preventive-maintenance schedules from UpKeep.

read_maintenance_history async 

read_maintenance_history(asset_id: str) -> list[WorkOrder]

Return completed work orders for an asset.