> This is the Markdown version of the page. For the complete Lattice Developers documentation index, fetch https://developer.anduril.com/llms.txt. Append .md to any page URL for its clean Markdown.

# Command agents and operate on tasks

> Creating, monitoring, and managing tasks using the Lattice SDK

Operators use the Tasks API to create, monitor, and manage tasks across their operational domain.

The specific tasks you command, and monitor, in Lattice depend on the mission requirements, and the environment you're integrating with.
This page shows you how to use the Lattice Sandboxes to create, monitor, and cancel tasks.

The `CreateTask` ([REST](/reference/rest/tasks/create-task))
API lets you create tasks in Lattice using custom task definitions published to the Lattice Schema Registry.
The `StreamTasks` ([REST](/reference/rest/tasks/stream-tasks) | [gRPC](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-stream-tasks))
API provides centralized visibility into task creation, updates, and status changes for all tasks in your environment.
The `CancelTask` ([REST](/reference/rest/tasks/cancel-task) | [gRPC](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-cancel-task))
API allows you to request cancellation of tasks that are no longer needed.

## Before you begin

* To create, monitor, or cancel tasks, [set up](/guides/getting-started/set-up) your Lattice environment.
* Familiarize yourself with [tasks](/guides/tasks/overview) and the task lifecycle.
* To create custom tasks, publish your schemas to the [Lattice Schema Registry](/guides/developer-tools/registry) (LSR).

## Create tasks using REST

When working with custom tasks using REST:

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`](/reference/rest/tasks/create-task) API.

The Lattice Schema Registry (LSR) lets you validate your task data using JSON schema:

#### Get the JSON schema

Log in to the [LSR](https://schema-registry.developer.anduril.com) using your credentials.

Navigate to your organization's repositories.

Select the repository that contains your custom task definitions.

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

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

Download the JSON schema files:

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

This command downloads the bundled JSON Schema file that includes all referenced schemas. For example:
Example JSON Schema structure for a custom task:

```json
{
  "$id": "org.example.reconnaissance.v1beta.Objective.jsonschema.json",
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "properties": {
    "entityId": {
      "type": "string"
    },
    "lla": {
      "$ref": "org.example.reconnaissance.v1beta.LLA.jsonschema.json"
    }
  },
  "type": "object"
}
```

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

#### Create a task

#### Prepare your task data

Define the data for your custom task that matches your schema structure.
This data will be encapsulated in the [`specification`](/reference/rest/tasks/create-task#request.body.specification)
field of the [`CreateTaskRequest`](/reference/rest/tasks/create-task#request) message.

**`Python`**

```python title="Python"
task_data = {
    "lla": {
        "latitudeDegrees": 37.7749,
        "longitudeDegrees": -122.4194,
        "altitudeHaeM": 100.0
    }
}
```

**`TypeScript`**

```typescript title="TypeScript"
const taskData = {
    lla: {
        latitudeDegrees: 37.7749,
        longitudeDegrees: -122.4194,
        altitudeHaeM: 100.0
    }
};
```

**`Java`**

```java title="Java"
ObjectMapper objectMapper = new ObjectMapper();
ObjectNode taskData = objectMapper.createObjectNode();

ObjectNode lla = taskData.putObject("lla");
lla.put("latitudeDegrees", 37.7749);
lla.put("longitudeDegrees", -122.4194);
lla.put("altitudeHaeM", 100.0);
```

**`Go`**

```go title="Go"
taskData := map[string]interface{}{
    "lla": map[string]interface{}{
        "latitudeDegrees":  37.7749,
        "longitudeDegrees": -122.4194,
        "altitudeHaeM":     100.0,
    },
}
```

**`Rust`**

```rust title="Rust"
let task_data = json!({
    "lla": {
        "latitudeDegrees": 37.7749,
        "longitudeDegrees": -122.4194,
        "altitudeHaeM": 100.0
    }
});
```

#### 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.

```py title={"Python"} startLine={33} maxLines={16}
#!/usr/bin/env python3
import asyncio
import json
import os
import sys
from anduril import Lattice
from anduril import (
    GoogleProtobufAny,
    Principal,
    Relations,
    System,
    TaskEntity
)
from jsonschema import validate  # type: ignore[import-untyped]

# Load environment variables
lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
environment_token = os.getenv('ENVIRONMENT_TOKEN')
sandboxes_token = os.getenv('SANDBOXES_TOKEN')  # Remove if not using Sandboxes

if not environment_token or not lattice_endpoint:
    print("Missing required environment variables.")
    sys.exit(1)

# Initialize Lattice client
client = Lattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: environment_token,  # type: ignore[arg-type]
    # Remove the following header if not developing on Sandboxes
    headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"} if sandboxes_token else None
)

def load_schema(schema_file):
    """Load a JSON schema from file."""
    with open(schema_file, 'r') as f:
        return json.load(f)

def validate_task_data(task_data, schema_file):
    """Validate task data against a JSON schema."""
    schema = load_schema(schema_file)

    try:
        validate(instance=task_data, schema=schema)
        print("Task data validation successful")
        return True
    except Exception as e:
        print(f"Validation error: {str(e)}")
        return False

async def create_task(task_data, agent_id, entity_id=None):
    """
    Create a new task using the Lattice SDK.

    Args:
        task_data: The validated task data
        agent_id: The entity ID of the agent to route the task to for execution
        entity_id: Optional entity ID to use as the initial entity
    """
    description = "Custom Task Definition"
    specification_type = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"
    specification = GoogleProtobufAny(
        type=specification_type,
        **task_data
    )
    author = Principal(system=System(service_name="example-service"))
    # Route the task to the agent by setting it as the assignee.
    relations = Relations(assignee=Principal(system=System(entity_id=agent_id)))

    try:
        # Set up initial entities if an entity ID was provided
        initial_entities = None
        if entity_id:
            try:
                print(f"Fetching entity with ID: {entity_id}")
                entity = client.entities.get_entity(entity_id)

                if entity:
                    task_entity = TaskEntity(entity=entity)
                    initial_entities = [task_entity]
                    print(f"Using entity {entity_id} as initial entity for task")
            except Exception as e:
                print(f"Error fetching entity {entity_id}: {str(e)}")

        # Create the task with the validated data
        print("Creating task...")
        response = client.tasks.create_task(
            description=description,
            specification=specification,
            author=author,
            relations=relations,
            is_executed_elsewhere=False,
            initial_entities=initial_entities
        )

        if response.version:
            print(f"Task created successfully with ID: {response.version.task_id}")
        return response
    except Exception as e:
        print(f"Error creating task: {str(e)}")
        return None

async def main():
    # Example task data for a reconnaissance objective: a fixed geodetic point.
    task_data = {
        "lla": {
            "latitudeDegrees": 37.7749,
            "longitudeDegrees": -122.4194,
            "altitudeHaeM": 100.0
        }
    }

    # Path to the JSON schema file (downloaded from Schema Registry)
    schema_file = 'org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json'

    # Validate the task data (skip if schema file not found)
    if os.path.isfile(schema_file):
        if not validate_task_data(task_data, schema_file):
            print("Task creation aborted due to validation errors.")
            return
    else:
        print("Schema file not found, skipping validation")

    # Create the task with validated data
    # Replace with the entity ID of the agent to route the task to.
    agent_id = "<AGENT_ID>"
    # You can pass an entity_id here if you want to associate the task with an entity
    entity_id = None  # Replace with an actual entity ID if needed
    result = await create_task(task_data, agent_id, entity_id)
    if result:
        print(f"Task created with status: {result.status}")

if __name__ == "__main__":
    asyncio.run(main())
```

```ts title={"TypeScript"} startLine={23} maxLines={20}
import { LatticeClient } from '@anduril-industries/lattice-sdk';
import * as fs from 'fs';
import Ajv from 'ajv';

// Load environment variables
const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;
const sandboxesToken = process.env.SANDBOXES_TOKEN;  // Remove if not using Sandboxes

if (!environmentToken || !latticeEndpoint) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

// Initialize Lattice client
const client = new LatticeClient({
    baseUrl: `https://${latticeEndpoint}`,
    token: environmentToken,
    // Remove the following header if not developing on Sandboxes
    headers: sandboxesToken ? { 'anduril-sandbox-authorization': `Bearer ${sandboxesToken}` } : undefined
});

/**
 * Loads a JSON schema from a file
 * @param schemaFile Path to the JSON schema file
 */
function loadSchema(schemaFile: string): Record<string, unknown> {
    const schemaData = fs.readFileSync(schemaFile, 'utf8');
    return JSON.parse(schemaData);
}

/**
 * Validates task data against a JSON schema
 * @param taskData The task data to validate
 * @param schemaFile Path to the JSON schema file
 */
function validateTaskData(taskData: Record<string, unknown>, schemaFile: string): boolean {
    try {
        const schema = loadSchema(schemaFile);
        const ajv = new Ajv.default();
        const validate = ajv.compile(schema);
        const valid = validate(taskData);

        if (!valid) {
            console.log('Validation errors:', validate.errors);
            return false;
        }

        console.log('Task data validation successful');
        return true;
    } catch (error) {
        console.log(`Validation error: ${error}`);
        return false;
    }
}

/**
 * Creates a task using the Lattice SDK
 * @param taskData The validated task data
 * @param entityId Optional entity ID to use as the initial entity
 */
async function createTask(taskData: Record<string, unknown>, agentId: string, entityId?: string) {
    const description = 'Custom Task Definition';
    const specificationType = 'type.googleapis.com/org.example.reconnaissance.v1beta.Objective';

    try {
        // Set up initial entities if an entity ID was provided
        let initialEntities;

        if (entityId) {
            try {
                console.log(`Fetching entity with ID: ${entityId}`);
                const entity = await client.entities.getEntity({ entityId });

                if (entity) {
                    const taskEntity = { entity };
                    initialEntities = [taskEntity];
                    console.log(`Using entity ${entityId} as initial entity for task`);
                }
            } catch (error) {
                console.log(`Error fetching entity ${entityId}: ${error}`);
            }
        }

        // Create the task with validated data
        console.log('Creating task...');
        const response = await client.tasks.createTask({
            description,
            specification: {
                "@type": specificationType,
                ...taskData
            },
            author: {
                system: {
                    serviceName: 'example-service'
                }
            },
            // Route the task to the agent by setting it as the assignee.
            relations: {
                assignee: {
                    system: {
                        entityId: agentId
                    }
                }
            },
            isExecutedElsewhere: false,
            initialEntities
        });

        console.log(`Task created successfully with ID: ${response.version?.taskId}`);
        return response;
    } catch (error) {
        console.log(`Error creating task: ${error}`);
        return null;
    }
}

// Example usage
async function main() {
    // Example task data for a reconnaissance objective: a fixed geodetic point.
    const taskData = {
        lla: {
            latitudeDegrees: 37.7749,
            longitudeDegrees: -122.4194,
            altitudeHaeM: 100.0
        }
    };

    // Path to the JSON schema file (downloaded from Schema Registry)
    const schemaFile = 'org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json';

    // Validate the task data
    if (validateTaskData(taskData, schemaFile)) {
        // Create the task with validated data
        // Replace with the entity ID of the agent to route the task to.
        const agentId = '<AGENT_ID>';
        // You can pass an entityId here if you want to associate the task with an entity
        const entityId = undefined;  // Replace with actual entity ID if needed
        const result = await createTask(taskData, agentId, entityId);
        if (result) {
            console.log(`Task created with status: ${result.status}`);
        }
    } else {
        console.log('Task creation aborted due to validation errors.');
    }
}

main().catch(error => {
    console.error('Unhandled error:', error);
    process.exit(1);
});
```

```java title={"Java"} startLine={25} maxLines={32}
package org.example;

import com.anduril.Lattice;
import com.anduril.resources.tasks.requests.TaskCreation;
import com.anduril.types.Entity;
import com.anduril.types.GoogleProtobufAny;
import com.anduril.types.Principal;
import com.anduril.types.Relations;
import com.anduril.types.Task;
import com.anduril.types.TaskEntity;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.networknt.schema.JsonSchema;
import com.networknt.schema.JsonSchemaFactory;
import com.networknt.schema.SpecVersion;
import com.networknt.schema.ValidationMessage;

import java.io.File;
import java.io.IOException;
import java.util.*;

public class CreateCustomTask {

    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * Load a JSON schema from file
     * @param schemaFile Path to the JSON schema file
     * @return The loaded JSON schema
     * @throws IOException If the schema file cannot be read or parsed
     */
    public static JsonNode loadSchema(String schemaFile) throws IOException {
        return objectMapper.readTree(new File(schemaFile));
    }

    /**
     * Validate task data against a JSON schema
     * @param taskData The task data to validate
     * @param schemaFile Path to the JSON schema file
     * @return true if validation succeeded, false otherwise
     */
    public static boolean validateTaskData(JsonNode taskData, String schemaFile) {
        try {
            // Load the schema
            JsonSchemaFactory factory = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V201909);
            JsonSchema schema = factory.getSchema(new File(schemaFile).toURI());

            // Validate
            Set<ValidationMessage> validationResult = schema.validate(taskData);

            if (validationResult.isEmpty()) {
                System.out.println("Task data validation successful");
                return true;
            } else {
                System.out.println("Validation errors:");
                for (ValidationMessage message : validationResult) {
                    System.out.println("- " + message);
                }
                return false;
            }
        } catch (Exception e) {
            System.out.println("Validation error: " + e.getMessage());
            return false;
        }
    }

    /**
     * Create a task using the Lattice SDK
     * @param client The Lattice client
     * @param taskData The validated task data as a JsonNode
     * @param entityId Optional entity ID to use as the initial entity
     * @return The created Task
     */
    public static Task createTask(Lattice client, JsonNode taskData, String agentId, String entityId) {
        String description = "Custom Task Definition";
        String specificationType = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective";

        @SuppressWarnings("unchecked")
        Map<String, Object> taskDataMap = objectMapper.convertValue(taskData, Map.class);

        GoogleProtobufAny specification = GoogleProtobufAny.builder()
            .type(specificationType)
            .additionalProperties(taskDataMap)
            .build();

        Principal author = Principal.builder()
            .system(com.anduril.types.System.builder()
                .serviceName("example-service")
                .build())
            .build();

        // Route the task to the agent by setting it as the assignee.
        Relations relations = Relations.builder()
            .assignee(Principal.builder()
                .system(com.anduril.types.System.builder()
                    .entityId(agentId)
                    .build())
                .build())
            .build();

        TaskCreation.Builder taskBuilder = TaskCreation.builder()
            .description(description)
            .specification(specification)
            .author(author)
            .relations(relations)
            .isExecutedElsewhere(false);

        if (entityId != null && !entityId.isEmpty()) {
            System.out.println("Fetching entity with ID: " + entityId);
            try {
                Entity entity = client.entities().getEntity(entityId);
                TaskEntity taskEntity = TaskEntity.builder().entity(entity).build();
                taskBuilder.initialEntities(List.of(taskEntity));
                System.out.println("Using entity " + entityId + " as initial entity for task");
            } catch (Exception e) {
                System.out.println("Error fetching entity: " + e.getMessage());
            }
        }

        System.out.println("Creating task...");
        Task task = client.tasks().createTask(taskBuilder.build());
        task.getVersion().ifPresent(v ->
            System.out.println("Task created with ID: " + v.getTaskId())
        );
        return task;
    }

    public static void main(String[] args) {
        try {
            // Get environment variables
            String latticeEndpoint = System.getenv("LATTICE_ENDPOINT");
            String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
            String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Remove if not using Sandboxes

            // Check required environment variables
            if (latticeEndpoint == null || environmentToken == null || sandboxesToken == null) {
                System.out.println("Missing required environment variables");
                System.exit(1);
            }

            // Initialize Lattice client
            Lattice client = Lattice.builder()
                .url("https://" + latticeEndpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Create example task data for a reconnaissance objective: a fixed geodetic point.
            ObjectNode taskData = objectMapper.createObjectNode();
            ObjectNode lla = taskData.putObject("lla");
            lla.put("latitudeDegrees", 37.7749);
            lla.put("longitudeDegrees", -122.4194);
            lla.put("altitudeHaeM", 100.0);

            // Path to the JSON schema file (downloaded from Schema Registry)
            String schemaFile = "org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json";

            // Optional entity ID to associate with the task
            String entityId = null; // Replace with actual entity ID if needed

            // The entity ID of the agent to route the task to.
            String agentId = "<AGENT_ID>";

            // Validate the task data
            if (new File(schemaFile).exists()) {
                if (!validateTaskData(taskData, schemaFile)) {
                    System.out.println("Task creation aborted due to validation errors.");
                    System.exit(1);
                }
            } else {
                System.out.println("Schema file not found, skipping validation");
            }

            Task task = createTask(client, taskData, agentId, entityId);
            task.getVersion().ifPresent(v ->
                System.out.println("Task completed with ID: " + v.getTaskId())
            );
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }
}
```

```go title={"Go"} startLine={17} maxLines={37}
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/xeipuuv/gojsonschema"
)

// loadSchema loads a JSON schema from file
func loadSchema(schemaFile string) (map[string]interface{}, error) {
	data, err := os.ReadFile(schemaFile)
	if err != nil {
		return nil, fmt.Errorf("failed to read schema file: %w", err)
	}

	var schema map[string]interface{}
	if err := json.Unmarshal(data, &schema); err != nil {
		return nil, fmt.Errorf("failed to parse schema: %w", err)
	}

	return schema, nil
}

// validateTaskData validates task data against a JSON schema
func validateTaskData(taskData map[string]interface{}, schemaFile string) bool {
	schemaLoader := gojsonschema.NewReferenceLoader("file://" + schemaFile)
	documentLoader := gojsonschema.NewGoLoader(taskData)

	result, err := gojsonschema.Validate(schemaLoader, documentLoader)
	if err != nil {
		fmt.Printf("Validation error: %v\n", err)
		return false
	}

	if !result.Valid() {
		fmt.Println("Task data is not valid:")
		for _, desc := range result.Errors() {
			fmt.Printf("- %s\n", desc)
		}
		return false
	}

	fmt.Println("Task data validation successful")
	return true
}

// createTask creates a new task using the Lattice SDK
func createTask(ctx context.Context, latticeClient *client.Client, taskData map[string]interface{}, agentID string, entityID string) (*Lattice.TaskVersion, error) {
	description := "Custom Task Definition"
	specificationType := "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"

	specification := &Lattice.GoogleProtobufAny{
		Type:            &specificationType,
		ExtraProperties: taskData,
	}

	serviceName := "example-service"
	author := &Lattice.Principal{
		System: &Lattice.System{
			ServiceName: &serviceName,
		},
	}

	// Route the task to the agent by setting it as the assignee.
	relations := &Lattice.Relations{
		Assignee: &Lattice.Principal{
			System: &Lattice.System{
				EntityID: &agentID,
			},
		},
	}

	var initialEntities []*Lattice.TaskEntity
	if entityID != "" {
		entity, err := latticeClient.Entities.GetEntity(ctx, &Lattice.GetEntityRequest{EntityID: entityID})
		if err != nil {
			fmt.Printf("Error fetching entity %s: %v\n", entityID, err)
		} else {
			taskEntity := &Lattice.TaskEntity{
				Entity: entity,
			}
			initialEntities = []*Lattice.TaskEntity{taskEntity}
			fmt.Printf("Using entity %s as initial entity for task\n", entityID)
		}
	}

	// Create task creation request
	isExecutedElsewhere := false
	taskCreation := &Lattice.TaskCreation{
		Description:         &description,
		Specification:       specification,
		Author:              author,
		Relations:           relations,
		IsExecutedElsewhere: &isExecutedElsewhere,
	}

	if len(initialEntities) > 0 {
		taskCreation.InitialEntities = initialEntities
	}

	// Create the task
	fmt.Println("Creating task...")
	response, err := latticeClient.Tasks.CreateTask(ctx, taskCreation)
	if err != nil {
		return nil, fmt.Errorf("error creating task: %v", err)
	}

	fmt.Printf("Task created successfully with ID: %s\n", *response.Version.TaskID)
	return response.Version, nil
}

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN") // Remove if not using Sandboxes

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	// Create the client
	latticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	// Create context for the request
	ctx := context.Background()

	// Example task data for a reconnaissance objective: a fixed geodetic point.
	taskData := map[string]interface{}{
		"lla": map[string]interface{}{
			"latitudeDegrees":  37.7749,
			"longitudeDegrees": -122.4194,
			"altitudeHaeM":     100.0,
		},
	}

	// Path to the JSON schema file (downloaded from Schema Registry)
	schemaFile := "org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json"

	// Validate the task data (skip if schema file not found)
	if _, err := os.Stat(schemaFile); err == nil {
		if !validateTaskData(taskData, schemaFile) {
			fmt.Println("Task creation aborted due to validation errors.")
			os.Exit(1)
		}
	} else {
		fmt.Println("Schema file not found, skipping validation")
	}

	// Create the task with validated data
	// Replace with the entity ID of the agent to route the task to.
	agentID := "<AGENT_ID>"
	entityID := "" // Replace with an actual entity ID if needed
	_, err := createTask(ctx, latticeClient, taskData, agentID, entityID)
	if err != nil {
		fmt.Printf("Failed to create task: %v\n", err)
		os.Exit(1)
	}
}
```

#### Create the task using validated data

Use the [`CreateTask`](/reference/rest/tasks/create-task) API to submit your task to Lattice. This API expects the following key fields:

* [`description`](/reference/rest/tasks/create-task#request.body.description): A human-readable description of the task (up to 4096 characters).
* [`specification`](/reference/rest/tasks/create-task#request.body.specification): Your validated task data wrapped in a Google Protocol Buffer `Any` message.
* [`author`](/reference/rest/tasks/create-task#request.body.author): Information about who or what created the task.
* [`relations`](/reference/rest/tasks/create-task#request.body.relations): Relationships for the task, such as a parent task or an assignee.
  Set [`relations.assignee`](/reference/rest/tasks/create-task#request.body.relations.assignee) to the agent's `Principal` to route the task to that agent for execution.

```py title={"Python"} startLine={50} maxLines={51}
#!/usr/bin/env python3
import asyncio
import json
import os
import sys
from anduril import Lattice
from anduril import (
    GoogleProtobufAny,
    Principal,
    Relations,
    System,
    TaskEntity
)
from jsonschema import validate  # type: ignore[import-untyped]

# Load environment variables
lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
environment_token = os.getenv('ENVIRONMENT_TOKEN')
sandboxes_token = os.getenv('SANDBOXES_TOKEN')  # Remove if not using Sandboxes

if not environment_token or not lattice_endpoint:
    print("Missing required environment variables.")
    sys.exit(1)

# Initialize Lattice client
client = Lattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: environment_token,  # type: ignore[arg-type]
    # Remove the following header if not developing on Sandboxes
    headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"} if sandboxes_token else None
)

def load_schema(schema_file):
    """Load a JSON schema from file."""
    with open(schema_file, 'r') as f:
        return json.load(f)

def validate_task_data(task_data, schema_file):
    """Validate task data against a JSON schema."""
    schema = load_schema(schema_file)

    try:
        validate(instance=task_data, schema=schema)
        print("Task data validation successful")
        return True
    except Exception as e:
        print(f"Validation error: {str(e)}")
        return False

async def create_task(task_data, agent_id, entity_id=None):
    """
    Create a new task using the Lattice SDK.

    Args:
        task_data: The validated task data
        agent_id: The entity ID of the agent to route the task to for execution
        entity_id: Optional entity ID to use as the initial entity
    """
    description = "Custom Task Definition"
    specification_type = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"
    specification = GoogleProtobufAny(
        type=specification_type,
        **task_data
    )
    author = Principal(system=System(service_name="example-service"))
    # Route the task to the agent by setting it as the assignee.
    relations = Relations(assignee=Principal(system=System(entity_id=agent_id)))

    try:
        # Set up initial entities if an entity ID was provided
        initial_entities = None
        if entity_id:
            try:
                print(f"Fetching entity with ID: {entity_id}")
                entity = client.entities.get_entity(entity_id)

                if entity:
                    task_entity = TaskEntity(entity=entity)
                    initial_entities = [task_entity]
                    print(f"Using entity {entity_id} as initial entity for task")
            except Exception as e:
                print(f"Error fetching entity {entity_id}: {str(e)}")

        # Create the task with the validated data
        print("Creating task...")
        response = client.tasks.create_task(
            description=description,
            specification=specification,
            author=author,
            relations=relations,
            is_executed_elsewhere=False,
            initial_entities=initial_entities
        )

        if response.version:
            print(f"Task created successfully with ID: {response.version.task_id}")
        return response
    except Exception as e:
        print(f"Error creating task: {str(e)}")
        return None

async def main():
    # Example task data for a reconnaissance objective: a fixed geodetic point.
    task_data = {
        "lla": {
            "latitudeDegrees": 37.7749,
            "longitudeDegrees": -122.4194,
            "altitudeHaeM": 100.0
        }
    }

    # Path to the JSON schema file (downloaded from Schema Registry)
    schema_file = 'org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json'

    # Validate the task data (skip if schema file not found)
    if os.path.isfile(schema_file):
        if not validate_task_data(task_data, schema_file):
            print("Task creation aborted due to validation errors.")
            return
    else:
        print("Schema file not found, skipping validation")

    # Create the task with validated data
    # Replace with the entity ID of the agent to route the task to.
    agent_id = "<AGENT_ID>"
    # You can pass an entity_id here if you want to associate the task with an entity
    entity_id = None  # Replace with an actual entity ID if needed
    result = await create_task(task_data, agent_id, entity_id)
    if result:
        print(f"Task created with status: {result.status}")

if __name__ == "__main__":
    asyncio.run(main())
```

```ts title={"TypeScript"} startLine={57} maxLines={60}
import { LatticeClient } from '@anduril-industries/lattice-sdk';
import * as fs from 'fs';
import Ajv from 'ajv';

// Load environment variables
const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;
const sandboxesToken = process.env.SANDBOXES_TOKEN;  // Remove if not using Sandboxes

if (!environmentToken || !latticeEndpoint) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

// Initialize Lattice client
const client = new LatticeClient({
    baseUrl: `https://${latticeEndpoint}`,
    token: environmentToken,
    // Remove the following header if not developing on Sandboxes
    headers: sandboxesToken ? { 'anduril-sandbox-authorization': `Bearer ${sandboxesToken}` } : undefined
});

/**
 * Loads a JSON schema from a file
 * @param schemaFile Path to the JSON schema file
 */
function loadSchema(schemaFile: string): Record<string, unknown> {
    const schemaData = fs.readFileSync(schemaFile, 'utf8');
    return JSON.parse(schemaData);
}

/**
 * Validates task data against a JSON schema
 * @param taskData The task data to validate
 * @param schemaFile Path to the JSON schema file
 */
function validateTaskData(taskData: Record<string, unknown>, schemaFile: string): boolean {
    try {
        const schema = loadSchema(schemaFile);
        const ajv = new Ajv.default();
        const validate = ajv.compile(schema);
        const valid = validate(taskData);

        if (!valid) {
            console.log('Validation errors:', validate.errors);
            return false;
        }

        console.log('Task data validation successful');
        return true;
    } catch (error) {
        console.log(`Validation error: ${error}`);
        return false;
    }
}

/**
 * Creates a task using the Lattice SDK
 * @param taskData The validated task data
 * @param entityId Optional entity ID to use as the initial entity
 */
async function createTask(taskData: Record<string, unknown>, agentId: string, entityId?: string) {
    const description = 'Custom Task Definition';
    const specificationType = 'type.googleapis.com/org.example.reconnaissance.v1beta.Objective';

    try {
        // Set up initial entities if an entity ID was provided
        let initialEntities;

        if (entityId) {
            try {
                console.log(`Fetching entity with ID: ${entityId}`);
                const entity = await client.entities.getEntity({ entityId });

                if (entity) {
                    const taskEntity = { entity };
                    initialEntities = [taskEntity];
                    console.log(`Using entity ${entityId} as initial entity for task`);
                }
            } catch (error) {
                console.log(`Error fetching entity ${entityId}: ${error}`);
            }
        }

        // Create the task with validated data
        console.log('Creating task...');
        const response = await client.tasks.createTask({
            description,
            specification: {
                "@type": specificationType,
                ...taskData
            },
            author: {
                system: {
                    serviceName: 'example-service'
                }
            },
            // Route the task to the agent by setting it as the assignee.
            relations: {
                assignee: {
                    system: {
                        entityId: agentId
                    }
                }
            },
            isExecutedElsewhere: false,
            initialEntities
        });

        console.log(`Task created successfully with ID: ${response.version?.taskId}`);
        return response;
    } catch (error) {
        console.log(`Error creating task: ${error}`);
        return null;
    }
}

// Example usage
async function main() {
    // Example task data for a reconnaissance objective: a fixed geodetic point.
    const taskData = {
        lla: {
            latitudeDegrees: 37.7749,
            longitudeDegrees: -122.4194,
            altitudeHaeM: 100.0
        }
    };

    // Path to the JSON schema file (downloaded from Schema Registry)
    const schemaFile = 'org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json';

    // Validate the task data
    if (validateTaskData(taskData, schemaFile)) {
        // Create the task with validated data
        // Replace with the entity ID of the agent to route the task to.
        const agentId = '<AGENT_ID>';
        // You can pass an entityId here if you want to associate the task with an entity
        const entityId = undefined;  // Replace with actual entity ID if needed
        const result = await createTask(taskData, agentId, entityId);
        if (result) {
            console.log(`Task created with status: ${result.status}`);
        }
    } else {
        console.log('Task creation aborted due to validation errors.');
    }
}

main().catch(error => {
    console.error('Unhandled error:', error);
    process.exit(1);
});
```

```java title={"Java"} startLine={76} maxLines={53}
package org.example;

import com.anduril.Lattice;
import com.anduril.resources.tasks.requests.TaskCreation;
import com.anduril.types.Entity;
import com.anduril.types.GoogleProtobufAny;
import com.anduril.types.Principal;
import com.anduril.types.Relations;
import com.anduril.types.Task;
import com.anduril.types.TaskEntity;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.networknt.schema.JsonSchema;
import com.networknt.schema.JsonSchemaFactory;
import com.networknt.schema.SpecVersion;
import com.networknt.schema.ValidationMessage;

import java.io.File;
import java.io.IOException;
import java.util.*;

public class CreateCustomTask {

    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * Load a JSON schema from file
     * @param schemaFile Path to the JSON schema file
     * @return The loaded JSON schema
     * @throws IOException If the schema file cannot be read or parsed
     */
    public static JsonNode loadSchema(String schemaFile) throws IOException {
        return objectMapper.readTree(new File(schemaFile));
    }

    /**
     * Validate task data against a JSON schema
     * @param taskData The task data to validate
     * @param schemaFile Path to the JSON schema file
     * @return true if validation succeeded, false otherwise
     */
    public static boolean validateTaskData(JsonNode taskData, String schemaFile) {
        try {
            // Load the schema
            JsonSchemaFactory factory = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V201909);
            JsonSchema schema = factory.getSchema(new File(schemaFile).toURI());

            // Validate
            Set<ValidationMessage> validationResult = schema.validate(taskData);

            if (validationResult.isEmpty()) {
                System.out.println("Task data validation successful");
                return true;
            } else {
                System.out.println("Validation errors:");
                for (ValidationMessage message : validationResult) {
                    System.out.println("- " + message);
                }
                return false;
            }
        } catch (Exception e) {
            System.out.println("Validation error: " + e.getMessage());
            return false;
        }
    }

    /**
     * Create a task using the Lattice SDK
     * @param client The Lattice client
     * @param taskData The validated task data as a JsonNode
     * @param entityId Optional entity ID to use as the initial entity
     * @return The created Task
     */
    public static Task createTask(Lattice client, JsonNode taskData, String agentId, String entityId) {
        String description = "Custom Task Definition";
        String specificationType = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective";

        @SuppressWarnings("unchecked")
        Map<String, Object> taskDataMap = objectMapper.convertValue(taskData, Map.class);

        GoogleProtobufAny specification = GoogleProtobufAny.builder()
            .type(specificationType)
            .additionalProperties(taskDataMap)
            .build();

        Principal author = Principal.builder()
            .system(com.anduril.types.System.builder()
                .serviceName("example-service")
                .build())
            .build();

        // Route the task to the agent by setting it as the assignee.
        Relations relations = Relations.builder()
            .assignee(Principal.builder()
                .system(com.anduril.types.System.builder()
                    .entityId(agentId)
                    .build())
                .build())
            .build();

        TaskCreation.Builder taskBuilder = TaskCreation.builder()
            .description(description)
            .specification(specification)
            .author(author)
            .relations(relations)
            .isExecutedElsewhere(false);

        if (entityId != null && !entityId.isEmpty()) {
            System.out.println("Fetching entity with ID: " + entityId);
            try {
                Entity entity = client.entities().getEntity(entityId);
                TaskEntity taskEntity = TaskEntity.builder().entity(entity).build();
                taskBuilder.initialEntities(List.of(taskEntity));
                System.out.println("Using entity " + entityId + " as initial entity for task");
            } catch (Exception e) {
                System.out.println("Error fetching entity: " + e.getMessage());
            }
        }

        System.out.println("Creating task...");
        Task task = client.tasks().createTask(taskBuilder.build());
        task.getVersion().ifPresent(v ->
            System.out.println("Task created with ID: " + v.getTaskId())
        );
        return task;
    }

    public static void main(String[] args) {
        try {
            // Get environment variables
            String latticeEndpoint = System.getenv("LATTICE_ENDPOINT");
            String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
            String sandboxesToken = System.getenv("SANDBOXES_TOKEN"); // Remove if not using Sandboxes

            // Check required environment variables
            if (latticeEndpoint == null || environmentToken == null || sandboxesToken == null) {
                System.out.println("Missing required environment variables");
                System.exit(1);
            }

            // Initialize Lattice client
            Lattice client = Lattice.builder()
                .url("https://" + latticeEndpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Create example task data for a reconnaissance objective: a fixed geodetic point.
            ObjectNode taskData = objectMapper.createObjectNode();
            ObjectNode lla = taskData.putObject("lla");
            lla.put("latitudeDegrees", 37.7749);
            lla.put("longitudeDegrees", -122.4194);
            lla.put("altitudeHaeM", 100.0);

            // Path to the JSON schema file (downloaded from Schema Registry)
            String schemaFile = "org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json";

            // Optional entity ID to associate with the task
            String entityId = null; // Replace with actual entity ID if needed

            // The entity ID of the agent to route the task to.
            String agentId = "<AGENT_ID>";

            // Validate the task data
            if (new File(schemaFile).exists()) {
                if (!validateTaskData(taskData, schemaFile)) {
                    System.out.println("Task creation aborted due to validation errors.");
                    System.exit(1);
                }
            } else {
                System.out.println("Schema file not found, skipping validation");
            }

            Task task = createTask(client, taskData, agentId, entityId);
            task.getVersion().ifPresent(v ->
                System.out.println("Task completed with ID: " + v.getTaskId())
            );
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }
}
```

```go title={"Go"} startLine={54} maxLines={64}
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
	"github.com/xeipuuv/gojsonschema"
)

// loadSchema loads a JSON schema from file
func loadSchema(schemaFile string) (map[string]interface{}, error) {
	data, err := os.ReadFile(schemaFile)
	if err != nil {
		return nil, fmt.Errorf("failed to read schema file: %w", err)
	}

	var schema map[string]interface{}
	if err := json.Unmarshal(data, &schema); err != nil {
		return nil, fmt.Errorf("failed to parse schema: %w", err)
	}

	return schema, nil
}

// validateTaskData validates task data against a JSON schema
func validateTaskData(taskData map[string]interface{}, schemaFile string) bool {
	schemaLoader := gojsonschema.NewReferenceLoader("file://" + schemaFile)
	documentLoader := gojsonschema.NewGoLoader(taskData)

	result, err := gojsonschema.Validate(schemaLoader, documentLoader)
	if err != nil {
		fmt.Printf("Validation error: %v\n", err)
		return false
	}

	if !result.Valid() {
		fmt.Println("Task data is not valid:")
		for _, desc := range result.Errors() {
			fmt.Printf("- %s\n", desc)
		}
		return false
	}

	fmt.Println("Task data validation successful")
	return true
}

// createTask creates a new task using the Lattice SDK
func createTask(ctx context.Context, latticeClient *client.Client, taskData map[string]interface{}, agentID string, entityID string) (*Lattice.TaskVersion, error) {
	description := "Custom Task Definition"
	specificationType := "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"

	specification := &Lattice.GoogleProtobufAny{
		Type:            &specificationType,
		ExtraProperties: taskData,
	}

	serviceName := "example-service"
	author := &Lattice.Principal{
		System: &Lattice.System{
			ServiceName: &serviceName,
		},
	}

	// Route the task to the agent by setting it as the assignee.
	relations := &Lattice.Relations{
		Assignee: &Lattice.Principal{
			System: &Lattice.System{
				EntityID: &agentID,
			},
		},
	}

	var initialEntities []*Lattice.TaskEntity
	if entityID != "" {
		entity, err := latticeClient.Entities.GetEntity(ctx, &Lattice.GetEntityRequest{EntityID: entityID})
		if err != nil {
			fmt.Printf("Error fetching entity %s: %v\n", entityID, err)
		} else {
			taskEntity := &Lattice.TaskEntity{
				Entity: entity,
			}
			initialEntities = []*Lattice.TaskEntity{taskEntity}
			fmt.Printf("Using entity %s as initial entity for task\n", entityID)
		}
	}

	// Create task creation request
	isExecutedElsewhere := false
	taskCreation := &Lattice.TaskCreation{
		Description:         &description,
		Specification:       specification,
		Author:              author,
		Relations:           relations,
		IsExecutedElsewhere: &isExecutedElsewhere,
	}

	if len(initialEntities) > 0 {
		taskCreation.InitialEntities = initialEntities
	}

	// Create the task
	fmt.Println("Creating task...")
	response, err := latticeClient.Tasks.CreateTask(ctx, taskCreation)
	if err != nil {
		return nil, fmt.Errorf("error creating task: %v", err)
	}

	fmt.Printf("Task created successfully with ID: %s\n", *response.Version.TaskID)
	return response.Version, nil
}

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN") // Remove if not using Sandboxes

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	// Create the client
	latticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	// Create context for the request
	ctx := context.Background()

	// Example task data for a reconnaissance objective: a fixed geodetic point.
	taskData := map[string]interface{}{
		"lla": map[string]interface{}{
			"latitudeDegrees":  37.7749,
			"longitudeDegrees": -122.4194,
			"altitudeHaeM":     100.0,
		},
	}

	// Path to the JSON schema file (downloaded from Schema Registry)
	schemaFile := "org.example.reconnaissance.v1beta.Objective.jsonschema.bundle.json"

	// Validate the task data (skip if schema file not found)
	if _, err := os.Stat(schemaFile); err == nil {
		if !validateTaskData(taskData, schemaFile) {
			fmt.Println("Task creation aborted due to validation errors.")
			os.Exit(1)
		}
	} else {
		fmt.Println("Schema file not found, skipping validation")
	}

	// Create the task with validated data
	// Replace with the entity ID of the agent to route the task to.
	agentID := "<AGENT_ID>"
	entityID := "" // Replace with an actual entity ID if needed
	_, err := createTask(ctx, latticeClient, taskData, agentID, entityID)
	if err != nil {
		fmt.Printf("Failed to create task: %v\n", err)
		os.Exit(1)
	}
}
```

```rs title={"Rust"} startLine={76} maxLines={64}
use anduril_lattice_sdk::prelude::*;

use std::env;

// Type URL of the built-in Investigate task (anduril.tasks.v2.Investigate).
// Swap this for your own registered type URL (e.g. one defined in your own
// .proto file and published to the Schema Registry) to create a task from a
// custom specification instead.
const SPECIFICATION_TYPE: &str = "type.googleapis.com/anduril.tasks.v2.Investigate";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    // Remove sandboxes_token handling if you are not developing on Sandboxes.
    let sandboxes_token = env::var("SANDBOXES_TOKEN").ok();

    if environment_token.is_empty() || lattice_endpoint.is_empty() {
        eprintln!("Missing required environment variables:");
        eprintln!("  LATTICE_ENDPOINT");
        eprintln!("  ENVIRONMENT_TOKEN");
        std::process::exit(1);
    }

    let mut custom_headers = HashMap::new();
    // Remove this header if you are not developing on Sandboxes.
    if let Some(sandboxes_token) = sandboxes_token {
        custom_headers.insert(
            "anduril-sandbox-authorization".to_string(),
            format!("Bearer {}", sandboxes_token),
        );
    }
    let config = ClientConfig {
        base_url: format!("https://{}", lattice_endpoint),
        token: Some(environment_token),
        custom_headers,
        ..Default::default()
    };
    let client = Lattice::new(config)?;

    // Example task data: an Investigate objective targeting a fixed point.
    // Over REST, the specification is a google.protobuf.Any: a `type` URL
    // alongside the task's flattened fields (carried in the `extra` map).
    let task_data = json!({
        "objective": {
            "point": {
                "lla": {
                    "lat": 37.7749,
                    "lon": -122.4194,
                    "alt": 100.0
                }
            }
        }
    });

    // Replace with the entity ID of the agent to route the task to.
    let agent_id = "<AGENT_ID>";
    // Optionally associate the task with an initial entity.
    let entity_id: Option<&str> = None;

    match create_task(&client, task_data, agent_id, entity_id).await {
        Some(task) => println!("Task created with status: {:?}", task.status),
        None => {}
    }

    Ok(())
}

/// Create a new task, optionally seeding it with an initial entity.
async fn create_task(
    client: &Lattice,
    task_data: Value,
    agent_id: &str,
    entity_id: Option<&str>,
) -> Option<Task> {
    // Build the google.protobuf.Any specification: a type URL plus the task's
    // flattened fields in the `extra` map.
    let mut extra = HashMap::new();
    if let Value::Object(map) = task_data {
        for (key, value) in map {
            extra.insert(key, value);
        }
    }
    let specification = GoogleProtobufAny {
        r#type: Some(SPECIFICATION_TYPE.to_string()),
        extra,
    };

    let author = Principal {
        system: Some(System {
            service_name: Some("example-service".to_string()),
            ..Default::default()
        }),
        ..Default::default()
    };
    // Route the task to the agent by setting it as the assignee.
    let relations = Relations {
        assignee: Some(Principal {
            system: Some(System {
                entity_id: Some(agent_id.to_string()),
                ..Default::default()
            }),
            ..Default::default()
        }),
        ..Default::default()
    };

    // Set up an initial entity if an entity ID was provided.
    let initial_entities = if let Some(entity_id) = entity_id {
        println!("Fetching entity with ID: {}", entity_id);
        match client.entities.get_entity(entity_id, None).await {
            Ok(entity) => {
                println!("Using entity {} as initial entity for task", entity_id);
                Some(vec![TaskEntity {
                    entity: Some(entity),
                    ..Default::default()
                }])
            }
            Err(error) => {
                println!("Error fetching entity {}: {}", entity_id, error);
                None
            }
        }
    } else {
        None
    };

    let request = TaskCreation {
        description: Some("Custom Task Definition".to_string()),
        specification: Some(specification),
        author: Some(author),
        relations: Some(relations),
        is_executed_elsewhere: Some(false),
        initial_entities,
        ..Default::default()
    };

    println!("Creating task...");
    match client.tasks.create_task(&request, None).await {
        Ok(task) => {
            if let Some(version) = &task.version {
                println!("Task created successfully with ID: {:?}", version.task_id);
            }
            Some(task)
        }
        Err(error) => {
            println!("Error creating task: {}", error);
            None
        }
    }
}

```

The response contains a [`Task`](/reference/rest/tasks/create-task#response.body.task) object with details including the assigned task ID and initial status,
which starts as `STATUS_CREATED` in the task lifecycle.

## Create tasks using gRPC

When you use gRPC, the Protobuf types are generated from your task definition directly, so there's
no separate JSON schema validation step. Instead, the compiler enforces the task's shape. You build your
custom task message, pack it into the [`specification`](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-create-task)
`Any` field, and call `CreateTask`.

#### Create a task

#### Build and pack the task specification

Construct your `Objective` message from the bindings generated from your
[task definition](/guides/tasks/define-a-task), then pack it into a `google.protobuf.Any`. The
type URL on the `Any` is what lets the receiving agent identify and unpack the task:

```py title={"Python (gRPC)"} startLine={28} maxLines={13}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof
from protobuf.wkt import Any

from anduril.taskmanager.v1.task_pub_pb import (
    Principal,
    Relations,
    System,
    User,
)
from anduril.taskmanager.v1.task_manager_api_pub_pb import CreateTaskRequest
from anduril.taskmanager.v1.task_manager_api_pub_connect import TaskManagerAPIClientSync

# Your custom task from schema-registry.developer.anduril.com, generated with the
# same bufbuild/py plugin as the Lattice SDK.
from org.example.reconnaissance.v1beta.objective_pb import (
    LLA,
    Objective,
)


def create_task(client, agent_entity_id):
    """Create an Objective task and route it to an agent."""
    # Build the custom task message. Objective is a oneof; here we set an LLA point.
    objective = Objective(
        target=Oneof(
            "lla",
            LLA(
                latitude_degrees=37.7749,
                longitude_degrees=-122.4194,
                altitude_hae_m=100.0,
            ),
        ),
    )

    # Pack the task into a protobuf Any for the specification field.
    specification = Any.pack(objective)

    request = CreateTaskRequest(
        description="Reconnaissance objective",
        specification=specification,
        # The service creating the task.
        author=Principal(agent=Oneof("user", User(user_id="operator@example.com"))),
        # Route the task to the agent by setting it as the assignee.
        relations=Relations(
            assignee=Principal(agent=Oneof("system", System(entity_id=agent_entity_id)))
        ),
    )

    response = client.create_task(request)
    return response.task


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(TaskManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Replace with the entity ID of the agent to route the task to.
    agent_entity_id = "<AGENT_ID>"

    task = create_task(client, agent_entity_id)
    print(f"Created task with ID: {task.version.task_id}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"} startLine={64} maxLines={14}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::taskmanager::v1::{
    principal, CreateTaskRequest, Principal, Relations, System, User,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::taskmanager::v1::tonic::task_manager_api_client::TaskManagerApiClient;
//Your custom task from schema-registry.developer.anduril.com
use org_example_reconnaissance_community_neoeinstein_prost::org::example::reconnaissance::v1beta::{
    objective, Lla, Objective,
};
use bearer_auth::EnvironmentTokenAuth;
use prost::Message;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Type URL of the Objective task defined in objective.proto.
const OBJECTIVE_TYPE: &str =
    "type.googleapis.com/org.example.reconnaissance.v1beta.Objective";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token().to_string();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = TaskManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Replace with the entity ID of the agent to route the task to.
    let agent_entity_id = "<AGENT_ID>".to_string();

    // Build the custom task message. Objective is a oneof; here we set an LLA point.
    let objective = Objective {
        target: Some(objective::Target::Lla(Lla {
            latitude_degrees: 37.7749,
            longitude_degrees: -122.4194,
            altitude_hae_m: 100.0,
        })),
    };

    // Pack the task into a google.protobuf.Any for the specification field.
    let specification = prost_types::Any {
        type_url: OBJECTIVE_TYPE.to_string(),
        value: objective.encode_to_vec(),
    };

    let request = Request::new(CreateTaskRequest {
        description: "Reconnaissance objective".to_string(),
        specification: Some(specification),
        // The operator creating the task.
        author: Some(Principal {
            agent: Some(principal::Agent::User(User {
                user_id: "operator@example.com".to_string(),
            })),
            ..Default::default()
        }),
        // Route the task to the agent by setting it as the assignee.
        relations: Some(Relations {
            assignee: Some(Principal {
                agent: Some(principal::Agent::System(System {
                    entity_id: agent_entity_id.clone(),
                    ..Default::default()
                })),
                ..Default::default()
            }),
            ..Default::default()
        }),
        ..Default::default()
    });

    let response = client.create_task(request).await?;
    if let Some(version) = response.into_inner().task.and_then(|t| t.version) {
        println!("Created task with ID: {}", version.task_id);
    }

    Ok(())
}

```

#### Create the task

Call the [`CreateTask`](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-create-task)
RPC with the packed specification. Set the following key fields on the
[`CreateTaskRequest`](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-create-task):

* `specification`: Your custom task message packed into a `google.protobuf.Any`.
* `author`: The `Principal` -- a user or system -- creating the task.
* `relations.assignee`: The agent's `Principal`. Set this to route the task to that agent for execution.
* `description`: A human-readable description of the task.

```py title={"Python (gRPC)"} startLine={42} maxLines={11}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof
from protobuf.wkt import Any

from anduril.taskmanager.v1.task_pub_pb import (
    Principal,
    Relations,
    System,
    User,
)
from anduril.taskmanager.v1.task_manager_api_pub_pb import CreateTaskRequest
from anduril.taskmanager.v1.task_manager_api_pub_connect import TaskManagerAPIClientSync

# Your custom task from schema-registry.developer.anduril.com, generated with the
# same bufbuild/py plugin as the Lattice SDK.
from org.example.reconnaissance.v1beta.objective_pb import (
    LLA,
    Objective,
)


def create_task(client, agent_entity_id):
    """Create an Objective task and route it to an agent."""
    # Build the custom task message. Objective is a oneof; here we set an LLA point.
    objective = Objective(
        target=Oneof(
            "lla",
            LLA(
                latitude_degrees=37.7749,
                longitude_degrees=-122.4194,
                altitude_hae_m=100.0,
            ),
        ),
    )

    # Pack the task into a protobuf Any for the specification field.
    specification = Any.pack(objective)

    request = CreateTaskRequest(
        description="Reconnaissance objective",
        specification=specification,
        # The service creating the task.
        author=Principal(agent=Oneof("user", User(user_id="operator@example.com"))),
        # Route the task to the agent by setting it as the assignee.
        relations=Relations(
            assignee=Principal(agent=Oneof("system", System(entity_id=agent_entity_id)))
        ),
    )

    response = client.create_task(request)
    return response.task


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(TaskManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Replace with the entity ID of the agent to route the task to.
    agent_entity_id = "<AGENT_ID>"

    task = create_task(client, agent_entity_id)
    print(f"Created task with ID: {task.version.task_id}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"} startLine={79} maxLines={28}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::taskmanager::v1::{
    principal, CreateTaskRequest, Principal, Relations, System, User,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::taskmanager::v1::tonic::task_manager_api_client::TaskManagerApiClient;
//Your custom task from schema-registry.developer.anduril.com
use org_example_reconnaissance_community_neoeinstein_prost::org::example::reconnaissance::v1beta::{
    objective, Lla, Objective,
};
use bearer_auth::EnvironmentTokenAuth;
use prost::Message;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Type URL of the Objective task defined in objective.proto.
const OBJECTIVE_TYPE: &str =
    "type.googleapis.com/org.example.reconnaissance.v1beta.Objective";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token().to_string();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = TaskManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Replace with the entity ID of the agent to route the task to.
    let agent_entity_id = "<AGENT_ID>".to_string();

    // Build the custom task message. Objective is a oneof; here we set an LLA point.
    let objective = Objective {
        target: Some(objective::Target::Lla(Lla {
            latitude_degrees: 37.7749,
            longitude_degrees: -122.4194,
            altitude_hae_m: 100.0,
        })),
    };

    // Pack the task into a google.protobuf.Any for the specification field.
    let specification = prost_types::Any {
        type_url: OBJECTIVE_TYPE.to_string(),
        value: objective.encode_to_vec(),
    };

    let request = Request::new(CreateTaskRequest {
        description: "Reconnaissance objective".to_string(),
        specification: Some(specification),
        // The operator creating the task.
        author: Some(Principal {
            agent: Some(principal::Agent::User(User {
                user_id: "operator@example.com".to_string(),
            })),
            ..Default::default()
        }),
        // Route the task to the agent by setting it as the assignee.
        relations: Some(Relations {
            assignee: Some(Principal {
                agent: Some(principal::Agent::System(System {
                    entity_id: agent_entity_id.clone(),
                    ..Default::default()
                })),
                ..Default::default()
            }),
            ..Default::default()
        }),
        ..Default::default()
    });

    let response = client.create_task(request).await?;
    if let Some(version) = response.into_inner().task.and_then(|t| t.version) {
        println!("Created task with ID: {}", version.task_id);
    }

    Ok(())
}

```

The response contains the created [`Task`](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-create-task),
including its assigned task ID.

## Monitor tasks

The `StreamTasks` ([REST](/reference/rest/tasks/stream-tasks) | [gRPC](/reference/grpc/anduril-taskmanager-v-1/task-manager-api/anduril-taskmanager-v-1-stream-tasks))
API establishes a server-sent events (SSE) stream that lets you monitor tasks in Lattice:

#### Initialize the task stream

In the following example, we specify a prefix to filter the task stream with:

```go title={"Go (REST)"} startLine={40} maxLines={25}
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	fmt.Println("Starting task stream...")

	ctx := context.Background()

	// Create task stream request
	request := &Lattice.TaskStreamRequest{
		HeartbeatIntervalMs:     Lattice.Int(10000),
		ExcludePreexistingTasks: Lattice.Bool(false),
		TaskType: &Lattice.TaskStreamRequestTaskType{
			TaskStreamRequestTaskTypeTaskTypePrefix: &Lattice.TaskStreamRequestTaskTypeTaskTypePrefix{
				// Define a prefix filter for tasks that belong to your organization
				TaskTypePrefix: "type.googleapis.com/<your-organization>.tasks",
			},
		},
	}

	// Start the task stream
	stream, err := LatticeClient.Tasks.StreamTasks(ctx, request)
	if err != nil {
		fmt.Printf("Failed to start task stream: %v\n", err)
		return
	}
	defer stream.Close()

	// Process the stream events as they arrive
	for {
		event, err := stream.Recv()
		if err != nil {
			fmt.Printf("Error receiving task stream event: %v\n", err)
			return
		}

		if event.Event == "heartbeat" {
			// Process heartbeat events
			timestamp := *event.GetHeartbeat().GetTimestamp()
			fmt.Printf("Heartbeat: %v\n", timestamp)
		} else {
			task := event.GetTaskEvent().GetTaskEvent().GetTask()
			taskId := task.GetVersion().GetTaskID()
			status := task.GetStatus().GetStatus()

			fmt.Printf("  TaskID: %v\n", *taskId)
			fmt.Printf("  Status: %v\n", *status)
		}
	}
}

```

```java title={"Java (REST)"} startLine={39} maxLines={25}
package org.example;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import com.anduril.AsyncLattice;
import com.anduril.resources.tasks.requests.TaskStreamRequest;
import com.anduril.resources.tasks.types.StreamTasksResponse;
import com.anduril.resources.tasks.types.TaskStreamRequestTaskType;
import com.anduril.resources.tasks.types.TaskStreamRequestTaskTypeTaskTypePrefix;
import com.anduril.types.Task;

/**
 * StreamTasks demonstrates how to use the Lattice SDK to stream task events.
 */
public class StreamTasks {
    public static void main(String[] args) {
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        // Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Initialize the Lattice client
            AsyncLattice client = AsyncLattice.builder()
                .url(endpoint)
                .token(environmentToken)
                // Remove the following if you are not developing on Sandboxes.
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            streamTasks(client);
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }

    /**
     * Stream tasks and process events as they arrive.
     *
     * @param client Lattice client
     */
    private static void streamTasks(AsyncLattice client) {
        System.out.println("Starting task stream");

        try {
            // Define a prefix filter for tasks that belong to your organization
            TaskStreamRequestTaskTypeTaskTypePrefix taskTypePrefix = TaskStreamRequestTaskTypeTaskTypePrefix.builder()
                .taskTypePrefix("type.googleapis.com/<your-organization>.tasks")
                .build();

            // Create task stream request
            TaskStreamRequest request = TaskStreamRequest.builder()
                .heartbeatIntervalMs(10000) // Send heartbeats every 10 seconds (default: 30 seconds)
                .excludePreexistingTasks(false) // Include tasks that existed before starting the stream
                .taskType(TaskStreamRequestTaskType.of(taskTypePrefix))
                .build();

            // Start streaming tasks
            CompletableFuture<Iterable<StreamTasksResponse>> streamFuture = client.tasks().streamTasks(request);

            // Process the stream events as they arrive
            Iterable<StreamTasksResponse> stream = streamFuture.get();
            for (StreamTasksResponse event : stream) {
                if (event.getHeartbeat() != null && event.getHeartbeat().isPresent()) {
                    // Process heartbeat events
                    String timestamp = event.getHeartbeat().get().getTimestamp().get();
                    System.out.println("Heartbeat: " + timestamp);
                } else if (event.getTaskEvent() != null && event.getTaskEvent().isPresent()) {
                    // Process task events
                    Task task = event.getTaskEvent().get()
                        .getTaskEvent().get()
                        .getTask().get();
                    String taskDescription = task.getDescription().get();

                    // Extract and print relevant task information
                    String taskId = task.getVersion().get().getTaskId().get();
                    String status = task.getStatus().get().toString();

                    System.out.println("Task Event: " + taskDescription);
                    System.out.println("  TaskID: " + taskId);
                    System.out.println("  Status: " + status);
                } else {
                    System.out.println("Unknown event type");
                }
            }
        } catch (ExecutionException e) {
            System.err.println("Execution error during task streaming: " + e.getCause().getMessage());
        } catch (InterruptedException e) {
            System.err.println("Task streaming interrupted: " + e.getMessage());
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            System.err.println("Error during task streaming: " + e.getMessage());
        }
    }
}

```

```py title={"Python (REST)"} startLine={25} maxLines={25}
from anduril import AsyncLattice
from anduril import TaskStreamRequestTaskTypeTaskTypePrefix
import asyncio
import os
import sys

lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
environment_token = os.getenv('ENVIRONMENT_TOKEN')

# Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
sandboxes_token = os.getenv('SANDBOXES_TOKEN')
if not environment_token or not lattice_endpoint or not sandboxes_token:
    print("Missing environment variables.")
    sys.exit(1)

# Initialize the Lattice client
client = AsyncLattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: str(environment_token),
    # Remove the following header if you are not developing on Sandboxes.
    headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}
)

async def stream_tasks():
    try:
        print("Starting task stream")

        # Stream tasks with various configuration options
        task_stream = client.tasks.stream_tasks(
            heartbeat_interval_ms=10000,  # Send heartbeats every 10 seconds (default: 30 seconds)
            exclude_preexisting_tasks=False,  # Include tasks that existed before starting the stream
            task_type=TaskStreamRequestTaskTypeTaskTypePrefix(
                # Define a prefix filter only tasks that belong to your orgnaization
                task_type_prefix="type.googleapis.com/<your-organization>.tasks"
            )
        )

        # Process the stream events as they arrive
        async for event in task_stream:
            if event.event == "heartbeat":
                print(f"Heartbeat: {event.timestamp}")
            elif event.event == "task_event" and event.task_event and event.task_event.task:
                task = event.task_event.task
                if task.version and task.status:
                    print(f"Task Event: {event.task_event.event_type}")
                    print(f"  TaskID: {task.version.task_id}")
                    print(f"  Status: {task.status.status}")
            else:
                print(f"Unknown event type: {event.event}")

    except asyncio.CancelledError:
        print("Task streaming cancelled.")
    except Exception as error:
        print(f"Exception during task streaming: {error}")

if __name__ == "__main__":
    asyncio.run(stream_tasks())

```

```ts title={"TypeScript (REST)"} startLine={18} maxLines={25}
import { LatticeClient } from "@anduril-industries/lattice-sdk";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;
// Remove sandboxesToken from the following statement if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;

if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` }
    }
);

async function streamTasks() {
    try {
        console.log("Starting task stream");

        // Stream tasks with various configuration options
        const taskStream = await client.tasks.streamTasks({
            heartbeatIntervalMs: 10000, // Send heartbeats every 10 seconds (default: 30 seconds)
            excludePreexistingTasks: false, // Include tasks that existed before starting the stream
            // Define a prefix filter for tasks that belong to your organization
            taskType: {
                taskTypePrefix: "type.googleapis.com/<your-organization>.tasks"
            }
        });

        // Process the stream events as they arrive
        for await (const event of taskStream) {
            if (event.event === "heartbeat") {
                console.log(`Heartbeat: ${event?.timestamp}`);
                continue;
            } else {
                console.log(`Task ID: ${event?.taskEvent?.task?.version?.taskId}`)
                console.log(`Task status: ${event?.taskEvent?.task?.status?.status}`)
            }
        }
    } catch (error) {
        console.log(`Exception during task streaming: ${error}`);
    }
}

streamTasks();

```

```rs title={"Rust (REST)"} startLine={33} maxLines={25}
use anduril_lattice_sdk::prelude::*;
use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    // Remove sandboxes_token if you are not developing on Sandboxes.
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    if environment_token.is_empty() || lattice_endpoint.is_empty() || sandboxes_token.is_empty() {
        eprintln!("Missing environment variables.");
        std::process::exit(1);
    }

    let config = ClientConfig {
        base_url: format!("https://{}", lattice_endpoint),
        token: Some(environment_token),
        // Remove custom_headers if you are not developing on Sandboxes.
        custom_headers: HashMap::from([(
            "anduril-sandbox-authorization".to_string(),
            format!("Bearer {}", sandboxes_token),
        )]),
        ..Default::default()
    };
    let client = Lattice::new(config)?;

    println!("Starting task stream");

    // Stream tasks with various configuration options.
    let request = TaskStreamRequest {
        // Send heartbeats every 10 seconds (default: 30 seconds).
        heartbeat_interval_ms: Some(10000),
        // Include tasks that existed before starting the stream.
        exclude_preexisting_tasks: Some(false),
        // Define a prefix filter for only tasks that belong to your organization.
        task_type: Some(TaskStreamRequestTaskType::TaskStreamRequestTaskTypeTaskTypePrefix(
            TaskStreamRequestTaskTypeTaskTypePrefix {
                task_type_prefix: "type.googleapis.com/<your-organization>.tasks".to_string(),
            },
        )),
        ..Default::default()
    };

    let stream = client.tasks.stream_tasks(&request, None).await?;
    let mut stream = std::pin::pin!(stream);

    while let Some(result) = stream.next().await {
        match result {
            Ok(StreamTasksResponse::Heartbeat { data, .. }) => {
                println!("Heartbeat: {:?}", data.heartbeat_object_fields.timestamp);
            }
            Ok(StreamTasksResponse::TaskEvent { data, .. }) => {
                if let Some(task_event) = &data.task_event_data_fields.task_event {
                    if let Some(task) = &task_event.task {
                        if let (Some(version), Some(status)) = (&task.version, &task.status) {
                            println!("Task Event: {:?}", task_event.event_type);
                            println!("  TaskID: {:?}", version.task_id);
                            println!("  Status: {:?}", status.status);
                        }
                    }
                }
            }
            Ok(other) => println!("Unknown event type: {:?}", other),
            Err(error) => eprintln!("Exception during task streaming: {}", error),
        }
    }

    Ok(())
}

```

```go title={"Go (gRPC)"} startLine={50} maxLines={30}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"os"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/taskmanager/v1/taskmanagerv1grpc"
	taskmanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/taskmanager/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Setup authentication
	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}

	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := taskmanagerv1grpc.NewTaskManagerAPIClient(conn)

	fmt.Println("Starting task stream...")

	ctx := context.Background()

	// Create task stream request
	request := &taskmanagerv1.StreamTasksRequest{
		HeartbeatIntervalMs:     35000,
		ExcludePreexistingTasks: false,
		TaskType: &taskmanagerv1.StreamTasksRequest_TaskTypePrefix{
			// Filter the stream to only receive your tasks in the stream
			TaskTypePrefix: "type.googleapis.com/<YOUR_ORGANIZATION>",
		},
	}

	// Start the task stream
	stream, err := client.StreamTasks(ctx, request)
	if err != nil {
		fmt.Printf("Failed to start task stream: %v\n", err)
		return
	}

	// Process the stream events as they arrive
	for {
		response, err := stream.Recv()
		if err == io.EOF {
			fmt.Println("Stream ended")
			return
		}
		if err != nil {
			fmt.Printf("Error receiving task stream event: %v\n", err)
			return
		}
		print(response)
		if response.GetHeartbeat() != nil {
			// Process heartbeat events
			timestamp := response.GetHeartbeat().GetTimestamp()
			fmt.Printf("Heartbeat: %v\n", timestamp.AsTime())
		} else if response.GetTaskEvent() != nil {
			// Process task events
			task := response.GetTaskEvent().GetTask()
			taskId := task.GetVersion().GetTaskId()
			status := task.GetStatus().GetStatus()

			fmt.Printf("  TaskID: %v\n", taskId)
			fmt.Printf("  Status: %v\n", status)
		}
	}
}

```

```py title={"Python (gRPC)"} startLine={32} maxLines={20}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof

from anduril.taskmanager.v1.task_manager_api_pub_pb import StreamTasksRequest
from anduril.taskmanager.v1.task_manager_api_pub_connect import TaskManagerAPIClientSync


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(TaskManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    print("Starting task stream...")

    request = StreamTasksRequest(
        heartbeat_interval_ms=35000,
        exclude_preexisting_tasks=False,
        # Filter the stream to only receive your tasks in the stream
        task_type=Oneof("task_type_prefix", "type.googleapis.com/<YOUR_ORGANIZATION>"),
    )

    try:
        stream = client.stream_tasks(request)
    except Exception as error:
        print(f"Failed to start task stream: {error}", file=sys.stderr)
        return

    for response in stream:
        if response.has_field("heartbeat"):
            timestamp = response.heartbeat.timestamp.to_datetime()
            print(f"Heartbeat: {timestamp}")
        elif response.has_field("task_event"):
            task = response.task_event.task
            task_id = task.version.task_id
            status = task.status.status
            print(f"  TaskID: {task_id}")
            print(f"  Status: {status}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"} startLine={56} maxLines={30}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::taskmanager::v1::{
    stream_tasks_request, StreamTasksRequest,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::taskmanager::v1::tonic::task_manager_api_client::TaskManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = TaskManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    println!("Starting task stream...");

    // Create task stream request. The task_type_prefix filters the stream so
    // that only task types matching the prefix are delivered.
    let request = Request::new(StreamTasksRequest {
        heartbeat_interval_ms: 35_000,
        exclude_preexisting_tasks: false,
        task_type: Some(stream_tasks_request::TaskType::TaskTypePrefix(
            "type.googleapis.com/<YOUR_ORGANIZATION>".to_string(),
        )),
        ..Default::default()
    });

    // Start the task stream
    let mut stream = match client.stream_tasks(request).await {
        Ok(resp) => resp.into_inner(),
        Err(e) => {
            eprintln!("Failed to start task stream: {}", e);
            return Ok(());
        }
    };

    // Process the stream events as they arrive
    loop {
        let response = match stream.message().await {
            Ok(Some(r)) => r,
            Ok(None) => {
                println!("Stream ended");
                break;
            }
            Err(e) => {
                eprintln!("Error receiving task stream event: {}", e);
                break;
            }
        };

        if let Some(heartbeat) = response.heartbeat {
            // Process heartbeat events
            if let Some(ts) = heartbeat.timestamp {
                println!("Heartbeat: {}.{:09}", ts.seconds, ts.nanos);
            }
        } else if let Some(task_event) = response.task_event {
            // Process task events
            if let Some(task) = task_event.task {
                let task_id = task
                    .version
                    .as_ref()
                    .map(|v| v.task_id.clone())
                    .unwrap_or_default();
                let status = task
                    .status
                    .as_ref()
                    .map(|s| s.status)
                    .unwrap_or(0);
                println!("  TaskID: {}", task_id);
                println!("  Status: {}", status);
            }
        }
    }

    Ok(())
}

```

The `StreamTasks` method accepts the following parameters:

**`heartbeatIntervalMs`** `Number`

The interval in milliseconds at which the server sends heartbeat events (default: 30000 ms).
Use heartbeats to verify the connection is still active. The minimum is 1000 ms; smaller
values are raised to 1000 ms.

---

**`excludePreexistingTasks`** `Boolean`

When set to `true`, the stream will only include tasks created after the stream starts.
When `false` (default), existing tasks will also be included in the stream.

---

**`taskType`** `Object`

Specifies which task types to include in the stream. Can filter by exact match or by prefix.

**`taskTypePrefix`** `String`

Only include tasks whose type starts with the specified prefix.
For example, `type.googleapis.com/<your-organization.tasks>`.

---

**`taskTypeUrls`** `List`

Only include tasks specified in the list. You must list the full URL of the
task definition for each of the tasks you want to filter from the stream.

---

---

#### Process stream events

Process the different types of events that arrive through the stream:

```go title={"Go (REST)"} startLine={59} maxLines={30}
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	fmt.Println("Starting task stream...")

	ctx := context.Background()

	// Create task stream request
	request := &Lattice.TaskStreamRequest{
		HeartbeatIntervalMs:     Lattice.Int(10000),
		ExcludePreexistingTasks: Lattice.Bool(false),
		TaskType: &Lattice.TaskStreamRequestTaskType{
			TaskStreamRequestTaskTypeTaskTypePrefix: &Lattice.TaskStreamRequestTaskTypeTaskTypePrefix{
				// Define a prefix filter for tasks that belong to your organization
				TaskTypePrefix: "type.googleapis.com/<your-organization>.tasks",
			},
		},
	}

	// Start the task stream
	stream, err := LatticeClient.Tasks.StreamTasks(ctx, request)
	if err != nil {
		fmt.Printf("Failed to start task stream: %v\n", err)
		return
	}
	defer stream.Close()

	// Process the stream events as they arrive
	for {
		event, err := stream.Recv()
		if err != nil {
			fmt.Printf("Error receiving task stream event: %v\n", err)
			return
		}

		if event.Event == "heartbeat" {
			// Process heartbeat events
			timestamp := *event.GetHeartbeat().GetTimestamp()
			fmt.Printf("Heartbeat: %v\n", timestamp)
		} else {
			task := event.GetTaskEvent().GetTaskEvent().GetTask()
			taskId := task.GetVersion().GetTaskID()
			status := task.GetStatus().GetStatus()

			fmt.Printf("  TaskID: %v\n", *taskId)
			fmt.Printf("  Status: %v\n", *status)
		}
	}
}

```

```java title={"Java (REST)"} startLine={69} maxLines={30}
package org.example;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import com.anduril.AsyncLattice;
import com.anduril.resources.tasks.requests.TaskStreamRequest;
import com.anduril.resources.tasks.types.StreamTasksResponse;
import com.anduril.resources.tasks.types.TaskStreamRequestTaskType;
import com.anduril.resources.tasks.types.TaskStreamRequestTaskTypeTaskTypePrefix;
import com.anduril.types.Task;

/**
 * StreamTasks demonstrates how to use the Lattice SDK to stream task events.
 */
public class StreamTasks {
    public static void main(String[] args) {
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        // Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        try {
            // Initialize the Lattice client
            AsyncLattice client = AsyncLattice.builder()
                .url(endpoint)
                .token(environmentToken)
                // Remove the following if you are not developing on Sandboxes.
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            streamTasks(client);
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }

    /**
     * Stream tasks and process events as they arrive.
     *
     * @param client Lattice client
     */
    private static void streamTasks(AsyncLattice client) {
        System.out.println("Starting task stream");

        try {
            // Define a prefix filter for tasks that belong to your organization
            TaskStreamRequestTaskTypeTaskTypePrefix taskTypePrefix = TaskStreamRequestTaskTypeTaskTypePrefix.builder()
                .taskTypePrefix("type.googleapis.com/<your-organization>.tasks")
                .build();

            // Create task stream request
            TaskStreamRequest request = TaskStreamRequest.builder()
                .heartbeatIntervalMs(10000) // Send heartbeats every 10 seconds (default: 30 seconds)
                .excludePreexistingTasks(false) // Include tasks that existed before starting the stream
                .taskType(TaskStreamRequestTaskType.of(taskTypePrefix))
                .build();

            // Start streaming tasks
            CompletableFuture<Iterable<StreamTasksResponse>> streamFuture = client.tasks().streamTasks(request);

            // Process the stream events as they arrive
            Iterable<StreamTasksResponse> stream = streamFuture.get();
            for (StreamTasksResponse event : stream) {
                if (event.getHeartbeat() != null && event.getHeartbeat().isPresent()) {
                    // Process heartbeat events
                    String timestamp = event.getHeartbeat().get().getTimestamp().get();
                    System.out.println("Heartbeat: " + timestamp);
                } else if (event.getTaskEvent() != null && event.getTaskEvent().isPresent()) {
                    // Process task events
                    Task task = event.getTaskEvent().get()
                        .getTaskEvent().get()
                        .getTask().get();
                    String taskDescription = task.getDescription().get();

                    // Extract and print relevant task information
                    String taskId = task.getVersion().get().getTaskId().get();
                    String status = task.getStatus().get().toString();

                    System.out.println("Task Event: " + taskDescription);
                    System.out.println("  TaskID: " + taskId);
                    System.out.println("  Status: " + status);
                } else {
                    System.out.println("Unknown event type");
                }
            }
        } catch (ExecutionException e) {
            System.err.println("Execution error during task streaming: " + e.getCause().getMessage());
        } catch (InterruptedException e) {
            System.err.println("Task streaming interrupted: " + e.getMessage());
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            System.err.println("Error during task streaming: " + e.getMessage());
        }
    }
}

```

```py title={"Python (REST)"} startLine={39} maxLines={30}
from anduril import AsyncLattice
from anduril import TaskStreamRequestTaskTypeTaskTypePrefix
import asyncio
import os
import sys

lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
environment_token = os.getenv('ENVIRONMENT_TOKEN')

# Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
sandboxes_token = os.getenv('SANDBOXES_TOKEN')
if not environment_token or not lattice_endpoint or not sandboxes_token:
    print("Missing environment variables.")
    sys.exit(1)

# Initialize the Lattice client
client = AsyncLattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: str(environment_token),
    # Remove the following header if you are not developing on Sandboxes.
    headers={"anduril-sandbox-authorization": f"Bearer {sandboxes_token}"}
)

async def stream_tasks():
    try:
        print("Starting task stream")

        # Stream tasks with various configuration options
        task_stream = client.tasks.stream_tasks(
            heartbeat_interval_ms=10000,  # Send heartbeats every 10 seconds (default: 30 seconds)
            exclude_preexisting_tasks=False,  # Include tasks that existed before starting the stream
            task_type=TaskStreamRequestTaskTypeTaskTypePrefix(
                # Define a prefix filter only tasks that belong to your orgnaization
                task_type_prefix="type.googleapis.com/<your-organization>.tasks"
            )
        )

        # Process the stream events as they arrive
        async for event in task_stream:
            if event.event == "heartbeat":
                print(f"Heartbeat: {event.timestamp}")
            elif event.event == "task_event" and event.task_event and event.task_event.task:
                task = event.task_event.task
                if task.version and task.status:
                    print(f"Task Event: {event.task_event.event_type}")
                    print(f"  TaskID: {task.version.task_id}")
                    print(f"  Status: {task.status.status}")
            else:
                print(f"Unknown event type: {event.event}")

    except asyncio.CancelledError:
        print("Task streaming cancelled.")
    except Exception as error:
        print(f"Exception during task streaming: {error}")

if __name__ == "__main__":
    asyncio.run(stream_tasks())

```

```ts title={"TypeScript (REST)"} startLine={39} maxLines={30}
import { LatticeClient } from "@anduril-industries/lattice-sdk";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;
// Remove sandboxesToken from the following statement if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;

if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

const client = new LatticeClient(
    {
        baseUrl: `https://${latticeEndpoint}`,
        token: environmentToken,
        // Remove the following if you are not developing on Sandboxes.
        headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` }
    }
);

async function streamTasks() {
    try {
        console.log("Starting task stream");

        // Stream tasks with various configuration options
        const taskStream = await client.tasks.streamTasks({
            heartbeatIntervalMs: 10000, // Send heartbeats every 10 seconds (default: 30 seconds)
            excludePreexistingTasks: false, // Include tasks that existed before starting the stream
            // Define a prefix filter for tasks that belong to your organization
            taskType: {
                taskTypePrefix: "type.googleapis.com/<your-organization>.tasks"
            }
        });

        // Process the stream events as they arrive
        for await (const event of taskStream) {
            if (event.event === "heartbeat") {
                console.log(`Heartbeat: ${event?.timestamp}`);
                continue;
            } else {
                console.log(`Task ID: ${event?.taskEvent?.task?.version?.taskId}`)
                console.log(`Task status: ${event?.taskEvent?.task?.status?.status}`)
            }
        }
    } catch (error) {
        console.log(`Exception during task streaming: ${error}`);
    }
}

streamTasks();

```

```rs title={"Rust (REST)"} startLine={51} maxLines={30}
use anduril_lattice_sdk::prelude::*;
use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    // Remove sandboxes_token if you are not developing on Sandboxes.
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    if environment_token.is_empty() || lattice_endpoint.is_empty() || sandboxes_token.is_empty() {
        eprintln!("Missing environment variables.");
        std::process::exit(1);
    }

    let config = ClientConfig {
        base_url: format!("https://{}", lattice_endpoint),
        token: Some(environment_token),
        // Remove custom_headers if you are not developing on Sandboxes.
        custom_headers: HashMap::from([(
            "anduril-sandbox-authorization".to_string(),
            format!("Bearer {}", sandboxes_token),
        )]),
        ..Default::default()
    };
    let client = Lattice::new(config)?;

    println!("Starting task stream");

    // Stream tasks with various configuration options.
    let request = TaskStreamRequest {
        // Send heartbeats every 10 seconds (default: 30 seconds).
        heartbeat_interval_ms: Some(10000),
        // Include tasks that existed before starting the stream.
        exclude_preexisting_tasks: Some(false),
        // Define a prefix filter for only tasks that belong to your organization.
        task_type: Some(TaskStreamRequestTaskType::TaskStreamRequestTaskTypeTaskTypePrefix(
            TaskStreamRequestTaskTypeTaskTypePrefix {
                task_type_prefix: "type.googleapis.com/<your-organization>.tasks".to_string(),
            },
        )),
        ..Default::default()
    };

    let stream = client.tasks.stream_tasks(&request, None).await?;
    let mut stream = std::pin::pin!(stream);

    while let Some(result) = stream.next().await {
        match result {
            Ok(StreamTasksResponse::Heartbeat { data, .. }) => {
                println!("Heartbeat: {:?}", data.heartbeat_object_fields.timestamp);
            }
            Ok(StreamTasksResponse::TaskEvent { data, .. }) => {
                if let Some(task_event) = &data.task_event_data_fields.task_event {
                    if let Some(task) = &task_event.task {
                        if let (Some(version), Some(status)) = (&task.version, &task.status) {
                            println!("Task Event: {:?}", task_event.event_type);
                            println!("  TaskID: {:?}", version.task_id);
                            println!("  Status: {:?}", status.status);
                        }
                    }
                }
            }
            Ok(other) => println!("Unknown event type: {:?}", other),
            Err(error) => eprintln!("Exception during task streaming: {}", error),
        }
    }

    Ok(())
}

```

```go title={"Go (gRPC)"} startLine={67} maxLines={30}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"os"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/taskmanager/v1/taskmanagerv1grpc"
	taskmanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/taskmanager/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables")
		os.Exit(1)
	}

	// Setup authentication
	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}

	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := taskmanagerv1grpc.NewTaskManagerAPIClient(conn)

	fmt.Println("Starting task stream...")

	ctx := context.Background()

	// Create task stream request
	request := &taskmanagerv1.StreamTasksRequest{
		HeartbeatIntervalMs:     35000,
		ExcludePreexistingTasks: false,
		TaskType: &taskmanagerv1.StreamTasksRequest_TaskTypePrefix{
			// Filter the stream to only receive your tasks in the stream
			TaskTypePrefix: "type.googleapis.com/<YOUR_ORGANIZATION>",
		},
	}

	// Start the task stream
	stream, err := client.StreamTasks(ctx, request)
	if err != nil {
		fmt.Printf("Failed to start task stream: %v\n", err)
		return
	}

	// Process the stream events as they arrive
	for {
		response, err := stream.Recv()
		if err == io.EOF {
			fmt.Println("Stream ended")
			return
		}
		if err != nil {
			fmt.Printf("Error receiving task stream event: %v\n", err)
			return
		}
		print(response)
		if response.GetHeartbeat() != nil {
			// Process heartbeat events
			timestamp := response.GetHeartbeat().GetTimestamp()
			fmt.Printf("Heartbeat: %v\n", timestamp.AsTime())
		} else if response.GetTaskEvent() != nil {
			// Process task events
			task := response.GetTaskEvent().GetTask()
			taskId := task.GetVersion().GetTaskId()
			status := task.GetStatus().GetStatus()

			fmt.Printf("  TaskID: %v\n", taskId)
			fmt.Printf("  Status: %v\n", status)
		}
	}
}

```

```py title={"Python (gRPC)"} startLine={45} maxLines={15}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof

from anduril.taskmanager.v1.task_manager_api_pub_pb import StreamTasksRequest
from anduril.taskmanager.v1.task_manager_api_pub_connect import TaskManagerAPIClientSync


def main():
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Create the API client over Connect gRPC with authentication
    client = create_client(TaskManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    print("Starting task stream...")

    request = StreamTasksRequest(
        heartbeat_interval_ms=35000,
        exclude_preexisting_tasks=False,
        # Filter the stream to only receive your tasks in the stream
        task_type=Oneof("task_type_prefix", "type.googleapis.com/<YOUR_ORGANIZATION>"),
    )

    try:
        stream = client.stream_tasks(request)
    except Exception as error:
        print(f"Failed to start task stream: {error}", file=sys.stderr)
        return

    for response in stream:
        if response.has_field("heartbeat"):
            timestamp = response.heartbeat.timestamp.to_datetime()
            print(f"Heartbeat: {timestamp}")
        elif response.has_field("task_event"):
            task = response.task_event.task
            task_id = task.version.task_id
            status = task.status.status
            print(f"  TaskID: {task_id}")
            print(f"  Status: {status}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"} startLine={74} maxLines={30}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::taskmanager::v1::{
    stream_tasks_request, StreamTasksRequest,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::taskmanager::v1::tonic::task_manager_api_client::TaskManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = TaskManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    println!("Starting task stream...");

    // Create task stream request. The task_type_prefix filters the stream so
    // that only task types matching the prefix are delivered.
    let request = Request::new(StreamTasksRequest {
        heartbeat_interval_ms: 35_000,
        exclude_preexisting_tasks: false,
        task_type: Some(stream_tasks_request::TaskType::TaskTypePrefix(
            "type.googleapis.com/<YOUR_ORGANIZATION>".to_string(),
        )),
        ..Default::default()
    });

    // Start the task stream
    let mut stream = match client.stream_tasks(request).await {
        Ok(resp) => resp.into_inner(),
        Err(e) => {
            eprintln!("Failed to start task stream: {}", e);
            return Ok(());
        }
    };

    // Process the stream events as they arrive
    loop {
        let response = match stream.message().await {
            Ok(Some(r)) => r,
            Ok(None) => {
                println!("Stream ended");
                break;
            }
            Err(e) => {
                eprintln!("Error receiving task stream event: {}", e);
                break;
            }
        };

        if let Some(heartbeat) = response.heartbeat {
            // Process heartbeat events
            if let Some(ts) = heartbeat.timestamp {
                println!("Heartbeat: {}.{:09}", ts.seconds, ts.nanos);
            }
        } else if let Some(task_event) = response.task_event {
            // Process task events
            if let Some(task) = task_event.task {
                let task_id = task
                    .version
                    .as_ref()
                    .map(|v| v.task_id.clone())
                    .unwrap_or_default();
                let status = task
                    .status
                    .as_ref()
                    .map(|s| s.status)
                    .unwrap_or(0);
                println!("  TaskID: {}", task_id);
                println!("  Status: {}", status);
            }
        }
    }

    Ok(())
}

```

The stream produces two main event types:

* **`heartbeat`**: Regular server heartbeats that confirm the connection is active.
* **`task_event`**: Notifications about task creation, updates, or status changes.

#### Verify the results

When successfully running the `StreamTasks` code, you'll see output similar to the following:

```bash
Starting task stream
Heartbeat: 2025-01-29T12:34:56.789Z
Task Event: TASK_CREATED
  TaskID: task-123456
  Status: STATUS_SENT
Task Event: TASK_UPDATED
  TaskID: task-123456
  Status: STATUS_MACHINE_RECEIPT
Heartbeat: 2025-01-29T12:35:06.789Z
Task Event: TASK_UPDATED
  TaskID: task-123456
  Status: STATUS_EXECUTING
```

## Cancel tasks

The [`CancelTask`](/reference/rest/tasks/cancel-task) API cancels a task by marking it for cancellation in the system.
The behavior depends on the task's current state:

* If the task has not been sent to an agent, it cancels immediately and transitions to a terminal state
  ([`STATUS_DONE_NOT_OK`](/reference/rest/tasks/update-task-status#request.body.newStatus.status)
  with [`ERROR_CODE_CANCELLED`](/reference/rest/tasks/update-task-status#request.body.newStatus.taskError.code)).
* If the task has been sent to an agent, the cancellation request is routed to the agent,
  which decides whether to accept or reject the cancellation.

> **Note**
>
> This section covers the operator's side of cancellation: sending the request and reading the
> result. The agent that receives the routed request must set up a task processor and handle
> cancellation requests -- see [Handle task cancellation](/guides/tasks/integrate-an-agent#handle-task-cancellation)
> in the agent integration guide.

#### Request task cancellation

Send a task cancellation request to Lattice. If the task has already been sent to an agent,
Lattice routes the request to the agent:

```go title={"Go (REST)"} startLine={53} maxLines={20}
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"

	Lattice "github.com/anduril/lattice-sdk-go/v5"
	"github.com/anduril/lattice-sdk-go/v5/client"
	"github.com/anduril/lattice-sdk-go/v5/option"
)

func main() {
	// Get environment variables
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	// Check required environment variables
	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables.")
		os.Exit(1)
	}

	// Replace with the ID of the task to cancel and, optionally, the entity ID
	// to attribute the cancellation to.
	taskId := "<TASK_ID>"
	entityIdValue := "<ENTITY_ID>"
	entityId := &entityIdValue

	// Initialize headers for sandbox authorization
	headers := http.Header{}
	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))

	// Create the client
	LatticeClient := client.NewClient(
		option.WithToken(environmentToken),
		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
		option.WithHTTPHeader(headers),
	)

	// Create context for the request
	ctx := context.Background()

	// Build author based on whether entity_id is provided
	var author *Lattice.Principal
	if entityId != nil {
		author = &Lattice.Principal{
			System: &Lattice.System{
				EntityID: entityId,
			},
		}
		fmt.Printf("Cancelling task: %s\n", taskId)
		fmt.Printf("Author entity: %s\n", *entityId)
	} else {
		userId := "operator_1"
		author = &Lattice.Principal{
			User: &Lattice.User{
				UserID: &userId,
			},
		}
		fmt.Printf("Cancelling task: %s\n", taskId)
	}

	// Create cancel task request
	cancelRequest := Lattice.TaskCancellation{
		TaskID: taskId,
		Author: author,
	}

	// Call the CancelTask API
	_, err := LatticeClient.Tasks.CancelTask(ctx, &cancelRequest)
	if err != nil {
		fmt.Printf("Error cancelling task: %v\n", err)
		os.Exit(1)
	}

	fmt.Printf("Cancel task response: {}\n")
}

```

```java title={"Java (REST)"} startLine={39} maxLines={20}
package org.example;

import com.anduril.Lattice;
import com.anduril.resources.tasks.requests.TaskCancellation;
import com.anduril.types.Principal;
import com.anduril.types.User;
public class CancelTask {
    public static void main(String[] args) {
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables.");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;

        // Replace with the ID of the task to cancel and, optionally, the entity ID
        // to attribute the cancellation to.
        String taskId = "<TASK_ID>";
        String entityId = "<ENTITY_ID>";

        try {
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Build author based on whether entity_id is provided
            Principal author;
            if (entityId != null) {
                author = Principal.builder()
                    .system(com.anduril.types.System.builder()
                        .entityId(entityId)
                        .build())
                    .build();
                System.out.println("Cancelling task: " + taskId);
                System.out.println("Author entity: " + entityId);
            } else {
                author = Principal.builder()
                    .user(User.builder()
                        .userId("operator_1")
                        .build())
                    .build();
                System.out.println("Cancelling task: " + taskId);
            }

            // Create cancel task request
            TaskCancellation cancelRequest = TaskCancellation.builder()
                .author(author)
                .build();

            // Call the CancelTask API
            client.tasks().cancelTask(taskId, cancelRequest);
        } catch (Exception e) {
            System.err.println("Error cancelling task: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"} startLine={22} maxLines={20}
from anduril import AsyncLattice, Principal, System, User
import asyncio
import os
import sys

lattice_endpoint = os.getenv('LATTICE_ENDPOINT')
environment_token = os.getenv('ENVIRONMENT_TOKEN')

# Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
sandboxes_token = os.getenv('SANDBOXES_TOKEN')
if not environment_token or not lattice_endpoint or not sandboxes_token:
    print("Missing required environment variables.")
    sys.exit(1)

client = AsyncLattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: str(environment_token),
    # Remove the following header if you are not developing on Sandboxes.
    headers={ "anduril-sandbox-authorization": f"Bearer {sandboxes_token}" }
)

async def cancel_task(task_id, entity_id=None):
    try:
        # Build author if entity_id is provided
        author = None
        if entity_id:
            author = Principal(
                system=System(
                    entity_id=entity_id
                )
            )
        else:
            author = Principal(
                user=User(
                    user_id="operator_1"
                )
            )

        # Call the cancel_task API
        response = await client.tasks.cancel_task(
            task_id=task_id,
            author=author
        )
        print(f"Cancel task response: {response}")
    except Exception as error:
        print(f"Error cancelling task: {error}")
        raise

async def main():
    # Replace with the ID of the task to cancel and, optionally, the entity ID
    # to attribute the cancellation to.
    task_id = "<TASK_ID>"
    entity_id = "<ENTITY_ID>"

    print(f"Cancelling task: {task_id}")
    if entity_id:
        print(f"Author entity: {entity_id}")

    await cancel_task(task_id, entity_id)

if __name__ == "__main__":
    asyncio.run(main())

```

```ts title={"TypeScript (REST)"} startLine={24} maxLines={20}
import { LatticeClient } from "@anduril-industries/lattice-sdk";

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;
const sandboxesToken = process.env.SANDBOXES_TOKEN;

if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.error("Missing required environment variables.");
    process.exit(1);
}

const client = new LatticeClient({
    baseUrl: latticeEndpoint.startsWith("https://") ? latticeEndpoint : `https://${latticeEndpoint}`,
    token: environmentToken,
    headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` }
});

async function App() {
    // Replace with the ID of the task to cancel and, optionally, the entity ID
    // to attribute the cancellation to.
    const taskId = "<TASK_ID>";
    const entityId: string | undefined = "<ENTITY_ID>";

    try {
        // Build author based on whether entity_id is provided
        let author;
        if (entityId) {
            author = {
                system: {
                    entityId: entityId
                }
            };
            console.log(`Cancelling task: ${taskId}`);
            console.log(`Author entity: ${entityId}`);
        } else {
            author = {
                user: {
                    userId: "operator_1"
                }
            };
            console.log(`Cancelling task: ${taskId}`);
        }

        // Call the CancelTask API
        await client.tasks.cancelTask({
            taskId: taskId,
            author: author
        });

        console.log("Task cancellation request sent successfully.");
    } catch (error) {
        console.error(`Error cancelling task: ${error}`);
        process.exit(1);
    }
}

App();

```

```rs title={"Rust (REST)"} startLine={52} maxLines={20}
use anduril_lattice_sdk::prelude::*;

use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    // Remove sandboxes_token if you are not developing on Sandboxes.
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    if environment_token.is_empty() || lattice_endpoint.is_empty() || sandboxes_token.is_empty() {
        eprintln!("Missing required environment variables:");
        eprintln!("  LATTICE_ENDPOINT");
        eprintln!("  ENVIRONMENT_TOKEN");
        eprintln!("  SANDBOXES_TOKEN");
        std::process::exit(1);
    }

    let config = ClientConfig {
        base_url: format!("https://{}", lattice_endpoint),
        token: Some(environment_token),
        // Remove custom_headers if you are not developing on Sandboxes.
        custom_headers: HashMap::from([(
            "anduril-sandbox-authorization".to_string(),
            format!("Bearer {}", sandboxes_token),
        )]),
        ..Default::default()
    };
    let client = Lattice::new(config)?;

    // Replace with the ID of the task to cancel and, optionally, the entity ID
    // to attribute the cancellation to.
    let task_id = "<TASK_ID>";
    let entity_id = Some("<ENTITY_ID>");

    println!("Cancelling task: {}", task_id);
    if let Some(entity_id) = entity_id {
        println!("Author entity: {}", entity_id);
    }

    cancel_task(&client, task_id, entity_id).await;

    Ok(())
}

/// Cancel a task, attributing the cancellation to a system entity when an
/// entity ID is provided, or to a user principal otherwise.
async fn cancel_task(client: &Lattice, task_id: &str, entity_id: Option<&str>) {
    // Build the author principal.
    let author = if let Some(entity_id) = entity_id {
        Principal {
            system: Some(System {
                entity_id: Some(entity_id.to_string()),
                ..Default::default()
            }),
            ..Default::default()
        }
    } else {
        Principal {
            user: Some(User {
                user_id: Some("operator_1".to_string()),
                ..Default::default()
            }),
            ..Default::default()
        }
    };

    let request = TaskCancellation {
        author: Some(author),
    };

    match client.tasks.cancel_task(task_id, &request, None).await {
        Ok(response) => println!("Cancel task response: {:?}", response),
        Err(error) => eprintln!("Error cancelling task: {}", error),
    }
}

```

```go title={"Go (gRPC)"} startLine={56} maxLines={20}
// This Go example is compatible with artifacts generated using
// the following grpc/go plugin: https://buf.build/anduril/lattice-sdk/sdks/main:grpc/go
package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/taskmanager/v1/taskmanagerv1grpc"
	taskmanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/taskmanager/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func main() {
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		fmt.Println("Missing required environment variables.")
		os.Exit(1)
	}

	// Replace with the ID of the task to cancel and, optionally, the entity ID
	// to attribute the cancellation to.
	taskId := "<TASK_ID>"
	entityId := "<ENTITY_ID>"

	// Setup authentication
	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")),
		grpc.WithPerRPCCredentials(auth),
	}

	conn, err := grpc.NewClient(latticeEndpoint, opts...)
	if err != nil {
		log.Fatalf("Did not connect: %v", err)
	}
	defer conn.Close()

	client := taskmanagerv1grpc.NewTaskManagerAPIClient(conn)

	ctx := context.Background()

	// Build author based on whether entity_id is provided
	var author *taskmanagerv1.Principal
	if entityId != "" {
		author = &taskmanagerv1.Principal{
			Agent: &taskmanagerv1.Principal_System{
				System: &taskmanagerv1.System{
					EntityId: entityId,
				},
			},
		}
		fmt.Printf("Cancelling task: %s\n", taskId)
		fmt.Printf("Author entity: %s\n", entityId)
	} else {
		author = &taskmanagerv1.Principal{
			Agent: &taskmanagerv1.Principal_User{
				User: &taskmanagerv1.User{
					UserId: "operator_1",
				},
			},
		}
		fmt.Printf("Cancelling task: %s\n", taskId)
	}

	// Create cancel task request
	cancelRequest := &taskmanagerv1.CancelTaskRequest{
		TaskId: taskId,
		Author: author,
	}

	// Call the CancelTask API
	_, err = client.CancelTask(ctx, cancelRequest)
	if err != nil {
		log.Fatalf("Error cancelling task: %v", err)
	}

	fmt.Printf("Cancel task response: {}\n")
}

```

```py title={"Python (gRPC)"} startLine={47} maxLines={10}
# This Python example is compatible with artifacts generated using the following
# Buf plugins: https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:bufbuild/py
# and https://schema-registry.developer.anduril.com/anduril/lattice-sdk/sdks/main:connectrpc/py

import os
import sys
from bearer_auth import create_client
from protobuf import Oneof

from anduril.taskmanager.v1.task_pub_pb import Principal, System, User
from anduril.taskmanager.v1.task_manager_api_pub_pb import CancelTaskRequest
from anduril.taskmanager.v1.task_manager_api_pub_connect import TaskManagerAPIClientSync


def main():
    # Load environment variables
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

    if not environment_token or not lattice_endpoint or not sandboxes_token:
        print("Missing required environment variables", file=sys.stderr)
        sys.exit(1)

    # Replace with the ID of the task to cancel and, optionally, the entity ID
    # to attribute the cancellation to.
    task_id = "<TASK_ID>"
    entity_id = "<ENTITY_ID>"

    # Create the API client over Connect gRPC with authentication
    client = create_client(TaskManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Build author based on whether entity_id was provided
    if entity_id:
        author = Principal(agent=Oneof("system", System(entity_id=entity_id)))
        print(f"Cancelling task: {task_id}")
        print(f"Author entity: {entity_id}")
    else:
        author = Principal(agent=Oneof("user", User(user_id="operator_1")))
        print(f"Cancelling task: {task_id}")

    request = CancelTaskRequest(task_id=task_id, author=author)

    try:
        client.cancel_task(request)
        print("Cancel task response: {}")
    except Exception as error:
        print(f"Error cancelling task: {error}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"} startLine={59} maxLines={20}
// This Rust example is compatible with artifacts generated using
// the following grpc/rust plugin: https://buf.build/anduril/lattice-sdk/sdks/main:community/neoeinstein-tonic
mod bearer_auth;

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::taskmanager::v1::{
    principal, CancelTaskRequest, Principal, System, User,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::taskmanager::v1::tonic::task_manager_api_client::TaskManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let environment_token = env::var("ENVIRONMENT_TOKEN")
        .expect("ENVIRONMENT_TOKEN environment variable not set");
    let lattice_endpoint = env::var("LATTICE_ENDPOINT")
        .expect("LATTICE_ENDPOINT environment variable not set");
    let sandboxes_token = env::var("SANDBOXES_TOKEN")
        .expect("SANDBOXES_TOKEN environment variable not set");

    // Replace with the ID of the task to cancel and, optionally, the entity ID
    // to attribute the cancellation to.
    let task_id = "<TASK_ID>".to_string();
    let entity_id = Some("<ENTITY_ID>".to_string());

    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

    let tls_config = ClientTlsConfig::new().with_native_roots();
    let channel = Channel::from_shared(format!("https://{}", lattice_endpoint))?
        .tls_config(tls_config)?
        .connect()
        .await?;

    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    let mut client = TaskManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Build the author principal. When an entity ID is provided we cancel the
    // task as a system (entity-backed agent); otherwise we cancel as a user.
    let author = match entity_id.as_deref() {
        Some(entity_id) if !entity_id.is_empty() => {
            println!("Cancelling task: {}", task_id);
            println!("Author entity: {}", entity_id);
            Principal {
                agent: Some(principal::Agent::System(System {
                    entity_id: entity_id.to_string(),
                    ..Default::default()
                })),
                ..Default::default()
            }
        }
        _ => {
            println!("Cancelling task: {}", task_id);
            Principal {
                agent: Some(principal::Agent::User(User {
                    user_id: "operator_1".to_string(),
                })),
                ..Default::default()
            }
        }
    };

    // Send the CancelTask request.
    let request = Request::new(CancelTaskRequest {
        task_id,
        author: Some(author),
    });
    client.cancel_task(request).await?;

    println!("Cancel task response: {{}}");
    Ok(())
}

```

The `CancelTask` API accepts the following parameters:

**`taskId`** `String` — required

The unique identifier of the task to cancel.

---

**`author`** `Principal` — required

Who or what is requesting the cancellation. Can be a user [`Principal`](/reference/rest/tasks/cancel-task#request.body.author)
with a [`userId`](/reference/rest/tasks/cancel-task#request.body.author.user.userId) or a
system `Principal` with an [`entityId`](/reference/rest/tasks/cancel-task#request.body.author.system.entityId).

---

Replace the `<TASK_ID>` and `<ENTITY_ID>` placeholders in the example with the ID of the task you want to cancel
and the ID of the entity attributed as the author, then run it:

```bash
python cancel_task.py
```

#### Verify the results

After requesting cancellation, the task will transition to one of two outcomes:

**Successful cancellation:**

When the agent accepts the cancellation, the task transitions to
`STATUS_DONE_NOT_OK` with `ERROR_CODE_CANCELLED`:

```bash
Cancelling task: task-123456
Task has been cancelled.
Task Status: STATUS_DONE_NOT_OK
Error Code: ERROR_CODE_CANCELLED
Error Message: Task cancelled by agent.
```

**Rejected cancellation:**

When the agent rejects the cancellation, the task retains its current status but includes a
`TaskError` with `ERROR_CODE_REJECTED`:

```bash
Cancelling task: task-123456
Task could not be cancelled.
Task Status: STATUS_EXECUTING
Error Code: ERROR_CODE_REJECTED
Error Message: Task is already active, and cannot be cancelled.
```

For more information, see `CancelTask` in the *Lattice API Reference*.

## What's next

* Learn how to integrate an agent to [listen for, and process, tasks](/guides/tasks/integrate-an-agent) using the Lattice SDK.