Skip to main content

Resources

Each resource provides clean CRUD methods with automatic timeout and exception handling.

Each section below says which credential works. An API key needs the scope for the operation, and a missing scope raises PermissionError. Operations marked access-token-only reject API keys, so create the client with access_token= for them.

Geofences​

An API key needs geofences:read to list, get, list by group, list groups and a group's geofences, and read an upload job; geofences:create to create, create with a group, upload, and bulk create; geofences:write to update, assign to a group, and test a group point; and geofences:delete to delete. Create, update, assign to a group, delete, upload, and bulk create also need an owner or manager, which for an API key is the role of the user who owns it. Bulk create also needs the bulk import feature on the workspace's plan.

# List geofences
response = await client.geofences.list(limit=100, offset=0)
for geofence in response.geofences:
    print(geofence.name)

# Create a geofence
geofence = await client.geofences.create(
    models.CreateGeofenceRequest(
        name="My Region",
        geometry={"type": "Polygon", "coordinates": [...]},
        group_name="my-group",  # Optional grouping
    )
)

# Get a geofence
geofence = await client.geofences.get(geofence_id="...")

# Update a geofence
geofence = await client.geofences.update(
    geofence_id="...",
    request=models.UpdateGeofenceRequest(name="New Name")
)

# Delete a geofence
await client.geofences.delete(geofence_id="...")

# Create with group (convenience method)
geofence = await client.geofences.create_with_group(
    name="Zone 1",
    geometry={...},
    group_name="delivery-zones",
)

# List groups, then the active geofences in one group (not paginated)
groups = await client.geofences.list_groups()
group_geofences = await client.geofences.list_group_geofences(group_id="...")

# Test which geofences in a group contain a point (needs geofences:write; rate limited)
result = await client.geofences.test_group_point(
    group_id="...",
    request=models.TestPointRequest(lat=37.7749, lng=-122.4194),
)

# Move an existing geofence into a group (updates group_name; there is no separate assign endpoint)
geofence = await client.geofences.assign_to_group(
    geofence_id="...", group_name="delivery-zones"
)

Workflows​

An API key needs workflows:read to list, get, export, and read performance, statistics, bottlenecks, step performance, versions, and the retry policy; workflows:create to create, duplicate, and import; workflows:write to update, toggle, activate, execute, test, restore a version, and update the retry policy; and workflows:delete to delete. These calls also need an owner or manager, including for an API key. Listing executions and getting one need workflows:read and any workspace member, but field workers see only the executions of their own device, with step inputs, outputs, and error messages removed.

# List workflows
response = await client.workflows.list()

# Create a workflow
workflow = await client.workflows.create(
    models.WorkflowIn(
        name="Entry Alert",
        nodes=[...],
        edges=[...],
    )
)

# Get a workflow
workflow = await client.workflows.get(workflow_id="...")

# Update a workflow
workflow = await client.workflows.update(
    workflow_id="...",
    request=models.WorkflowIn(name="Updated", nodes=[...], edges=[...])
)

# Toggle active/paused
workflow = await client.workflows.toggle(workflow_id="...")

# Delete a workflow
await client.workflows.delete(workflow_id="...")

# Activate a draft workflow
workflow = await client.workflows.activate(workflow_id="...")

Workflow Execution​

# Execute a workflow
result = await client.workflows.execute(
    workflow_id="...",
    request=None,  # Optional payload
)

# Test a workflow with sample data
result = await client.workflows.test(
    workflow_id="...",
    request=None,  # Optional payload
)

# List executions for a workflow
executions = await client.workflows.list_executions(
    workflow_id="...",
    limit=50,
    offset=0,
)

# Get execution details
execution = await client.workflows.get_execution(
    workflow_id="...",
    execution_id="...",
)

Workflow Monitoring​

# Get performance metrics
performance = await client.workflows.get_performance(workflow_id="...")

# Get statistics
stats = await client.workflows.get_statistics(workflow_id="...")

# Get retry policy
policy = await client.workflows.get_retry_policy(workflow_id="...")

# Update retry policy
policy = await client.workflows.update_retry_policy(
    workflow_id="...",
    request=models.WorkflowRetryPolicyUpdateSchema(
        workflow_id="...",
        default_retry_policy="3x_exponential",
        step_overrides=[],
    )
)

# Get bottleneck analysis
bottlenecks = await client.workflows.get_bottlenecks(workflow_id="...")

# Get step-level performance
step_perf = await client.workflows.get_step_performance(workflow_id="...")

# List all executions across workspace
all_executions = await client.workflows.list_all_executions(limit=100)

Workflow Versioning​

# List versions
versions = await client.workflows.list_versions(workflow_id="...")

# Restore a previous version
workflow = await client.workflows.restore_version(
    workflow_id="...",
    version_number=3,
)

# Duplicate a workflow
duplicate = await client.workflows.duplicate(
    workflow_id="...",
    request={"name": "Copy of Workflow"},  # Optional dict payload
)

# Export workflow as JSON
export_data = await client.workflows.export_workflow(workflow_id="...")

# Import workflow from JSON
workflow = await client.workflows.import_workflow(
    request=models.WorkflowImportSchema(
        version=export_data["version"],
        workflow=export_data["workflow"],
    )
)
tip

Use the workflow builders to create workflow configurations programmatically.

Webhooks​

Webhook calls accept only an access token, not an API key, and need an owner or manager. get_metrics also needs the platform admin role.

# List webhooks
response = await client.webhooks.list()

# Create a webhook. webhook.secret is its signing secret; no other call returns it.
webhook = await client.webhooks.create(
    models.CreateWebhookRequest(
        name="My Webhook",
        url="https://example.com/webhook",
        events=["geofence.enter", "geofence.exit"],
    )
)

# Get a webhook
webhook = await client.webhooks.get(webhook_id="...")

# Update a webhook
webhook = await client.webhooks.update(
    webhook_id="...",
    request=models.UpdateWebhookRequest(url="https://new-url.com/webhook")
)

# Rotate the signing secret. result.secret is the new one, returned only here;
# deliveries switch to it immediately.
result = await client.webhooks.rotate_secret(webhook_id="...")

# Delete a webhook
await client.webhooks.delete(webhook_id="...")

# Test a webhook
result = await client.webhooks.test(
    webhook_id="...",
    request=models.TestWebhookRequest(...)  # Optional
)

Webhook Delivery Tracking​

# List deliveries for a webhook
deliveries = await client.webhooks.list_deliveries(
    webhook_id="...",
    limit=50,
    offset=0,
)

# Get delivery details
delivery = await client.webhooks.get_delivery(
    webhook_id="...",
    delivery_id="...",
)

# Retry a failed delivery
result = await client.webhooks.retry_delivery(
    webhook_id="...",
    delivery_id="...",
)

Webhook Monitoring​

# Get webhook metrics
metrics = await client.webhooks.get_metrics(webhook_id="...")

# Get success/failure timeline
timeline = await client.webhooks.get_success_timeline(webhook_id="...")

Dead Letter Queue (DLQ)​

# List DLQ entries
dlq_entries = await client.webhooks.list_dlq_entries(limit=50)

# Retry from DLQ
result = await client.webhooks.retry_dlq(dlq_entry_id="...")

Devices​

An API key needs devices:read to list, get, and read sessions, session locations, and session notes; devices:create to create (creating with a device_id that's already registered in the workspace re-registers that device and also needs devices:read and devices:write); devices:write and devices:read together to update and to start, pause, resume, or end a shift; devices:write to update a location or add a session note; and devices:delete to delete. Owners and managers can get, list, update the metadata of, delete, and read the events, sessions, and session locations of any workspace device, and field workers only their own. Listing session notes follows the same rule, but adding one needs an owner or manager even on your own device. update_location and the shift calls act only on a device assigned to the caller, including for managers.

# List devices (the wrapper's list() sends limit and offset, which this endpoint rejects,
# so call the generated method)
response = await client.raw.devices.apps_devices_api_list_devices()

# Create a device
device = await client.devices.create(
    models.DeviceIn(
        name="Truck 001",
        device_id="truck-001",
    )
)
device_uuid = device.id  # Use device UUID for route params

# Get a device
device = await client.devices.get(device_id=device_uuid)

# Update a device
device = await client.devices.update(
    device_id=device_uuid,
    request=models.DeviceIn(name="Updated Name")
)

# Update device location
result = await client.devices.update_location(
    device_id=device_uuid,
    request=models.LocationUpdateIn(
        latitude=37.7749,
        longitude=-122.4194,
    )
)

# Delete a device
await client.devices.delete(device_id=device_uuid)

Shifts​

Shift calls act only on the caller's own device. Calling them for another user's device raises PermissionError, as does starting or resuming a shift before the member has accepted the tracking disclosure. A transition the current shift state does not allow (starting an active shift, pausing one that is not active, ending when there is no shift) raises ValidationError. Each call returns ShiftActionOut.

await client.devices.start_shift(device_id=device_uuid)
await client.devices.pause_shift(device_id=device_uuid)
await client.devices.resume_shift(device_id=device_uuid)
await client.devices.end_shift(device_id=device_uuid)

Sessions​

# List completed sessions (limit 1-100 and offset paginate; started_after is inclusive
# and started_before exclusive, both on start time). With include_open=True the
# current shift comes back separately as open_session; it is never in `sessions`
# or counted in total_count.
page = await client.devices.list_sessions(
    device_id=device_uuid, limit=20, offset=0, include_open=True
)
active_session = page.open_session  # None when no shift is active or paused

# The examples below use a completed session. A device that has never ended a
# shift has none.
if not page.sessions:
    raise LookupError("This device has no completed sessions yet")
session_id = page.sessions[0].id

# Get one session
session = await client.devices.get_session(device_id=device_uuid, session_id=session_id)

# Location history for a completed session, one page at a time (limit 1-10000).
# Pass the first page's snapshot_at to later pages so live appends cannot shift offsets.
first = await client.devices.get_session_locations(
    device_id=device_uuid, session_id=session_id, limit=1000
)
second = await client.devices.get_session_locations(
    device_id=device_uuid,
    session_id=session_id,
    limit=1000,
    offset=1000,
    snapshot_at=first.snapshot_at,
)

# Or get one simplified track (max_points 2-10000, a target rather than a cap).
# Pagination is ignored in this mode, and an open session rejects it with ValidationError.
track = await client.devices.get_session_locations(
    device_id=device_uuid, session_id=session_id, max_points=500
)

# Session notes
notes = await client.devices.list_session_notes(device_id=device_uuid, session_id=session_id)
note = await client.devices.add_session_note(
    device_id=device_uuid,
    session_id=session_id,
    request=models.SessionNoteIn(body="Gate was locked"),
)

Locations (Public Ingest API)​

Both an access token and an API key work. Ingest and batch ingest need devices:write, and stats need devices:read.

Location ingest is asynchronous: the API validates the payload and returns 202 Accepted, then processes the point in the background. The device_id must belong to a device registered in your workspace. An unknown device_id is not rejected synchronously (the call still returns 202); the point is dropped during processing, so register the device first.

# Ingest single location
result = await client.locations.ingest(
    device_id="device-123",
    lat=37.7749,
    lon=-122.4194,
    accuracy=10.0,      # Optional: GPS accuracy in meters
    speed=5.5,          # Optional: Speed in m/s
    heading=90.0,       # Optional: Heading in degrees
    altitude=100.0,     # Optional: Altitude in meters
    metadata={"key": "value"},  # Optional
)

# Ingest batch (up to 5000 points)
result = await client.locations.ingest_batch(
    locations=[
        {"device_id": "dev-1", "lat": 37.77, "lon": -122.41, "ts": "2024-01-15T10:00:00Z"},
        {"device_id": "dev-1", "lat": 37.78, "lon": -122.42, "ts": "2024-01-15T10:01:00Z"},
    ],
    idempotency_key="batch-001",  # Optional: For deduplication
)

# Get ingestion stats
stats = await client.locations.get_stats()

Integrations​

Integrations are reusable service connections (webhooks, Slack, SMS, etc.) for workflow actions.

Listing accepts an access token or an API key with integrations:read. Get, create, update, delete, and test accept only an access token and need an owner or manager.

# List integrations (call the generated method; the wrapper's list() sends limit and offset)
response = await client.raw.integrations.apps_integrations_api_list_integrations()

# Create a generic integration
integration = await client.integrations.create(
    models.CreateIntegrationSchema(
        name="My Slack",
        type="slack",
        config={"webhook_url": "https://hooks.slack.com/..."},
    )
)

# Create webhook integration (convenience method)
integration = await client.integrations.create_webhook(
    name="My Webhook",
    url="https://example.com/hook",
    method="POST",
    headers={"Authorization": "Bearer xxx"},
)

# Get an integration
integration = await client.integrations.get(integration_id="...")

# Update an integration
integration = await client.integrations.update(
    integration_id="...",
    request=models.UpdateIntegrationSchema(name="Updated Name")
)

# Test an integration
result = await client.integrations.test(integration_id="...")

# Delete an integration
await client.integrations.delete(integration_id="...")

Workspaces​

Workspaces are the top-level organizational unit containing all your resources.

Workspace calls accept only an access token, not an API key. Any member can read the workspace, its usage, and its members, though field workers see device tracking state only for themselves. Updating settings, changing roles, removing members, and managing invitations need an owner or manager.

# Get current workspace
workspace = await client.workspaces.get()
print(f"Workspace: {workspace.name} ({workspace.slug})")

# Update workspace settings
workspace = await client.workspaces.update(
    request=models.WorkspaceIn(name="Updated Workspace Name")
)

# Get usage metrics for current billing period
usage = await client.workspaces.get_usage()
print(f"Location events: {usage.location_events}")
print(f"Action deliveries: {usage.action_deliveries}")
print(f"Event units: {usage.event_units}")
print(f"Tier: {usage.tier} (limit: {usage.tier_limit})")

Members​

Any workspace member can list members. Changing a role or removing a member needs an owner or manager, otherwise the call raises PermissionError; an unknown user raises NotFoundError. Roles are owner, manager and field_worker, and an unknown role raises ValidationError. Managers cannot promote anyone to owner or change an owner or another manager. The last owner cannot be demoted. Members are added through invitations, not directly.

members = await client.workspaces.list_members()

await client.workspaces.update_member_role(
    user_id="...", request=models.UpdateMemberRoleIn(role="manager")
)

await client.workspaces.remove_member(user_id="...")

Invitations​

from datetime import datetime, timedelta, timezone

# Managers cannot invite owners.
invitation = await client.workspaces.create_invitation(
    models.CreateInvitationIn(email="new.member@example.com", role="field_worker")
)

# Only owners and managers can list or manage invitations. A page holds at most 100
# pending invitations, newest first. To walk them all, pass each response's
# next_cursor as `before` instead of using offset.
page = await client.workspaces.list_invitations(limit=50)

await client.workspaces.resend_invitation(invite_id="...")
# Reissue up to 200 invitations by ID; omit invitation_ids to reissue the newest pending page
await client.workspaces.resend_missing_invitations(models.BatchResendIn(invitation_ids=["..."]))
await client.workspaces.extend_invitation(
    invite_id="...",
    # expires_at must be in the future and at most 90 days from now
    request=models.ExtendInvitationIn(
        expires_at=datetime.now(timezone.utc) + timedelta(days=30)
    ),
)
await client.workspaces.cancel_invitation(invite_id="...")

Storage​

With an API key, creating a presigned URL and completing an upload need storage:write, listing files and getting a download URL need storage:read, and deleting needs storage:delete. Listing image files and downloading device files also need devices:read. See Authentication.

# Create presigned URL for upload
result = await client.storage.create_presigned_url(
    models.PresignedUrlRequest(
        file_type="geofences",
        filename="boundaries.geojson",
        file_size=1024,
    )
)

# List files of one type (call the generated method; the wrapper's list_files() sends limit and offset)
response = await client.raw.storage.apps_storage_api_list_files(file_type="geofences")

# Get download URL
result = await client.storage.get_download_url(file_id="...")

Deleting an uploaded file doesn't work yet. The delete endpoint looks the file up by its name directly under the workspace, but uploads are stored under a per-file folder with a timestamped name, so the call can't find them.

Account​

The account resource provides access to user profile, API key management, dashboard metrics, and notifications. Account calls accept only an access token, not an API key. That includes creating and managing API keys.

User Profile​

# Get current user's profile
profile = await client.account.get_profile()
print(f"User: {profile.email}")

# Update profile
profile = await client.account.update_profile(
    models.UpdateProfileRequest(name="New Name")
)

API Key Management​

# List all API keys
keys = await client.account.list_api_keys()
for key in keys:
    print(f"{key.name}: {key.key_prefix}...")

# Get a specific API key
key = await client.account.get_api_key(key_id="...")

# Create a new API key
new_key = await client.account.create_api_key(
    models.ApiKeyCreateRequest(name="Production Key")
)
print(f"New key (save this!): {new_key.api_key.api_key}")

# Update an API key
key = await client.account.update_api_key(
    key_id="...",
    request=models.ApiKeyUpdateRequest(name="Renamed Key")
)

# Rotate an API key (generates new key value)
rotated = await client.account.rotate_api_key(key_id="...")
print(f"New key value: {rotated.api_key.api_key}")

# Delete an API key
await client.account.delete_api_key(key_id="...")

Dashboard Metrics​

# Get dashboard metrics
metrics = await client.account.get_dashboard_metrics()
print(f"Active workflows: {metrics.active_workflows}")
print(f"Total events: {metrics.events_total}")

Notifications​

# Get notifications
notifications = await client.account.get_notifications()
for notif in notifications:
    print(f"[{notif['read']}] {notif['message']}")

# Mark a notification as read
await client.account.mark_notification_read(notification_id="...")

# Mark all notifications as read
await client.account.mark_all_notifications_read()
note

Notification methods return List[Dict[str, Any]] as there is no specific response model for notifications in the generated API.

Raw API Access​

For endpoints not covered by the clean API, use client.raw:

# Access raw generated APIs
response = await client.raw.geofences.apps_geofences_api_list_geofences()
response = await client.raw.workflows.apps_workflows_api_toggle_workflow(workflow_id="...")
response = await client.raw.locations.apps_public_locations_api_get_ingest_stats()

Raw API methods follow the pattern: apps_{resource}_api_{operation}