Tasks

Create a task

Create tasks using a custom task definition

This guide shows you how to create tasks in Lattice using custom task definitions that you’ve published to the Lattice Schema Registry (LSR). Custom task definitions allow you to extend Lattice’s tasking capabilities with your own specialized workflows.

When working with custom tasks:

  1. First, you retrieve your JSON Schema definitions from the Schema Registry
  2. Then you validate your task data against these schema definitions
  3. Finally, you use the validated data to create tasks in Lattice using the CreateTask API

Before you begin

  • You need to have custom schemas defined and published to the Lattice Schema Registry.
  • To create tasks, set up your Lattice environment.
  • Familiarize yourself with tasks in Lattice.

Get the JSON schemas

When you push your Protobuf definitions to the Schema Registry, they’re automatically converted to JSON Schema files. These files help validate your task data before sending it to Lattice.

1

Log in to the Schema Registry dashboard using your credentials.

2

Navigate to your organization’s repositories.

3

Select the repository that contains your custom task definitions.

4

Browse to the specific protobuf definition you want to use for your task.

5

Select the plugin dropdown and choose “JSON Schema” to view the available JSON Schema files.

6

Download the JSON Schema files using curl. For example:

$curl -H "Authorization: Bearer $BUF_TOKEN" \
>     -o myschema.jsonschema.bundle.json \
>     https://schema-registry.developer.anduril.com/plugins/org/repository/myschema/jsonschema/bundle

This command downloads the bundled JSON Schema file that includes all referenced schemas.

Example JSON Schema structure for a custom task:

1{
2  "$id": "org.repository.package.v1.schema.jsonschema.json",
3  "$schema": "https://json-schema.org/draft/2020-12/schema",
4  "properties": {
5    "pattern": {
6      "enum": [
7        "SURVEILLANCE_PATTERN_GRID",
8        "SURVEILLANCE_PATTERN_SPIRAL",
9        "SURVEILLANCE_PATTERN_RANDOM",
10        "SURVEILLANCE_PATTERN_PERIMETER"
11      ],
12      "type": "string"
13    },
14    "minAltitudeM": {
15      "$ref": "google.protobuf.FloatValue.jsonschema.json"
16    },
17    "maxAltitudeM": {
18      "$ref": "google.protobuf.FloatValue.jsonschema.json"
19    },
20    "priority": {
21      "$ref": "google.protobuf.UInt32Value.jsonschema.json"
22    },
23    "sensorModes": {
24      "items": {
25        "enum": [
26          "SENSOR_MODE_EO",
27          "SENSOR_MODE_IR",
28          "SENSOR_MODE_SAR",
29          "SENSOR_MODE_SIGINT"
30        ],
31        "type": "string"
32      },
33      "type": "array"
34    }
35  },
36  "type": "object"
37}

Create a task

To create a task using a custom task definition, do the following:

1

Prepare your task data

Define the data for your custom task that matches your schema structure. This data will be encapsulated in the specification field of the CreateTaskRequest message.

1task_data = {
2    "pattern": "SURVEILLANCE_PATTERN_GRID",
3    "minAltitudeM": 100.0,
4    "maxAltitudeM": 500.0,
5    "priority": 50,
6    "sensorModes": ["SENSOR_MODE_EO", "SENSOR_MODE_IR"]
7}
2

Validate task data using the schema

Before sending data to Lattice, validate it against the JSON Schema to ensure it matches your custom task definition. This prevents issues when submitting invalid task data to the API.

1#!/usr/bin/env python3
2import asyncio
3import json
4import os
5import sys
6from anduril import Lattice
7from anduril import (
8    GoogleProtobufAny,
9    Principal,
10    System,
11    TaskEntity
12)
13from jsonschema import validate  # type: ignore[import-untyped]
14
15# Load environment variables
16lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
17environment_token = os.getenv('ENVIRONMENT_TOKEN')
18sandboxes_token = os.getenv('SANDBOXES_TOKEN')  # Remove if not using Sandboxes
19
20if not environment_token or not lattice_endpoint:
21    print("Missing required environment variables.")
22    sys.exit(1)
23
24# Initialize Lattice client
25client = Lattice(
26    base_url=f"https://{lattice_endpoint}",
27    token=lambda: environment_token,  # type: ignore[arg-type]
28    # Remove the following header if not developing on Sandboxes
29    headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"} if sandboxes_token else None
30)
31
32def load_schema(schema_file):
33    """Load a JSON schema from file."""
34    with open(schema_file, 'r') as f:
35        return json.load(f)
36
37def validate_task_data(task_data, schema_file):
38    """Validate task data against a JSON schema."""
39    schema = load_schema(schema_file)
40
41    try:
42        validate(instance=task_data, schema=schema)
43        print("Task data validation successful")
44        return True
45    except Exception as e:
46        print(f"Validation error: {str(e)}")
47        return False
48
49async def create_task(task_data, entity_id=None):
50    """
51    Create a new task using the Lattice SDK.
52
53    Args:
54        task_data: The validated task data
55        entity_id: Optional entity ID to use as the initial entity
56    """
57    description = "Custom Task Definition"
58    specification_type = "type.googleapis.com/org.repository.package.v1.schema"
59    specification = GoogleProtobufAny(
60        type=specification_type,
61        **task_data
62    )
63    author = Principal(system=System(service_name="example-service"))
64
65    try:
66        # Set up initial entities if an entity ID was provided
67        initial_entities = None
68        if entity_id:
69            try:
70                print(f"Fetching entity with ID: {entity_id}")
71                entity = client.entities.get_entity(entity_id)
72
73                if entity:
74                    task_entity = TaskEntity(entity=entity)
75                    initial_entities = [task_entity]
76                    print(f"Using entity {entity_id} as initial entity for task")
77            except Exception as e:
78                print(f"Error fetching entity {entity_id}: {str(e)}")
79
80        # Create the task with the validated data
81        print("Creating task...")
82        response = client.tasks.create_task(
83            description=description,
84            specification=specification,
85            author=author,
86            is_executed_elsewhere=False,
87            initial_entities=initial_entities
88        )
89
90        if response.version:
91            print(f"Task created successfully with ID: {response.version.task_id}")
92        return response
93    except Exception as e:
94        print(f"Error creating task: {str(e)}")
95        return None
96
97async def main():
98    # Example task data for an aerial surveillance mission
99    task_data = {
100        "pattern": "SURVEILLANCE_PATTERN_GRID",
101        "minAltitudeM": 100.0,
102        "maxAltitudeM": 500.0,
103        "priority": 50,
104        "sensorModes": ["SENSOR_MODE_EO", "SENSOR_MODE_IR"]
105    }
106
107    # Path to the JSON schema file (downloaded from Schema Registry)
108    schema_file = 'org.repository.package.v1.schema.jsonschema.bundle.json'
109
110    # Validate the task data (skip if schema file not found)
111    if os.path.isfile(schema_file):
112        if not validate_task_data(task_data, schema_file):
113            print("Task creation aborted due to validation errors.")
114            return
115    else:
116        print("Schema file not found, skipping validation")
117
118    # Create the task with validated data
119    # You can pass an entity_id here if you want to associate the task with an entity
120    entity_id = None  # Replace with an actual entity ID if needed
121    result = await create_task(task_data, entity_id)
122    if result:
123        print(f"Task created with status: {result.status}")
124
125if __name__ == "__main__":
126    asyncio.run(main())
3

Create the task using validated data

Use the CreateTask API to submit your task to Lattice. This API expects the following key fields:

  • description: A human-readable description of the task (up to 4096 characters)
  • specification: Your validated task data wrapped in a Google Protocol Buffer Any message
  • author: Information about who or what created the task
  • initialEntities: Optional list of entities that should be associated with the task
1#!/usr/bin/env python3
2import asyncio
3import json
4import os
5import sys
6from anduril import Lattice
7from anduril import (
8    GoogleProtobufAny,
9    Principal,
10    System,
11    TaskEntity
12)
13from jsonschema import validate  # type: ignore[import-untyped]
14
15# Load environment variables
16lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
17environment_token = os.getenv('ENVIRONMENT_TOKEN')
18sandboxes_token = os.getenv('SANDBOXES_TOKEN')  # Remove if not using Sandboxes
19
20if not environment_token or not lattice_endpoint:
21    print("Missing required environment variables.")
22    sys.exit(1)
23
24# Initialize Lattice client
25client = Lattice(
26    base_url=f"https://{lattice_endpoint}",
27    token=lambda: environment_token,  # type: ignore[arg-type]
28    # Remove the following header if not developing on Sandboxes
29    headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"} if sandboxes_token else None
30)
31
32def load_schema(schema_file):
33    """Load a JSON schema from file."""
34    with open(schema_file, 'r') as f:
35        return json.load(f)
36
37def validate_task_data(task_data, schema_file):
38    """Validate task data against a JSON schema."""
39    schema = load_schema(schema_file)
40
41    try:
42        validate(instance=task_data, schema=schema)
43        print("Task data validation successful")
44        return True
45    except Exception as e:
46        print(f"Validation error: {str(e)}")
47        return False
48
49async def create_task(task_data, entity_id=None):
50    """
51    Create a new task using the Lattice SDK.
52
53    Args:
54        task_data: The validated task data
55        entity_id: Optional entity ID to use as the initial entity
56    """
57    description = "Custom Task Definition"
58    specification_type = "type.googleapis.com/org.repository.package.v1.schema"
59    specification = GoogleProtobufAny(
60        type=specification_type,
61        **task_data
62    )
63    author = Principal(system=System(service_name="example-service"))
64
65    try:
66        # Set up initial entities if an entity ID was provided
67        initial_entities = None
68        if entity_id:
69            try:
70                print(f"Fetching entity with ID: {entity_id}")
71                entity = client.entities.get_entity(entity_id)
72
73                if entity:
74                    task_entity = TaskEntity(entity=entity)
75                    initial_entities = [task_entity]
76                    print(f"Using entity {entity_id} as initial entity for task")
77            except Exception as e:
78                print(f"Error fetching entity {entity_id}: {str(e)}")
79
80        # Create the task with the validated data
81        print("Creating task...")
82        response = client.tasks.create_task(
83            description=description,
84            specification=specification,
85            author=author,
86            is_executed_elsewhere=False,
87            initial_entities=initial_entities
88        )
89
90        if response.version:
91            print(f"Task created successfully with ID: {response.version.task_id}")
92        return response
93    except Exception as e:
94        print(f"Error creating task: {str(e)}")
95        return None
96
97async def main():
98    # Example task data for an aerial surveillance mission
99    task_data = {
100        "pattern": "SURVEILLANCE_PATTERN_GRID",
101        "minAltitudeM": 100.0,
102        "maxAltitudeM": 500.0,
103        "priority": 50,
104        "sensorModes": ["SENSOR_MODE_EO", "SENSOR_MODE_IR"]
105    }
106
107    # Path to the JSON schema file (downloaded from Schema Registry)
108    schema_file = 'org.repository.package.v1.schema.jsonschema.bundle.json'
109
110    # Validate the task data (skip if schema file not found)
111    if os.path.isfile(schema_file):
112        if not validate_task_data(task_data, schema_file):
113            print("Task creation aborted due to validation errors.")
114            return
115    else:
116        print("Schema file not found, skipping validation")
117
118    # Create the task with validated data
119    # You can pass an entity_id here if you want to associate the task with an entity
120    entity_id = None  # Replace with an actual entity ID if needed
121    result = await create_task(task_data, entity_id)
122    if result:
123        print(f"Task created with status: {result.status}")
124
125if __name__ == "__main__":
126    asyncio.run(main())

The response contains a Task object with details including the assigned task ID and initial status, which starts as STATUS_CREATED in the task lifecycle.

What’s next?