Workflow Builders

Build workflows programmatically with helper functions that generate valid workflow configurations.

build_geofence_webhook_workflow

Creates a workflow that triggers on geofence events and calls a webhook.

from spatialflow import build_geofence_webhook_workflow

workflow_in = build_geofence_webhook_workflow(
    name="Entry Alert",
    geofence_ids=["uuid-1", "uuid-2"],
    webhook_url="https://webhook.site/your-id",
    trigger_type="geofence_enter",  # or "geofence_exit", "geofence_dwell"
    description="Alert when devices enter zones",
    webhook_method="POST",
    webhook_headers={"Authorization": "Bearer xxx"},
)

# Create the workflow
workflow = await client.workflows.create(workflow_in)

Parameters

Parameter	Type	Default	Description
name	str	required	Workflow name
geofence_ids	List[str]	required	List of geofence UUIDs to monitor
webhook_url	str	required	URL to call when triggered
trigger_type	TriggerType	"geofence_enter"	Event type to trigger on
description	str	None	Optional description
webhook_method	str	"POST"	HTTP method
webhook_headers	dict	{"Content-Type": "application/json"}	HTTP headers

Trigger Types

• "geofence_enter" - Device enters a geofence
• "geofence_exit" - Device exits a geofence
• "geofence_dwell" - Device dwells in a geofence

build_geofence_integration_workflow

Creates a workflow using a pre-configured integration.

from spatialflow import build_geofence_integration_workflow

# First, create an integration
integration = await client.integrations.create_webhook(
    name="My Webhook",
    url="https://example.com/hook",
)

# Then use it in a workflow
workflow_in = build_geofence_integration_workflow(
    name="Entry Alert",
    geofence_ids=["uuid-1"],
    integration_id=integration.id,
    trigger_type="geofence_enter",
)

workflow = await client.workflows.create(workflow_in)

Parameters

Parameter	Type	Default	Description
name	str	required	Workflow name
geofence_ids	List[str]	required	List of geofence UUIDs to monitor
integration_id	str	required	ID of the integration to use
trigger_type	TriggerType	"geofence_enter"	Event type to trigger on
description	str	None	Optional description

End-to-End Example

Complete workflow: create geofences, build a workflow, and trigger it with location updates.

import asyncio
import uuid
from spatialflow import SpatialFlow, models, build_geofence_webhook_workflow

async def main():
    async with SpatialFlow(api_key="sf_xxx") as client:
        # 1. Create geofences with grouping
        group_id = str(uuid.uuid4())
        geofence_ids = []

        for i in range(3):
            geofence = await client.geofences.create(
                models.CreateGeofenceRequest(
                    name=f"Zone {i+1}",
                    group_id=group_id,
                    group_name="test-zones",
                    geometry={
                        "type": "Polygon",
                        "coordinates": [[
                            [-122.4 + i*0.01, 37.8],
                            [-122.4 + i*0.01, 37.79],
                            [-122.39 + i*0.01, 37.79],
                            [-122.39 + i*0.01, 37.8],
                            [-122.4 + i*0.01, 37.8],
                        ]]
                    }
                )
            )
            geofence_ids.append(geofence.id)

        # 2. Create workflow with webhook action
        webhook_url = "https://webhook.site/your-unique-id"
        workflow_in = build_geofence_webhook_workflow(
            name="Entry Alert",
            geofence_ids=geofence_ids,
            webhook_url=webhook_url,
            trigger_type="geofence_enter",
        )
        workflow = await client.workflows.create(workflow_in)

        # 3. Enable the workflow
        await client.workflows.toggle(workflow_id=workflow.id)

        # 4. Send location outside geofences
        await client.locations.ingest(
            device_id="test-device-001",
            lat=37.75,
            lon=-122.45,
        )

        # 5. Send location inside geofence (triggers webhook)
        await client.locations.ingest(
            device_id="test-device-001",
            lat=37.795,
            lon=-122.395,
        )

        print("Check webhook.site for the triggered event!")

        # Cleanup
        await client.workflows.delete(workflow_id=workflow.id)
        for gid in geofence_ids:
            await client.geofences.delete(geofence_id=gid)

asyncio.run(main())

Workflow Structure

The builders generate a workflow with nodes and edges compatible with the SpatialFlow workflow engine:

# Generated structure
{
    "name": "Entry Alert",
    "description": None,
    "nodes": [
        {
            "id": "trigger-1",
            "type": "trigger",
            "position": {"x": 100, "y": 100},
            "data": {
                "label": "Geofence Enter",
                "triggerType": "geofence_enter",
                "config": {
                    "type": "geofence_enter",
                    "geofence_ids": ["uuid-1", "uuid-2"]
                }
            }
        },
        {
            "id": "action-1",
            "type": "action",
            "position": {"x": 400, "y": 100},
            "data": {
                "label": "Webhook",
                "actionType": "webhook",
                "config": {
                    "url": "https://example.com/hook",
                    "method": "POST",
                    "headers": {"Content-Type": "application/json"}
                }
            }
        }
    ],
    "edges": [
        {"id": "edge-1", "source": "trigger-1", "target": "action-1"}
    ]
}