Reference

Entities

client.entities.publish_entity(...) -> Entity

📝 Description

Publish an entity for ingest into the Entities API. Entities created with this method are "owned" by the originator: other sources, such as the UI, may not edit or delete these entities. The server validates entities at API call time and returns an error if the entity is invalid.

An entity ID must be provided when calling this endpoint. If the entity referenced by the entity ID does not exist then it will be created. Otherwise the entity will be updated. An entity will only be updated if its provenance.sourceUpdateTime is greater than the provenance.sourceUpdateTime of the existing entity.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.entities.publish_entity()

⚙️ Parameters

request: Entity

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.entities.get_entity(...) -> Entity

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.entities.get_entity(
    entity_id="entityId",
)

⚙️ Parameters

entity_id: str — ID of the entity to return

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.entities.override_entity(...) -> Entity

📝 Description

Only fields marked with overridable can be overridden. Please refer to our documentation to see the comprehensive list of fields that can be overridden. The entity in the request body should only have a value set on the field specified in the field path parameter. Field paths are rooted in the base entity object and must be represented using lower_snake_case. Do not include "entity" in the field path.

Note that overrides are applied in an eventually consistent manner. If multiple overrides are created concurrently for the same field path, the last writer wins.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.entities.override_entity(
    entity_id="entityId",
    field_path="mil_view.disposition",
)

⚙️ Parameters

entity_id: str — The unique ID of the entity to override

field_path: str — fieldPath to override

entity: typing.Optional[Entity]

The entity containing the overridden fields. The service will extract the overridable fields from the object and ignore all other fields.

provenance: typing.Optional[Provenance] — Additional information about the source of the override.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.entities.remove_entity_override(...) -> Entity

📝 Description

This operation clears the override value from the specified field path on the entity.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.entities.remove_entity_override(
    entity_id="entityId",
    field_path="mil_view.disposition",
)

⚙️ Parameters

entity_id: str — The unique ID of the entity to undo an override from.

field_path: str — The fieldPath to clear overrides from.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.entities.long_poll_entity_events(...) -> EntityEventResponse

📝 Description

This is a long polling API that will first return all pre-existing data and then return all new data as it becomes available. If you want to start a new polling session then open a request with an empty 'sessionToken' in the request body. The server will return a new session token in the response. If you want to retrieve the next batch of results from an existing polling session then send the session token you received from the server in the request body. If no new data is available then the server will hold the connection open for up to 5 minutes. After the 5 minute timeout period, the server will close the connection with no results and you may resume polling with the same session token. If your session falls behind more than 3x the total number of entities in the environment, the server will terminate your session. In this case you must start a new session by sending a request with an empty session token.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.entities.long_poll_entity_events(
    session_token="sessionToken",
)

⚙️ Parameters

session_token: str — Long-poll session identifier. Leave empty to start a new polling session.

batch_size: typing.Optional[int] — Maximum size of response batch. Defaults to 100. Must be between 1 and 2000 (inclusive).

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.entities.stream_entities(...) -> typing.Iterator[bytes]

📝 Description

Establishes a server-sent events (SSE) connection that streams entity data in real-time. This is a one-way connection from server to client that follows the SSE protocol with text/event-stream content type.

This endpoint enables clients to maintain a real-time view of the common operational picture (COP) by first streaming all pre-existing entities that match filter criteria, then continuously delivering updates as entities are created, modified, or deleted.

The server first sends events with type PREEXISTING for all live entities matching the filter that existed before the stream was open, then streams CREATE events for newly created entities, UPDATE events when existing entities change, and DELETED events when entities are removed. The stream remains open indefinitely unless preExistingOnly is set to true.

Heartbeat messages can be configured to maintain connection health and detect disconnects by setting the heartbeatIntervalMS parameter. These heartbeats help keep the connection alive and allow clients to verify the server is still responsive.

Clients can optimize bandwidth usage by specifying which entity components they need populated using the componentsToInclude parameter. This allows receiving only relevant data instead of complete entities.

The connection automatically recovers from temporary disconnections, resuming the stream where it left off. Unlike polling approaches, this provides real-time updates with minimal latency and reduced server load.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.entities.stream_entities()

⚙️ Parameters

heartbeat_interval_ms: typing.Optional[int] — at what interval to send heartbeat events, defaults to 30s.

pre_existing_only: typing.Optional[bool] — only stream pre-existing entities in the environment and then close the connection, defaults to false.

components_to_include: typing.Optional[typing.List[str]] — list of components to include, leave empty to include all components.

filter: typing.Optional[Statement]

Optional root of a Statement filter expression tree. If provided, only entities matching the filter are streamed. Applied dynamically: an entity that begins matching is delivered as a CREATE, and one that stops matching is delivered as a DELETE. Mirrors the filter on the gRPC StreamEntityComponents endpoint.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

Tasks

client.tasks.create_task(...) -> Task

📝 Description

Creates a new Task in the system with the specified parameters.

This method initiates a new task with a unique ID (either provided or auto-generated), sets the initial task state to STATUS_CREATED, and establishes task ownership. The task can be assigned to a specific agent through the Relations field.

Once created, a task enters the lifecycle workflow and can be tracked, updated, and managed through other Tasks API endpoints.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.create_task()

⚙️ Parameters

task_id: typing.Optional[str]

If non-empty, will set the requested Task ID, otherwise will generate a new random GUID. Will reject if supplied Task ID does not match [A-Za-z0-9_-.]{5,36}.

display_name: typing.Optional[str] — Human readable display name for this Task, should be short (<100 chars).

description: typing.Optional[str] — Longer, free form human readable description of this Task.

specification: typing.Optional[GoogleProtobufAny] — The path for the Protobuf task definition, and the complete task data.

author: typing.Optional[Principal]

relations: typing.Optional[Relations]

Any relationships associated with this Task, such as a parent Task or an assignee this Task is designated to for execution.

is_executed_elsewhere: typing.Optional[bool]

If set, then the service will not trigger execution of this task on an agent. Useful for when ingesting tasks from an external system that is triggering execution of tasks on agents.

initial_entities: typing.Optional[typing.List[TaskEntity]]

Indicates an initial set of entities that can be used to execute an entity aware task. For example, an entity Objective, an entity Keep In Zone, etc.

retry_strategy: typing.Optional[RetryStrategy] — Any retry strategy for task execution or update.

delivery_constraints: typing.Optional[DeliveryConstraints] — Describes scheduling constraints for Lattice when delivering the task to the agent.

execution_constraints: typing.Optional[ExecutionConstraints] — Describes scheduling constraints for the agent executing the task after it has been delivered.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.get_task(...) -> Task

📝 Description

Retrieves a specific Task by its ID, with options to select a particular task version or view.

This method returns detailed information about a task including its current status, specification, relations, and other metadata. The response includes the complete Task object with all associated fields.

By default, the method returns the latest definition version of the task from the manager's perspective.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.get_task(
    task_id="taskId",
)

⚙️ Parameters

task_id: str — ID of task to return

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.update_task_status(...) -> Task

📝 Description

Updates the status of a Task as it progresses through its lifecycle.

This method allows agents or operators to report the current state of a task, which could include changes to task status, and error information.

Each status update increments the task's status_version. When updating status, clients must provide the current version to ensure consistency. The system rejects updates with mismatched versions to prevent race conditions.

Terminal states (STATUS_DONE_OK and STATUS_DONE_NOT_OK) are permanent; once a task reaches these states, no further updates are allowed.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.update_task_status(
    task_id="taskId",
)

⚙️ Parameters

task_id: str — ID of task to update status of

status_version: typing.Optional[int]

The status version of the task to update. This version number increments to indicate the task's current stage in its status lifecycle. Specifically, whenever a task's status updates, the status version increments by one. Any status updates received with a lower status version number than what is known are considered stale and ignored.

new_status: typing.Optional[TaskStatus] — The new status of the task.

author: typing.Optional[Principal]

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.cancel_task(...) -> Task

📝 Description

Cancels a task by marking it for cancellation in the system.

This method initiates task cancellation based on the task's current state:

If the task has not been sent to an agent, it cancels immediately and transitions the task to a terminal state (STATUS_DONE_NOT_OK with ERROR_CODE_CANCELLED).
If the task has already been sent to an agent, the cancellation request is routed to the agent. The agent is then responsible for deciding whether cancellation is possible or not:
- If the task can be cancelled, the agent must use UpdateTaskStatus and set the task status to STATUS_DONE_NOT_OK.
- If the task cannot be cancelled, the agent must use UpdateTaskStatus to attach a TaskError to the task with the error code ERROR_CODE_REJECTED and a message explaining why the task cannot be cancelled.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.cancel_task(
    task_id="taskId",
)

⚙️ Parameters

task_id: str — The ID of task to cancel

author: typing.Optional[Principal] — Who or what is requesting to cancel this task.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.query_tasks(...) -> TaskQueryResults

📝 Description

Searches for Tasks that match specified filtering criteria and returns matching tasks in paginated form.

This method allows filtering tasks based on the following criteria:

Parent-task relationships.
Task status with inclusive or exclusive filtering.
Update time ranges.

Lattice returns the results in pages. When more results are available than can be returned in a single response, a page_token is provided. Use the page_token in subsequent requests to retrieve the next set of results.

By default, this returns the latest task version for each matching task.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.query_tasks()

⚙️ Parameters

page_token: typing.Optional[str] — If set, returns results starting from the given pageToken.

parent_task_id: typing.Optional[str]

If present matches tasks with this parent task ID. This is mutually exclusive with all other query parameters, for example, either provide parent task ID, or any of the remaining parameters, but not both.

status_filter: typing.Optional[TaskQueryStatusFilter]

update_time_range: typing.Optional[TaskQueryUpdateTimeRange] — If provided, only provides tasks updated within the time range.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.stream_tasks(...) -> typing.Iterator[bytes]

📝 Description

Establishes a server streaming connection that delivers task updates in real-time using Server-Sent Events (SSE).

The stream delivers all existing non-terminal tasks when first connected, followed by real-time updates for task creation and status changes. Additionally, heartbeat messages are sent periodically to maintain the connection.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.stream_tasks()

⚙️ Parameters

heartbeat_interval_ms: typing.Optional[int] — The time interval, in milliseconds, that determines the frequency at which to send heartbeat events. Defaults to 30000 (30 seconds).

rate_limit: typing.Optional[int]

The time interval, in milliseconds, after an update for a given task before another one will be sent for the same task. If set, value must be >= 250.

exclude_preexisting_tasks: typing.Optional[bool]

Optional flag to only include tasks created or updated after the stream is initiated, and not any previous preexisting tasks. If unset or false, the stream will include any new tasks and task updates, as well as all preexisting tasks.

task_type: typing.Optional[TaskStreamRequestTaskType] — Optional filter that only returns tasks with specific types. If not provided, all task types will be streamed.

update_start_time: typing.Optional[Timestamp] — If provided, returns tasks which have been updated since the given time.

parent_task_id: typing.Optional[str]

A filter for tasks with a specific parent task ID. Note: This filter is mutually exclusive with all other filter fields (updateStartTime, assignee, statusFilter, taskType). Either provide parentTaskId or any combination of the other filters, but not both.

assignee: typing.Optional[Principal] — A filter for tasks assigned to a specific principal.

status_filter: typing.Optional[TaskStreamRequestStatusFilter] — A filter for task statuses (inclusive or exclusive).

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.listen_as_agent(...) -> AgentRequest

📝 Description

Establishes a server streaming connection that delivers tasks to taskable agents for execution.

This method creates a persistent connection from Tasks API to an agent, allowing the server to push tasks to the agent as they become available. The agent receives a stream of tasks that match its selector criteria (entity IDs).

The stream delivers three types of requests:

ExecuteRequest: Contains a new task for the agent to execute
CancelRequest: Indicates a task should be canceled
CompleteRequest: Indicates a task should be completed

This is the primary method for taskable agents to receive and process tasks in real-time. Agents should maintain this connection and process incoming tasks according to their capabilities.

When an agent receives a task, it should update the task status using the UpdateStatus endpoint to provide progress information back to Tasks API.

This is a long polling API that will block until a new task is ready for delivery. If no new task is available then the server will hold on to your request for up to 5 minutes, after that 5 minute timeout period you will be expected to reinitiate a new request.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.listen_as_agent()

⚙️ Parameters

agent_selector: typing.Optional[EntityIdsSelector] — Selector criteria to determine which Agent Tasks the agent receives

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.stream_as_agent(...) -> typing.Iterator[bytes]

📝 Description

Establishes a server streaming connection that delivers tasks to taskable agents for execution using Server-Sent Events (SSE).

This method creates a connection from the Tasks API to an agent that streams relevant tasks to the listener agent. The agent receives a stream of tasks that match the entities specified by the tasks' selector criteria.

The stream delivers three types of requests:

ExecuteRequest: Contains a new task for the agent to execute
CancelRequest: Indicates a task should be canceled
CompleteRequest: Indicates a task should be completed

Additionally, heartbeat messages are sent periodically to maintain the connection.

This is recommended method for taskable agents to receive and process tasks in real-time. Agents should maintain connection to this stream and process incoming tasks according to their capabilities.

When an agent receives a task, it should update the task status using the UpdateStatus endpoint to provide progress information back to Tasks API.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.stream_as_agent()

⚙️ Parameters

agent_selector: typing.Optional[EntityIdsSelector] — The selector criteria to determine which tasks the agent receives.

heartbeat_interval_ms: typing.Optional[int] — The time interval, in milliseconds, that determines the frequency at which to send heartbeat events. Defaults to 30000 (30 seconds).

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.stream_manual_control_frames(...) -> typing.Iterator[bytes]

📝 Description

Establishes a server streaming connection that delivers manual control frames to agents using server-sent events (SSE).

This endpoint streams manual control frames, for example, for joystick movements, for a specific task to the executing agent. The agent should open this stream before reporting STATUS_EXECUTING to ensure it is ready to receive control input when the operator begins sending frames.

Each frame includes epoch and sequence metadata for handling concurrent control sessions and detecting stale or out-of-order frames. Heartbeat messages are sent periodically to maintain the connection.

The stream terminates automatically when the task reaches a terminal state (STATUS_DONE_OK or STATUS_DONE_NOT_OK).

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.stream_manual_control_frames(
    task_id="taskId",
)

⚙️ Parameters

task_id: str — The ID of the manual control task to receive frames for.

heartbeat_interval_ms: typing.Optional[int] — The time interval, in milliseconds, that determines the frequency at which to send heartbeat events. Defaults to 30000 (30 seconds).

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.tasks.send_manual_control_frame(...) -> ManualControlFrameResponse

📝 Description

Sends a single manual control frame, for example, joystick input, for a manual control task.

Call this endpoint repeatedly to control the asset executing the task. Lattice forwards each frame to the executing agent, which receives it using the stream manual control frames endpoint. The task must be in STATUS_EXECUTING.

The client manages the control session. When it begins controlling, it generates a sessionEpochMicros and includes it in every request, along with a sequence that increases with each frame. A higher epoch replaces the current session, which allows another operator to take over control. A lower epoch is rejected as stale.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.tasks.send_manual_control_frame(
    task_id="taskId",
    session_epoch_micros="sessionEpochMicros",
)

⚙️ Parameters

task_id: str — The ID of the manual control task to send the frame for.

session_epoch_micros: str

Unix timestamp in microseconds identifying the control session. The client generates this value when it begins a control session, and includes the same value in every request for that session. Higher values replace previous sessions. Lower values are rejected as stale.

sequence: typing.Optional[str]

The sequence number of this frame within the current session. The client must increase it with each frame so that agents can detect out-of-order delivery. Reset it to 1 when starting a new session.

specification: typing.Optional[GoogleProtobufAny]

The control instructions for this frame. The format is specific to the task, and is passed through to the agent without being inspected by Lattice.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

Objects

client.objects.list_objects(...) -> ListResponse

📝 Description

Lists objects in your environment. You can define a prefix to list a subset of your objects. If you do not set a prefix, Lattice returns all available objects. By default this endpoint will list local objects only.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.objects.list_objects()

⚙️ Parameters

prefix: typing.Optional[str] — Filters the objects based on the specified prefix path. If no path is specified, all objects are returned.

since_timestamp: typing.Optional[datetime.datetime]

Filters out objects whose last_updated_at is earlier than this timestamp.

last_updated_at records when an object arrived on the node that holds it, so this filter selects objects that arrived since the given time. It is not the time the object was authored: a copy that reaches a node later carries the later arrival time.

page_token: typing.Optional[str] — Opaque cursor for continuing the same list request. Start a new listing without the previous cursor if any query parameter or listing scope changes.

all_objects_in_mesh: typing.Optional[bool] — Lists objects across all environment nodes in a Lattice Mesh. When false or unset, only objects held by the local node are returned.

max_page_size: typing.Optional[int] — Sets the maximum number of items that should be returned on a single page.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.objects.get_object(...) -> typing.Iterator[bytes]

📝 Description

Fetches an object from your environment using the objectPath path parameter.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.objects.get_object(
    object_path="objectPath",
)

⚙️ Parameters

object_path: str — The path of the object to fetch.

accept_encoding: typing.Optional[GetObjectRequestAcceptEncoding] — If set, Lattice will compress the response using the specified compression method. If the header is not defined, or the compression method is set to identity, no compression will be applied to the response.

priority: typing.Optional[str] — Indicates a client's preference for the priority of the response. The value is a structured header as defined in RFC 9218. If you do not set the header, Lattice uses the default priority set for the environment. Incremental delivery directives are not supported and will be ignored.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.objects.upload_object(...) -> PathMetadata

📝 Description

Uploads an object. The object must be 1 GiB or smaller.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.objects.upload_object(
    object_path="objectPath",
)

⚙️ Parameters

object_path: str — Path of the Object that is to be uploaded.

request: typing.Union[bytes, typing.Iterator[bytes], typing.AsyncIterator[bytes]]

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.objects.delete_object(...)

📝 Description

Deletes an object from your environment given the objectPath path parameter.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.objects.delete_object(
    object_path="objectPath",
)

⚙️ Parameters

object_path: str — The path of the object to delete.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.objects.get_object_metadata(...)

📝 Description

Returns metadata for a specified object path. Use this to fetch metadata such as object size (size_bytes), its expiry time (expiry_time), or when it arrived on the node holding it (last_updated_at).

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.objects.get_object_metadata(
    object_path="objectPath",
)

⚙️ Parameters

object_path: str — The path of the object to query.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

oauth

client.oauth.get_token(...) -> GetTokenResponse

📝 Description

Gets a new short-lived token using the specified client credentials

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.oauth.get_token()

⚙️ Parameters

grant_type: typing.Literal — The type of grant being requested

client_id: typing.Optional[str] — The client identifier

client_secret: typing.Optional[str] — The client secret

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

Video

client.video.list_egress_streams(...) -> ListEgressStreamsResponse

📝 Description

Returns a list of active egress stream objects. Results are ordered by egress stream create time. If the egress backend is unreachable, the listed streams might be stale or degraded.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.list_egress_streams()

⚙️ Parameters

page_size: typing.Optional[int]

Desired number of egress streams per page. Defaults to 50 if left blank, and capped at 100. The response may contain fewer than max page size.

page_token: typing.Optional[str]

To retrieve the next page, pass the nextPageToken from the previous response. Leave empty for the first page.

Keep the rest of the request identical between pages, otherwise the server may reject it.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.create_egress_stream(...) -> CreateEgressStreamResponse

📝 Description

Creates an egress stream that publishes a live stream to a downstream consumer. A stream in STREAM_STATUS_UNAVAILABLE is rejected as not-live.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.create_egress_stream()

⚙️ Parameters

ingress_id: typing.Optional[str] — Identifier of the live ingress stream to re-publish as an egress stream.

rtsp: typing.Optional[RtspSettings]

srt: typing.Optional[SrtSettings]

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.get_egress_stream(...) -> GetEgressStreamResponse

📝 Description

Retrieves an egress stream object and its associated metadata.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.get_egress_stream(
    egress_id="egressId",
)

⚙️ Parameters

egress_id: str — Identifier of the egress stream to retrieve.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.delete_egress_stream(...) -> DeleteEgressStreamResponse

📝 Description

Deletes the egress stream for a live stream. Returns NOT_FOUND if no matching active egress stream exists.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.delete_egress_stream(
    egress_id="egressId",
)

⚙️ Parameters

egress_id: str — Identifier of the egress stream to delete.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.list_ingress_streams(...) -> ListIngressStreamsResponse

📝 Description

Returns a list of top level ingress stream objects, including ingress streams and internal Anduril streams. Will only return active streams. Results are ordered by ingress stream create time.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.list_ingress_streams()

⚙️ Parameters

page_size: typing.Optional[int]

Desired number of ingress streams per page. Defaults to 50 if left blank, and capped at 100. The response may contain fewer than requested.

page_token: typing.Optional[str]

To retrieve the next page, pass the nextPageToken from the previous response. Leave empty for the first page.

Keep the rest of the request identical between pages, otherwise the server may reject it.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.create_ingress_stream(...) -> CreateIngressStreamResponse

📝 Description

Creates a video ingress stream, returning metadata that you can use to stream live video to Lattice. Exactly one of rtsp or srt must be set on the request.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.create_ingress_stream()

⚙️ Parameters

ingress_id: typing.Optional[str]

Caller-supplied identifier for the new stream. If omitted, the service generates a GUID. If supplied, a consistent and recognizable pattern is recommended. A common convention is a group prefix (organization, platform, or asset) followed by a specific identifier using underscore or dot as a separator, for example, drone_1, vessel_2, or teamalpha.drone1.

When supplied, an ingressId must be between 4 and 36 characters long and use only printable ASCII characters with no spaces; the 36-character ceiling leaves room for a full GUID. A value outside that length range, or one containing spaces, control characters, or non-ASCII characters, is rejected, as is an ingressId that another ingress stream is already using.

title: typing.Optional[str]

Human-readable title for the stream. A title is required: surrounding whitespace is trimmed before it is stored, and what remains must be non-empty, valid UTF-8, and no longer than 64 characters. Otherwise the request is rejected.

mpeg_ts: typing.Optional[MpegTsSettings]

Receive an MPEG-TS push from the producer. The service allocates a UDP port and returns the URL the producer must push to in CreateIngressStreamResponse.

MPEG-TS ingress is supported only at the edge, in closed networks. When Lattice runs in a cloud environment reached over the public internet, MPEG-TS ingress may be disabled per deployment. When it is disabled, a request that selects mpegTs is rejected with a gRPC error rather than accepted, so callers should be prepared to fall back to RTSP or SRT. An MPEG-TS stream created at the edge can still be listed and inspected on the IngressStream read model even when cloud ingress is disabled.

rtsp: typing.Optional[RtspSettings] — Pull from a caller-supplied RTSP URL.

srt: typing.Optional[SrtSettings]

Receive an SRT push from the producer. The service returns a URL and sessionId in CreateIngressStreamResponse.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.get_ingress_stream(...) -> GetIngressStreamResponse

📝 Description

Retrieves a top level ingress stream object and its associated metadata. This includes ingress streams and internal Anduril streams.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.get_ingress_stream(
    ingress_id="ingressId",
)

⚙️ Parameters

ingress_id: str — Identifier of the ingress stream to retrieve.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

client.video.delete_ingress_stream(...) -> DeleteIngressStreamResponse

📝 Description

Deletes a video ingress stream and transitions the stream to STREAM_STATUS_ARCHIVED. Any egress streams consuming this stream will be stopped automatically.

🔌 Usage

from anduril import Lattice
from anduril.environment import LatticeEnvironment

client = Lattice(
    client_id="<clientId>",
    client_secret="<clientSecret>",
    environment=LatticeEnvironment.DEFAULT,
)

client.video.delete_ingress_stream(
    ingress_id="ingressId",
)

⚙️ Parameters

ingress_id: str — Identifier of the ingress stream to delete.

request_options: typing.Optional[RequestOptions] — Request-specific configuration.

FilesExpand file tree

reference.md

Latest commit

History

reference.md

File metadata and controls

Reference

Entities

📝 Description

🔌 Usage

⚙️ Parameters

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

Tasks

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

Objects

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

📝 Description

🔌 Usage

⚙️ Parameters

oauth

📝 Description

🔌 Usage

⚙️ Parameters

Video

📝 Description

🔌 Usage

⚙️ Parameters