### Install Lattice SDK Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Install the SDK using pip. ```sh pip install anduril-lattice-sdk ``` -------------------------------- ### Initialize Lattice Client and Stream Tasks Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Initializes the Lattice client and starts streaming task updates. This is useful for monitoring task status in real-time. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.stream_tasks() ``` -------------------------------- ### Get Object from Lattice Environment Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Fetches a specific object from your environment using its path. Supports optional compression and priority settings. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.objects.get_object( object_path="objectPath", ) ``` -------------------------------- ### Create a new Task Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Initiates a new task with a unique ID, sets its initial state, and establishes ownership. Tasks can be assigned to specific agents via the Relations field. Use this method to start a task within the system's lifecycle workflow. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.create_task() ``` -------------------------------- ### Get Object Metadata Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Retrieves metadata for a specified object path, such as size or update timestamp. Initialize the Lattice client before use. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.objects.get_object_metadata( object_path="objectPath", ) ``` -------------------------------- ### Get OAuth Token Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Obtains a new short-lived token using client credentials. The Lattice client must be initialized with client ID and secret. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.oauth.get_token() ``` -------------------------------- ### Get Entity Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Retrieve an entity by its ID. This is used to fetch existing entity data from the Entities API. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.entities.get_entity( entity_id="entityId", ) ``` -------------------------------- ### Retrieve a specific Task by ID Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Fetches detailed information about a task, including its status, specification, and metadata. By default, it returns the latest definition version. Use this to get the current state and details of a task. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.get_task( task_id="taskId", ) ``` -------------------------------- ### client.entities.long_poll_entity_events Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## client.entities.long_poll_entity_events ### Description This is a long polling API that returns pre-existing data and then new data as it becomes available. It supports starting new sessions with an empty session token and resuming existing sessions with a provided token. The server may hold connections open for up to 5 minutes if no new data is available. ### Method POST (inferred from usage pattern) ### Endpoint /entities/long_poll_entity_events (inferred) ### Parameters #### Query Parameters - **session_token** (str) - Optional - Long-poll session identifier. Leave empty to start a new polling session. - **batch_size** (typing.Optional[int]) - Optional - Maximum size of response batch. Defaults to 100. Must be between 1 and 2000 (inclusive). #### Request Body None explicitly defined, but session_token is passed in the body in the example. ### Request Example ```python client.entities.long_poll_entity_events( session_token="sessionToken", ) ``` ### Response #### Success Response (200) - **EntityEventResponse** - Contains pre-existing and new entity event data. ``` -------------------------------- ### Override Request Timeout Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Specify a custom timeout in seconds for a particular method call, overriding the client's default timeout. This example overrides the timeout for long polling entity events. ```python # Override timeout for a specific method client.entities.long_poll_entity_events(..., request_options={ "timeout_in_seconds": 1 }) ``` -------------------------------- ### Long Poll Entity Events Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Use this method to retrieve entity events through a long polling mechanism. Start a new session by leaving session_token empty, or continue an existing session by providing the last received token. The server may hold the connection for up to 5 minutes if no new data is available. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.entities.long_poll_entity_events( session_token="sessionToken", ) ``` -------------------------------- ### Listen as Agent for Tasks Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Establishes a connection to listen for tasks as an agent. Requires client initialization with credentials and environment. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.listen_as_agent() ``` -------------------------------- ### Initialize Lattice Client Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Instantiate the synchronous Lattice client with client ID and secret. ```python from anduril import Lattice client = Lattice( client_id="", client_secret="", ) client.entities.long_poll_entity_events( session_token="sessionToken", ) ``` -------------------------------- ### Initialize Async Lattice Client Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Instantiate the asynchronous Lattice client with client ID and secret. ```python import asyncio from anduril import AsyncLattice client = AsyncLattice( client_id="", client_secret="", ) async def main() -> None: await client.entities.long_poll_entity_events( session_token="sessionToken", ) asyncio.run(main()) ``` -------------------------------- ### Configure Lattice Environment Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Initialize the Lattice client with a specific environment. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( environment=LatticeEnvironment.DEFAULT, ) ``` -------------------------------- ### Stream Tasks as Agent Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Establishes a server streaming connection to receive tasks for execution. Recommended for taskable agents to process tasks in real-time. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.stream_as_agent() ``` -------------------------------- ### Configure Custom HTTPX Client with Proxy and Transport Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Integrate a custom httpx.Client instance with the Lattice SDK to manage advanced network configurations like proxies and specific transport settings. ```python import httpx from anduril import Lattice client = Lattice( ..., httpx_client=httpx.Client( proxy="http://my.test.proxy.example.com", transport=httpx.HTTPTransport(local_address="0.0.0.0"), ), ) ``` -------------------------------- ### List Objects in Lattice Environment Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Use this to list all objects in your environment or a subset defined by a prefix. By default, it lists local objects only. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.objects.list_objects() ``` -------------------------------- ### Upload Object Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Use this snippet to upload an object to your environment. Ensure you have initialized the Lattice client with your credentials. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.objects.upload_object() ``` -------------------------------- ### client.objects.get_object Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Fetches an object from your environment using the objectPath path parameter. ```APIDOC ## get_object ### Description Fetches an object from your environment using the objectPath path parameter. ### Method GET ### Endpoint /objects/{object_path} ### Parameters #### Path Parameters - **object_path** (str) - The path of the object to fetch. #### Query Parameters - **accept_encoding** (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** (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** (Optional[RequestOptions]) - Request-specific configuration. ``` -------------------------------- ### Upload Object to Lattice Environment Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Uploads an object to your Lattice environment. The object size must be 1 GiB or smaller. ```python client.objects.upload_object() ``` -------------------------------- ### client.tasks.listen_as_agent Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md This method is used by agents to listen for tasks. It establishes a connection to the Tasks API and receives tasks based on specified criteria. Agents should use this to receive and process tasks in real-time. ```APIDOC ## client.tasks.listen_as_agent ### Description This method is used by agents to listen for tasks. It establishes a connection to the Tasks API and receives tasks based on specified criteria. Agents should use this to receive and process tasks in real-time. ### Method Not specified (likely a streaming or persistent connection) ### Endpoint Not specified ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.listen_as_agent() ``` ### Response #### Success Response This method returns an iterator that yields task-related requests (ExecuteRequest, CancelRequest, CompleteRequest) and heartbeat messages. #### Response Example (Iterator yielding task requests and heartbeats) ``` -------------------------------- ### client.tasks.listen_as_agent Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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). ```APIDOC ## client.tasks.listen_as_agent ### 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. ### Method STREAM ### Endpoint N/A (SDK method) ### Parameters (No specific parameters are documented for this method in the provided text, beyond the implicit connection establishment and task selection criteria.) ``` -------------------------------- ### Stream Manual Control Frames for Task Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Establishes a server streaming connection to receive manual control frames for a specific task. The agent should open this stream before reporting STATUS_EXECUTING. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.stream_manual_control_frames( task_id="taskId", ) ``` -------------------------------- ### Stream Entities in Real-Time Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Establishes a server-sent events (SSE) connection to stream entity data in real-time. This method provides continuous updates for entity creation, modification, and deletion, and can automatically recover from disconnections. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.entities.stream_entities() ``` -------------------------------- ### client.objects.list_objects Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## list_objects ### 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. ### Method GET ### Endpoint /objects ### Parameters #### Query Parameters - **prefix** (Optional[str]) - Filters the objects based on the specified prefix path. If no path is specified, all objects are returned. - **since_timestamp** (Optional[datetime.datetime]) - Sets the age for the oldest objects to query across the environment. - **page_token** (Optional[str]) - Base64 and URL-encoded cursor returned by the service to continue paging. - **all_objects_in_mesh** (Optional[bool]) - Lists objects across all environment nodes in a Lattice Mesh. - **max_page_size** (Optional[int]) - Sets the maximum number of items that should be returned on a single page. - **request_options** (Optional[RequestOptions]) - Request-specific configuration. ``` -------------------------------- ### Configure Client Timeout Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Initialize the Lattice client with a default timeout value. This sets the overall timeout for all operations unless overridden. ```python from anduril import Lattice client = Lattice(..., timeout=20.0) ``` -------------------------------- ### client.objects.upload_object Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Uploads an object to your environment. Requires the object's path and the data to be uploaded. Optional request options can be provided for configuration. ```APIDOC ## client.objects.upload_object ### Description Uploads an object to your environment. ### Parameters #### Path Parameters - **object_path** (str) - Required - Path of the Object that is to be uploaded. #### Request Body - **request** (bytes | Iterator[bytes] | AsyncIterator[bytes]) - Required - The data to upload. #### Request Options - **request_options** (Optional[RequestOptions]) - Optional - Request-specific configuration. ### Request Example ```python client.objects.upload_object( object_path="objectPath", request=b"data", ) ``` ``` -------------------------------- ### OAuth Client Credentials Flow Authentication Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Initialize the client using client ID and secret for automatic OAuth token management. ```python # Option 2: OAuth client credentials flow (automatic token management) client = Lattice( ..., client_id="your-client-id", client_secret="your-client-secret", ) ``` -------------------------------- ### client.tasks.stream_as_agent Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Establishes a server streaming connection that delivers tasks to taskable agents for execution using Server-Sent Events (SSE). Agents should maintain connection to this stream and process incoming tasks. ```APIDOC ## client.tasks.stream_as_agent ### 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 ExecuteRequest, CancelRequest, and CompleteRequest, along with periodic heartbeat messages. This is the recommended method for taskable agents to receive and process tasks in real-time. ### Method Not specified (likely a streaming or persistent connection) ### Endpoint Not specified ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.stream_as_agent() ``` ### Response #### Success Response (200) Returns an iterator yielding bytes, representing the Server-Sent Events stream of tasks and control messages. ``` -------------------------------- ### client.tasks.stream_manual_control_frames Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Establishes a server streaming connection that delivers manual control frames to agents using server-sent events (SSE). This endpoint streams manual control frames for a specific task to the executing agent. ```APIDOC ## client.tasks.stream_manual_control_frames ### 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. ### Method Not specified (likely a streaming or persistent connection) ### Endpoint Not specified ### Parameters #### Path Parameters * **task_id** (str) - Required - The ID of the manual control task to receive frames for. #### Query Parameters * **heartbeat_interval_ms** (typing.Optional[int]) - Optional - The time interval, in milliseconds, that determines the frequency at which to send heartbeat events. Defaults to 30000 (30 seconds). #### Request Body None ### Request Example ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.stream_manual_control_frames( task_id="taskId", ) ``` ### Response #### Success Response (200) Returns an iterator yielding bytes, representing the Server-Sent Events stream of manual control frames. ``` -------------------------------- ### Direct Bearer Token Authentication Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Initialize the client using a pre-generated bearer token, bypassing the OAuth flow. ```python from anduril import Lattice # Option 1: Direct bearer token (bypass OAuth flow) client = Lattice( ..., token="my-pre-generated-bearer-token", ) ``` -------------------------------- ### client.tasks.create_task Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## client.tasks.create_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. ### Parameters #### Path Parameters - **task_id** (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** (Optional[str]) - Human readable display name for this Task, should be short (<100 chars). - **description** (Optional[str]) - Longer, free form human readable description of this Task. - **specification** (Optional[GoogleProtobufAny]) - The path for the Protobuf task definition, and the complete task data. - **author** (Optional[Principal]) - - **relations** (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** (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** (Optional[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. - **request_options** (Optional[RequestOptions]) - Request-specific configuration. ``` -------------------------------- ### Paginate Object List Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Iterate through paginated results using SyncPager. ```python from anduril import Lattice client = Lattice( client_id="", client_secret="", ) client.objects.list_objects() ``` -------------------------------- ### Query Tasks Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Searches for tasks based on specified criteria and returns results in a paginated format. Use this to find tasks matching specific filters. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.query_tasks() ``` -------------------------------- ### client.tasks.stream_tasks Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. Heartbeat messages are sent periodically to maintain the connection. ```APIDOC ## client.tasks.stream_tasks ### 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. ### Method STREAM ### Endpoint N/A (SDK method) ### Parameters #### Query Parameters - **heartbeat_interval_ms** (int) - Optional - The time interval, in milliseconds, that determines the frequency at which to send heartbeat events. Defaults to 30000 (30 seconds). - **rate_limit** (int) - Optional - 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** (bool) - Optional - 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** (TaskStreamRequestTaskType) - Optional - Optional filter that only returns tasks with specific types. If not provided, all task types will be streamed. - **update_start_time** (Timestamp) - Optional - If provided, returns tasks which have been updated since the given time. - **parent_task_id** (str) - Optional - 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** (Principal) - Optional - A filter for tasks assigned to a specific principal. - **status_filter** (TaskStreamRequestStatusFilter) - Optional - A filter for task statuses (inclusive or exclusive). - **request_options** (RequestOptions) - Optional - Request-specific configuration. ``` -------------------------------- ### client.objects.upload_object Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Uploads an object. The object must be 1 GiB or smaller. ```APIDOC ## upload_object ### Description Uploads an object. The object must be 1 GiB or smaller. ### Method POST ### Endpoint /objects ### Parameters #### Request Body - **file** (file) - The object to upload. - **request_options** (Optional[RequestOptions]) - Request-specific configuration. ``` -------------------------------- ### client.oauth.get_token Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Obtains a new short-lived token using client credentials. Supports different grant types and optional client ID/secret. ```APIDOC ## client.oauth.get_token ### Description Gets a new short-lived token using the specified client credentials. ### Parameters #### Query Parameters - **grant_type** (Literal) - Required - The type of grant being requested. - **client_id** (Optional[str]) - Optional - The client identifier. - **client_secret** (Optional[str]) - Optional - The client secret. #### Request Options - **request_options** (Optional[RequestOptions]) - Optional - Request-specific configuration. ### Request Example ```python client.oauth.get_token( grant_type="client_credentials", ) ``` ``` -------------------------------- ### client.tasks.query_tasks Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Searches for Tasks that match specified filtering criteria and returns matching tasks in paginated form. This method allows filtering tasks based on multiple criteria including parent task relationships, task status, update time ranges, task view, task assignee, and task type. Results are returned in pages. By default, this returns the latest task version for each matching task from the manager's perspective. ```APIDOC ## client.tasks.query_tasks ### Description Searches for Tasks that match specified filtering criteria and returns matching tasks in paginated form. This method allows filtering tasks based on multiple criteria including parent task relationships, task status, update time ranges, task view, task assignee, and task type. Results are returned in pages. By default, this returns the latest task version for each matching task from the manager's perspective. ### Method Not specified (SDK method) ### Endpoint None ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters None specified in the source text. ``` -------------------------------- ### client.entities.publish_entity Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Publishes an entity for ingest into the Entities API. Entities created are owned by the originator and cannot be edited or deleted by other sources. The server validates entities upon API call. An entity ID must be provided; if it doesn't exist, it will be created, otherwise it will be updated if the new entity's sourceUpdateTime is greater than the existing one. ```APIDOC ## publish_entity ### Description Publishes 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. ### Method Not specified, assumed to be a client method call. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **request** (Entity) - Required - The entity to publish. - **request_options** (Optional[RequestOptions]) - Optional - Request-specific configuration. ### Request Example ```python client.entities.publish_entity(request=entity_object) ``` ### Response #### Success Response Returns the published or updated Entity. #### Response Example ```json { "example": "Entity object" } ``` ``` -------------------------------- ### client.tasks.cancel_task Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## client.tasks.cancel_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. ### Method Not specified (SDK method) ### Endpoint None ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **task_id** (str) - Required - The ID of task to cancel - **author** (typing.Optional[Principal]) - Optional - Who or what is requesting to cancel this task. - **request_options** (typing.Optional[RequestOptions]) - Optional - Request-specific configuration. ``` -------------------------------- ### Publish Entity Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Publish an entity for ingest into the Entities API. The entity ID must be provided; if it does not exist, it will be created, otherwise it will be updated if the new entity's provenance is newer. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.entities.publish_entity() ``` -------------------------------- ### List Deleted Objects in Lattice Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Retrieves paginated records of deleted objects on the local node. Useful for diagnosing object synchronization issues. Each record includes the path and checksum of the suppressed object. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.objects.list_deleted_objects() ``` -------------------------------- ### Delete Object Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Deletes an object from your environment using its path. The client must be initialized prior to calling this function. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.objects.delete_object( object_path="objectPath", ) ``` -------------------------------- ### Access Raw Response Data Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Use the `.with_raw_response` property to access raw response data including headers and status code. ```python from anduril import Lattice client = Lattice(...) response = client.entities.with_raw_response.long_poll_entity_events(...) print(response.headers) # access the response headers print(response.status_code) # access the response status code print(response.data) # access the underlying object ``` -------------------------------- ### Stream Entity Responses Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Use the streaming API to receive responses as a generator. ```python from anduril import Lattice client = Lattice( client_id="", client_secret="", ) client.entities.stream_entities() ``` -------------------------------- ### client.objects.list_deleted_objects Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Returns paginated records of force-distributed objects deleted on the local node. Useful for operators diagnosing why an object visible on one node is missing on another. Each record identifies the exact (path, checksum) pair suppressed from re-sync by the distribution manager. Node-scoped: each node returns only its own records. ```APIDOC ## list_deleted_objects ### Description Returns paginated records of force-distributed objects deleted on the local node. Useful for operators diagnosing why an object visible on one node is missing on another. Each record identifies the exact (path, checksum) pair suppressed from re-sync by the distribution manager. Node-scoped: each node returns only its own records. ### Method GET ### Endpoint /objects/deleted ### Parameters #### Query Parameters - **page_token** (Optional[str]) - Opaque cursor returned by a prior response to continue paging. - **max_page_size** (Optional[int]) - Maximum number of records to return in a single response. Server enforces an upper bound. - **request_options** (Optional[RequestOptions]) - Request-specific configuration. ``` -------------------------------- ### client.objects.get_object_metadata Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Retrieves metadata for a specified object path, including size, expiry time, and last updated timestamp. Optional request options can be provided. ```APIDOC ## 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 its latest update timestamp (last_updated_at). ### Parameters #### Path Parameters - **object_path** (str) - Required - The path of the object to query. #### Request Options - **request_options** (Optional[RequestOptions]) - Optional - Request-specific configuration. ### Request Example ```python client.objects.get_object_metadata( object_path="objectPath", ) ``` ``` -------------------------------- ### client.objects.delete_object Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Deletes an object from your environment given its path. Optional request options can be provided for configuration. ```APIDOC ## client.objects.delete_object ### Description Deletes an object from your environment given the objectPath path parameter. ### Parameters #### Path Parameters - **object_path** (str) - Required - The path of the object to delete. #### Request Options - **request_options** (Optional[RequestOptions]) - Optional - Request-specific configuration. ### Request Example ```python client.objects.delete_object( object_path="objectPath", ) ``` ``` -------------------------------- ### client.tasks.get_task Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## client.tasks.get_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. ### Parameters #### Path Parameters - **task_id** (str) - ID of task to return - **request_options** (Optional[RequestOptions]) - Request-specific configuration. ``` -------------------------------- ### Cancel Task Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Initiates the cancellation of a task. If the task is sent to an agent, the cancellation request is routed to the agent for final decision. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.cancel_task( task_id="taskId", ) ``` -------------------------------- ### Iterate Through Paginated Pages Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Access typed responses and individual items within each page of a paginated result. ```python # You can also iterate through pages and access the typed response per page pager = client.objects.list_objects(...) for page in pager.iter_pages(): print(page.response) # access the typed response for each page for item in page: print(item) ``` -------------------------------- ### client.tasks.update_task_status Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## client.tasks.update_task_status ### 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. ### Method Not specified (SDK method) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **task_id** (str) - Required - ID of task to update status of - **status_version** (typing.Optional[int]) - Optional - 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]) - Optional - The new status of the task. - **author** (typing.Optional[Principal]) - Optional - **request_options** (typing.Optional[RequestOptions]) - Optional - Request-specific configuration. ``` -------------------------------- ### client.entities.stream_entities Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md 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. ```APIDOC ## client.entities.stream_entities ### Description Establishes a server-sent events (SSE) connection that streams entity data in real-time. It first streams pre-existing entities matching filter criteria, then continuously delivers updates (CREATE, UPDATE, DELETE events). The stream remains open indefinitely by default and supports heartbeat messages for connection health. ### Method GET (inferred from SSE) ### Endpoint /entities/stream (inferred) ### Parameters #### Query Parameters - **preExistingOnly** (bool) - Optional - If true, only streams pre-existing entities and closes the connection. - **heartbeatIntervalMS** (int) - Optional - Configures heartbeat messages to maintain connection health. - **componentsToInclude** (list[str]) - Optional - Specifies which entity components to populate, optimizing bandwidth. ### Request Example ```python client.entities.stream_entities() ``` ### Response #### Success Response (200) - **typing.Iterator[bytes]** - An iterator yielding entity data in SSE format. ``` -------------------------------- ### client.entities.get_entity Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Retrieves a specific entity by its ID. This method allows fetching entity details that have been published or managed through the Entities API. ```APIDOC ## get_entity ### Description Retrieves a specific entity by its unique identifier. ### Method Not specified, assumed to be a client method call. ### Parameters #### Path Parameters None #### Query Parameters - **entity_id** (str) - Required - ID of the entity to return. #### Request Body None ### Request Example ```python client.entities.get_entity(entity_id="entityId") ``` ### Response #### Success Response Returns the requested Entity. #### Response Example ```json { "example": "Entity object" } ``` ``` -------------------------------- ### Update Task Status Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Updates the status of a task, ensuring consistency with versioning. Use this to report progress or completion of a task. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.tasks.update_task_status( task_id="taskId", ) ``` -------------------------------- ### Configure Max Retries for Entity Events Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Set the maximum number of retry attempts for a specific request, such as long polling entity events. The default retry limit is 2. ```python client.entities.long_poll_entity_events(..., request_options={ "max_retries": 1 }) ``` -------------------------------- ### Handle API Errors Source: https://github.com/anduril/lattice-sdk-python/blob/master/README.md Catch ApiError exceptions to handle non-success status codes from the API. ```python from anduril.core.api_error import ApiError try: client.entities.long_poll_entity_events(...) except ApiError as e: print(e.status_code) print(e.body) ``` -------------------------------- ### Remove Entity Override Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Clear the override value for a specified field path on an entity. This operation reverts the field to its non-overridden state. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.entities.remove_entity_override( entity_id="entityId", field_path="mil_view.disposition", ) ``` -------------------------------- ### client.entities.override_entity Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Overrides a specific field within an entity. This is used to update fields that are marked as overridable. Overrides are applied eventually consistently, with the last writer winning in case of concurrent operations. ```APIDOC ## override_entity ### Description Overrides a specific field within an entity. Only fields marked with overridable 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. ### Method Not specified, assumed to be a client method call. ### Parameters #### Path Parameters None #### Query Parameters - **entity_id** (str) - Required - The unique ID of the entity to override. - **field_path** (str) - Required - fieldPath to override. #### Request Body - **entity** (Optional[Entity]) - Optional - The entity containing the overridden fields. The service will extract the overridable fields from the object and ignore all other fields. - **provenance** (Optional[Provenance]) - Optional - Additional information about the source of the override. - **request_options** (Optional[RequestOptions]) - Optional - Request-specific configuration. ### Request Example ```python client.entities.override_entity(entity_id="entityId", field_path="mil_view.disposition", entity=override_entity_object) ``` ### Response #### Success Response Returns the updated Entity. #### Response Example ```json { "example": "Entity object" } ``` ``` -------------------------------- ### Override Entity Field Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Override a specific field of an entity. Only fields marked as overridable can be modified. Overrides are applied eventually consistently, with the last writer winning in case of concurrent updates. ```python from anduril import Lattice from anduril.environment import LatticeEnvironment client = Lattice( client_id="", client_secret="", environment=LatticeEnvironment.DEFAULT, ) client.entities.override_entity( entity_id="entityId", field_path="mil_view.disposition", ) ``` -------------------------------- ### client.entities.remove_entity_override Source: https://github.com/anduril/lattice-sdk-python/blob/master/reference.md Clears the override value from a specified field path on an entity. This operation reverts the field to its default or previously set value, removing any active override. ```APIDOC ## remove_entity_override ### Description This operation clears the override value from the specified field path on the entity. ### Method Not specified, assumed to be a client method call. ### Parameters #### Path Parameters None #### Query Parameters - **entity_id** (str) - Required - The unique ID of the entity to remove the override from. - **field_path** (str) - Required - fieldPath to remove the override from. #### Request Body None ### Request Example ```python client.entities.remove_entity_override(entity_id="entityId", field_path="mil_view.disposition") ``` ### Response #### Success Response Returns the Entity with the override removed. #### Response Example ```json { "example": "Entity object" } ``` ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.