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

# Watch entities

> Monitoring entities in Lattice using the Lattice SDK

This shows how to use the SDK to fetch entities from Lattice and
stream real-time information about entity components.

Complete the steps to learn how to use the following APIs:

* [`GetEntity`](/reference/rest/entities/get-entity) -- Retrieves a single entity from Lattice.
* [`StreamEntities`](/reference/rest/entities/stream-entities) -- Establishes a persistent connection to stream entity events.

## Before you begin

* To configure your app to watch entities, [set up the Lattice SDK](/guides/getting-started/set-up).
* Learn about required components and various [entity shapes](/guides/entities/overview) in Lattice.

#### gRPC authentication

These gRPC examples authenticate with a static environment token attached as request metadata.
If you are using OAuth 2.0 client credentials instead, [set up the token refresh module](/guides/getting-started/authenticate#client-credentials) before running the examples on this page.

## Get an entity

Get details of a specific entity using [`entity_ID`](/reference/rest/entities/publish-entity#request.body.entityId)
and the [`GetEntity`](/reference/rest/entities/get-entity) API:

#### Get the entity ID

Open the Lattice UI and find the entity in the Entity Explorer. Copy its ID from the **Entity ID** column:

![Shows the Entity Explorer in the Lattice Developer Console.](/_fern-img/72c242c7f0277f5a308bd2efb784cfcd9675f4dab9d1790bd9b195c6df74c7d0.webp)

#### Get the entity object

Use the `GetEntity` API action to retrieve a single entity object from Lattice.
Replace `$ENTITY_ID` in the following example with the entity ID you copied in the
previous step:

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

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

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

func main() {
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

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

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

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

	for {
		ctx := context.Background()
		entity, err := LatticeClient.Entities.GetEntity(ctx, &Lattice.GetEntityRequest{EntityID: "<ENTITY_ID>"})
		if err != nil {
			fmt.Printf("Error fetching entity: %v\n", err)
		} else {
			fmt.Printf("Asset name     | %s\n", *entity.GetAliases().GetName())
			fmt.Printf("Asset location | %f, %f\n", *entity.GetLocation().GetPosition().GetLatitudeDegrees(),
				*entity.GetLocation().GetPosition().GetLongitudeDegrees())
		}
		time.Sleep(5 * time.Second)
	}
}

```

```java title={"Java (REST)"} startLine={28} maxLines={20}
package org.example;
import com.anduril.Lattice;
import com.anduril.types.Entity;
import java.time.LocalDateTime;

public class GetEntityEnvToken {
    private static final String ENTITY_ID = "<ENTITY_ID>";

    public static void main(String[] args) {
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        // Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");

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

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

            while (true) {
                try {
                    Entity entity = client.entities().getEntity(ENTITY_ID);
                    System.out.println("Timestamp: " + LocalDateTime.now());
                    System.out.println("Asset name: " + entity.getAliases());
                    if (entity.getLocation() != null) {
                        System.out.println("Location: " + entity.getLocation());
                    }
                    System.out.println("---------------");
                    Thread.sleep(5000);
                } catch (Exception e) {
                    System.err.println("Error: " + e.getMessage());
                    Thread.sleep(10000);
                }
            }
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }
}

```

```py title={"Python (REST)"} startLine={21} maxLines={20}
from anduril import Lattice
import asyncio
import os
import sys

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

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

client = Lattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: str(environment_token),
    headers={ "anduril-sandbox-authorization": f"Bearer {sandboxes_token}" }
)

async def app(entity_id):
    try:
        while(True):
            entity = client.entities.get_entity(entity_id=entity_id)
            if entity.aliases:
                print(f"Asset name     | {entity.aliases.name}")
            if entity.location and entity.location.position:
                print(f"Asset location | {entity.location.position.latitude_degrees}, {entity.location.position.longitude_degrees}")
            await asyncio.sleep(5)
    except (asyncio.CancelledError, KeyboardInterrupt):
        print(">>>Exiting...")
    except Exception as error:
        print(f"Exception: {error}")

if __name__ == "__main__":
    asyncio.run(app("<ENTITY_ID>"))

```

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

const latticeEndpoint = process.env.LATTICE_ENDPOINT;
const environmentToken = process.env.ENVIRONMENT_TOKEN;
// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}
const client = new LatticeClient({
    baseUrl: `https://${latticeEndpoint}`,
    token: environmentToken,
    headers: { "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}` }
});

async function App(entityId: string) {
    try {
        const entity = await client.entities.getEntity({ entityId });
        if (entity && Object.keys(entity).length > 0) {
            console.log(`Asset name     | ${entity.aliases?.name}`);
            console.log(`Asset location | ${entity.location?.position?.latitudeDegrees}, ${entity.location?.position?.longitudeDegrees}`);
        } else {
            console.log('Entity object is empty.');
        }
    } catch (error) {
        console.log(`Encountered the following error while fetching entity: ${error}`);
    }
}

(async function runIndefinitely() {
    while (true) {
        await App("<ENTITY_ID>");
        await new Promise(resolve => setTimeout(resolve, 1000));
    }
})();

```

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

import (
	"context"
	"log"
	"os"
	"time"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

type BearerTokenAuth struct {
	Token          string
	SandboxesToken string
}

func (b *BearerTokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
		"authorization":                 "Bearer " + b.Token,
		"anduril-sandbox-authorization": "Bearer " + b.SandboxesToken,
	}, nil
}

func (b *BearerTokenAuth) RequireTransportSecurity() bool {
	return true
}

func main() {
	ctx := context.Background()

	environmentToken := os.Getenv("ENVIRONMENT_TOKEN")
	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes.
	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}
	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

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

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

	client := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	// Periodically get the latest entity state
	for {
		entity, err := client.GetEntity(ctx, &entitymanagerv1.GetEntityRequest{
			EntityId: "Demo-Sim-Asset1",
		})
		if err != nil {
			log.Fatalf("Error getting entity: %v", err)
		}

		// Process each entity.
		log.Printf("Fetching entity at location: %f %f", entity.GetEntity().GetLocation().GetPosition().LatitudeDegrees, entity.GetEntity().GetLocation().GetPosition().LongitudeDegrees)

		time.Sleep(5 * time.Second)
	}
}

```

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

import os
import sys
import time
from connectrpc.interceptor import MetadataInterceptorSync
from connectrpc.protocol import ProtocolType
from connectrpc.request import RequestContext
from pyqwest import SyncClient, SyncHTTPTransport

from anduril.entitymanager.v1.entity_manager_api_pub_pb import GetEntityRequest
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync


class BearerTokenInterceptor(MetadataInterceptorSync):
    """Connect interceptor that adds bearer token headers to requests."""

    def __init__(self, token: str, sandboxes_token: str):
        self.token = token
        self.sandboxes_token = sandboxes_token

    def on_start_sync(self, ctx: RequestContext) -> None:
        ctx.request_headers["authorization"] = f"Bearer {self.token}"
        ctx.request_headers["anduril-sandbox-authorization"] = (
            f"Bearer {self.sandboxes_token}"
        )
        return None

    def on_end_sync(self, token, ctx: RequestContext, error) -> None:
        return


def main():
    # Load environment variables
    environment_token = os.getenv("ENVIRONMENT_TOKEN")
    lattice_endpoint = os.getenv("LATTICE_ENDPOINT")
    # Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
    sandboxes_token = os.getenv("SANDBOXES_TOKEN")

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

    # Create the EntityManager API client over Connect gRPC with a bearer token
    # interceptor. pyqwest trusts no certificates by default, so enable the
    # system certificate pool for TLS.
    http_client = SyncClient(transport=SyncHTTPTransport(tls_include_system_certs=True))
    client = EntityManagerAPIClientSync(
        f"https://{lattice_endpoint}",
        http_client=http_client,
        interceptors=[BearerTokenInterceptor(environment_token, sandboxes_token)],
        protocol=ProtocolType.GRPC,
    )

    # Periodically get the latest entity state
    while True:
        try:
            request = GetEntityRequest(entity_id="<ENTITY_ID>")

            # Call the GetEntity RPC
            response = client.get_entity(request)

            # Log entity location if available
            if (
                response.entity
                and response.entity.location
                and response.entity.location.position
            ):
                position = response.entity.location.position
                print(
                    f"Fetching entity at location: "
                    f"{position.latitude_degrees} {position.longitude_degrees}"
                )

        except Exception as error:
            print(f"Error getting entity: {error}", file=sys.stderr)

        time.sleep(5)


if __name__ == "__main__":
    main()

```

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

use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::GetEntityRequest;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;
use std::time::Duration;

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

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

    // Parse tokens into metadata values
    let auth_header: MetadataValue<_> = format!("Bearer {}", environment_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

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

    // Create EntityManager API client with authentication interceptor
    let mut client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    println!("Starting EntityManager API client...");
    println!("Connected to: {}", lattice_endpoint);
    println!();

    // Periodically get the latest entity state
    loop {
        let request = Request::new(GetEntityRequest {
            entity_id: "<ENTITY_ID>".to_string(),
        });

        match client.get_entity(request).await {
            Ok(response) => {
                // Log entity location if available
                if let Some(entity) = response.get_ref().entity.as_ref() {
                    if let Some(location) = entity.location.as_ref() {
                        if let Some(position) = location.position.as_ref() {
                            println!(
                                "Fetching entity at location: {} {}",
                                position.latitude_degrees, position.longitude_degrees
                            );
                        }
                    }
                }
            }
            Err(e) => {
                eprintln!("Error getting entity: {}", e);
            }
        }

        // Wait 5 seconds before next request
        tokio::time::sleep(Duration::from_secs(5)).await;
    }
}

```

#### Verify the response

If successful, you'll see the entity's`aliases.name` and its
real-time location logged in the console:

```bash maxLines=3
Asset name     | Demo-Sim-Asset1
Asset location | 37.7749, -122.4194
```

## Stream entities

The [`StreamEntities`](/reference/rest/entities/stream-entities) API establishes a
persistent connection to stream entity events as they occur. The stream sends two types
of events: [`entity`](/reference/rest/entities/stream-entities#response.body.entity)
and [`heartbeat`](/reference/rest/entities/stream-entities#response.body.heartbeat).

Use the following optional parameters to control
the frequency and type of data fetched from your environment:
[`heartbeatIntervalMS`](/reference/rest/entities/stream-entities#request.body.heartbeatIntervalMS),
[`preExistingOnly`](/reference/rest/entities/stream-entities#request.body.preExistingOnly),
and [`componentsToInclude`](/reference/rest/entities/stream-entities#request.body.componentsToInclude).

To stream entities from Lattice, do the following:

#### Stream all components

To get stream of all entity components from your environment, including
new entities as they are updated, use the default options.

By default, the `preExistingOnly` option is set to `false`, instructing
Lattice to establish a persistent, real-time connection with the client:

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

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"

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

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

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

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatal("Required environment variables not set.")
	}

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

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

	// Create the entity stream.
	stream, err := latticeClient.Entities.StreamEntities(ctx, &Lattice.EntityStreamRequest{
		PreExistingOnly: Lattice.Bool(false),
	})
	if err != nil {
		log.Fatalf("Failed to create entity stream: %v", err)
	}
	defer stream.Close()

	for {
		select {
		case <-ctx.Done():
			log.Printf("Context canceled: %v", ctx.Err())
			return
		default:
			// Continue processing
		}
		message, err := stream.Recv()

		if errors.Is(err, io.EOF) {
			log.Println("Stream completed successfully.")
			return
		}

		if err != nil {
			log.Printf("Error receiving message: %v", err)
			continue
		}

		// Process the message based on its type
		switch message.Event {
		case "heartbeat":
			timestamp := *message.Heartbeat.Timestamp
			log.Printf("Heartbeat: %s", timestamp)
		case "entity":
			log.Printf("Entity: %s", *message.Entity.Entity.EntityID)
		default:
			log.Printf("Unknown event type: %s", message.Event)
		}
	}
}

```

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

import com.anduril.AsyncLattice;
import com.anduril.resources.entities.requests.EntityStreamRequest;
import com.anduril.resources.entities.types.StreamEntitiesResponse;
import com.anduril.types.EntityStreamEvent;



/**
 * Example application that streams all entities using the Lattice SDK.
 */
public class StreamAllEntities {
    
    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");
        
        // Check required variables
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;
        
        try {
            AsyncLattice client = AsyncLattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();
            
            System.out.println("Starting entity stream...");
            try {
                EntityStreamRequest request = EntityStreamRequest
                    .builder()
                    .preExistingOnly(false)
                    .build();

                client.entities().streamEntities(request)
                    .thenAccept(responses -> {
                        for (StreamEntitiesResponse response : responses) {
                            // If the event is of type heartbeat, log the timestamp.
                            if (response.getHeartbeat() != null && response.getHeartbeat().isPresent()) {
                                System.out.println("Heartbeat received: " + response.getHeartbeat().get().getTimestamp().get());
                            } 
                            // If the event is of type entity, log the entity ID.
                            else if (response.getEntity() != null && response.getEntity().isPresent()) {
                                EntityStreamEvent event = response.getEntity().get();
                                System.out.println("Entity: " + event.getEntity().get().getEntityId().get());
                            }
                        }
                    })
                    .exceptionally(ex -> {
                        System.err.println("Exception while streaming entities: " + ex.getMessage());
                        return null;
                    })
                    .join();
            } catch (Exception e) {
                System.err.println("Error in streaming loop: " + e.getMessage());
            }
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }
}
```

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

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

# Remove sandboxes_token from the following statements if you are not developing on Sandboxes.
sandboxes_token = os.getenv('SANDBOXES_TOKEN')
if not environment_token or not lattice_endpoint or not sandboxes_token:
    print("Missing environment variables.")
    sys.exit(1)
    
client = AsyncLattice(
    base_url=f"https://{lattice_endpoint}",
    token=lambda: str(environment_token),
    # Remove the following header if you are not developing on Sandboxes.
    headers={ "anduril-sandbox-authorization": f"Bearer {sandboxes_token}" }
)

async def app():
    try:
        event_stream = client.entities.stream_entities(pre_existing_only=False)
        async for event in event_stream:
            # If the event is of type heartbeat, log the timestamp.
            if event.event == "heartbeat":
                print(f'Heartbeat: {event.timestamp}')
            # If the event is of type entity, log the entity ID.
            elif event.entity:
                print(f'Entity: {event.entity.entity_id}')
    except asyncio.CancelledError:
        print("Streaming cancelled...")
    except Exception as error:
        print(f"Exception: {error}")
        
if __name__ == "__main__":
    asyncio.run(app())

```

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

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

// Remove sandboxesToken from the following statement if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

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

async function App() {
    try {
        // Start streaming entities
        const eventStream = await client.entities.streamEntities({ preExistingOnly: false });
        
        // Process the stream
        for await (const event of eventStream) {
            // Check if it's a heartbeat event.
            if (event.event === "heartbeat") {
                console.log(`Heartbeat received: ${event.timestamp}`);
                continue;
            }
            // Process entity event.
            const entity = event.entity;
            if (entity) {
                console.log(`Entity: ${entity.entityId}`);
            }
        }
    } catch (error) {
        console.log(`Exception while streaming entities: ${error}`);
    }
}

App();
```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func main() {
	ctx := context.Background()

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

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}
	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

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

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

	client := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	stream, err := client.StreamEntityComponents(ctx, &entitymanagerv1.StreamEntityComponentsRequest{
		IncludeAllComponents: true,
	})
	if err != nil {
		log.Fatalf("Error creating stream: %v", err)
	}

	log.Println("Starting to receive stream data...")
	for {
		response, err := stream.Recv()
		if err == io.EOF {
			log.Println("End of stream reached")
			break
		}
		if err != nil {
			log.Fatalf("Error receiving stream data: %v", err)
		}

		// Process each entity.
		log.Printf("Streaming entity ID: %s", response.GetEntityEvent().GetEntity().EntityId)
	}
}

```

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

import os
import sys
from bearer_auth import create_client

from anduril.entitymanager.v1.entity_manager_api_pub_pb import (
    StreamEntityComponentsRequest,
)
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync


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

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

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

    request = StreamEntityComponentsRequest(include_all_components=True)

    try:
        stream = client.stream_entity_components(request)
    except Exception as error:
        print(f"Error creating stream: {error}", file=sys.stderr)
        sys.exit(1)

    print("Starting to receive stream data...")

    try:
        for response in stream:
            # Heartbeat messages carry no entity_event, so skip them.
            if not response.entity_event or not response.entity_event.entity:
                continue
            entity = response.entity_event.entity
            print(f"Streaming entity ID: {entity.entity_id}")
        print("End of stream reached")
    except Exception as error:
        print(f"Error receiving stream data: {error}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

```

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

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::StreamEntityComponentsRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

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

    // Set up authentication handler
    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

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

    // Fetch a fresh access token and build metadata values
    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    // Create EntityManager API client with authentication interceptor
    let mut client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Subscribe to all components on every entity
    let request = Request::new(StreamEntityComponentsRequest {
        include_all_components: true,
        ..Default::default()
    });

    let mut stream = client.stream_entity_components(request).await?.into_inner();

    println!("Starting to receive stream data...");
    while let Some(response) = stream.message().await? {
        if let Some(entity) = response.entity_event.and_then(|e| e.entity) {
            println!("Streaming entity ID: {}", entity.entity_id);
        }
    }
    println!("End of stream reached");

    Ok(())
}

```

If successful, you get a stream of all entities as they are updated in Lattice:

```bash maxLines=3
Entity: Demo-Sim-Asset2
Entity: adsbEntity
Entity: esim.adsb.aus-4005
```

#### Stream specific components

Use `componentsToInclude` and provide a list of components in `snake_case`.
For example, `aliases`, and `location_uncertainty`:

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

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"

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

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

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

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatal("Required environment variables not set.")
	}

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

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

	// Create the entity stream.
	stream, err := latticeClient.Entities.StreamEntities(ctx, &Lattice.EntityStreamRequest{
		PreExistingOnly: Lattice.Bool(false),
		// Define a list of components to control which entities are fetched.
		// If set, Lattice streams only entities with the components you provide.
		ComponentsToInclude: []string{"aliases", "location_uncertainty"},
	})
	if err != nil {
		log.Fatalf("Failed to create entity stream: %v", err)
	}
	defer stream.Close()

	for {
		select {
		case <-ctx.Done():
			log.Printf("Context canceled: %v", ctx.Err())
			return
		default:
			// Continue processing
		}
		message, err := stream.Recv()

		// Handle stream completion
		if errors.Is(err, io.EOF) {
			log.Println("Stream completed successfully.")
			return
		}

		if err != nil {
			log.Printf("Error receiving message: %v", err)
			continue
		}

		// Process the event based whether it is a heartbeat or entity event.
		switch message.Event {
		case "heartbeat":
			timestamp := *message.Heartbeat.Timestamp
			log.Printf("Heartbeat: %s", timestamp)
		case "entity":
			log.Printf("Entity: %s", *message.Entity.Entity.Aliases.Name)
		default:
			log.Printf("Unknown event type: %s", message.Event)
		}
	}
}

```

```java title={"Java (REST)"} startLine={40} maxLines={20}
package org.example;
import java.util.Arrays;
import java.util.Optional;

import com.anduril.AsyncLattice;
import com.anduril.resources.entities.requests.EntityStreamRequest;
import com.anduril.resources.entities.types.StreamEntitiesResponse;
import com.anduril.types.EntityStreamEvent;



/**
 * Example application that streams all entities using the Lattice SDK.
 */
public class StreamSpecificComponents {
    
    public static void main(String[] args) {
        // Get environment variables
        String endpoint = System.getenv("LATTICE_ENDPOINT");
        String environmentToken = System.getenv("ENVIRONMENT_TOKEN");
        String sandboxesToken = System.getenv("SANDBOXES_TOKEN");
        
        // Check required variables
        if (endpoint == null || environmentToken == null || sandboxesToken == null) {
            System.err.println("Missing required environment variables");
            System.exit(1);
        }
        if (!endpoint.startsWith("https://")) endpoint = "https://" + endpoint;
        
        try {
            AsyncLattice client = AsyncLattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();
            
            System.out.println("Starting entity stream...");
            
            try {
                EntityStreamRequest request = EntityStreamRequest.builder()
                    .preExistingOnly(false)
                    .componentsToInclude(Optional.of(Arrays.asList("aliases", "location_uncertainty")))
                    .build();

                client.entities().streamEntities(request)
                    .thenAccept(responses -> {
                        for (StreamEntitiesResponse response : responses) {
                            // If the event is of type heartbeat, log the timestamp.
                            if (response.getHeartbeat() != null && response.getHeartbeat().isPresent()) {
                                System.out.println("Heartbeat received: " + response.getHeartbeat().get().getTimestamp().get());
                            } 
                            // If the event is of type entity, log the entity ID.
                            else if (response.getEntity() != null && response.getEntity().isPresent()) {
                                EntityStreamEvent event = response.getEntity().get();
                                System.out.println("Entity: " + event.getEntity().get().getAliases().get().getName().get());
                            }
                        }
                    })
                    .exceptionally(ex -> {
                        System.err.println("Exception while streaming entities: " + ex.getMessage());
                        return null;
                    })
                    .join();
            } catch (Exception e) {
                System.err.println("Error in streaming loop: " + e.getMessage());
            }
        } catch (Exception e) {
            System.err.println("Failed to initialize: " + e.getMessage());
        }
    }
}
```

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

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

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

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

async def app():
    try:
        event_stream = client.entities.stream_entities(
            pre_existing_only=False,
            components_to_include=["aliases", "location_uncertainty"]
        )
        async for event in event_stream:
            # If the event is of type heartbeat, log the timestamp.
            if event.event == "heartbeat":
                print(f'Heartbeat: {event.timestamp}')
            # If the event is of type entity, log the entity name.
            elif event.entity and event.entity.aliases:
                print(f'Entity: {event.entity.aliases.name}')
    except asyncio.CancelledError:
        print("Streaming cancelled...")
    except Exception as error:
        print(f"Exception: {error}")

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

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

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

// Remove sandboxesToken from the following statement if you are not developing on Sandboxes.
const sandboxesToken = process.env.SANDBOXES_TOKEN;
if (!latticeEndpoint || !environmentToken || !sandboxesToken) {
    console.log('Missing required environment variables.');
    process.exit(1);
}

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

async function App() {
    try {
        // Start streaming entities
        const eventStream = await client.entities.streamEntities(
            {
                preExistingOnly: false,
                componentsToInclude: ["aliases", "location_uncertainty"]
            }
        );
        
        // Process the stream
        for await (const event of eventStream) {
            // Check if it's a heartbeat event.
            if (event.event === "heartbeat") {
                console.log(`Heartbeat received: ${event.timestamp}`);
                continue;
            }
            // Process entity event.
            const entity = event.entity;
            if (entity) {
                console.log(`Entity: ${entity.aliases?.name}`);
            }
        }
    } catch (error) {
        console.log(`Exception while streaming entities: ${error}`);
    }
}

App();
```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func main() {
	ctx := context.Background()

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

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		log.Fatalf("Missing required environment variables")
	}
	auth := &BearerTokenAuth{
		Token:          environmentToken,
		SandboxesToken: sandboxesToken,
	}

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

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

	client := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	stream, err := client.StreamEntityComponents(ctx, &entitymanagerv1.StreamEntityComponentsRequest{
		ComponentsToInclude: []string{"aliases", "location"},
	})
	if err != nil {
		log.Fatalf("Error creating stream: %v", err)
	}

	log.Println("Starting to receive stream data...")
	for {
		response, err := stream.Recv()
		if err == io.EOF {
			// End of stream
			log.Println("End of stream reached.")
			break
		}
		if err != nil {
			log.Fatalf("Error receiving stream data: %v", err)
		}

		entity := response.GetEntityEvent().GetEntity()
		if position := entity.GetLocation().GetPosition(); position != nil {
			log.Printf("Entity %s at location: %f, %f",
				entity.EntityId,
				position.LatitudeDegrees,
				position.LongitudeDegrees,
			)
		}
	}
}

```

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

import os
import sys
from bearer_auth import create_client

from anduril.entitymanager.v1.entity_manager_api_pub_pb import (
    StreamEntityComponentsRequest,
)
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync


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

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

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

    request = StreamEntityComponentsRequest(components_to_include=["aliases", "location"])

    try:
        stream = client.stream_entity_components(request)
    except Exception as error:
        print(f"Error creating stream: {error}", file=sys.stderr)
        sys.exit(1)

    print("Starting to receive stream data...")

    try:
        for response in stream:
            # Heartbeat messages carry no entity_event, so skip them.
            if not response.entity_event or not response.entity_event.entity:
                continue
            entity = response.entity_event.entity
            if not entity.location or not entity.location.position:
                continue
            position = entity.location.position
            if position.latitude_degrees or position.longitude_degrees:
                print(
                    f"Entity {entity.entity_id} at location: "
                    f"{position.latitude_degrees}, {position.longitude_degrees}"
                )
        print("End of stream reached.")
    except Exception as error:
        print(f"Error receiving stream data: {error}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

```

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

use anduril_lattice_sdk_community_neoeinstein_prost::anduril::entitymanager::v1::StreamEntityComponentsRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

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

    // Set up authentication handler
    let auth = EnvironmentTokenAuth::new(
        environment_token,
        sandboxes_token,
    );

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

    // Fetch a fresh access token and build metadata values
    let access_token = auth.get_token().await?;
    let sandboxes_token = auth.sandboxes_token();
    let auth_header: MetadataValue<_> = format!("Bearer {}", access_token).parse()?;
    let sandbox_header: MetadataValue<_> = format!("Bearer {}", sandboxes_token).parse()?;

    // Create EntityManager API client with authentication interceptor
    let mut client = EntityManagerApiClient::with_interceptor(
        channel,
        move |mut req: Request<()>| {
            req.metadata_mut()
                .insert("authorization", auth_header.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", sandbox_header.clone());
            Ok(req)
        },
    );

    // Subscribe to only the aliases and location components on every entity
    let request = Request::new(StreamEntityComponentsRequest {
        components_to_include: vec!["aliases".to_string(), "location".to_string()],
        ..Default::default()
    });

    let mut stream = client.stream_entity_components(request).await?.into_inner();

    println!("Starting to receive stream data...");
    while let Some(response) = stream.message().await? {
        if let Some(entity) = response.entity_event.and_then(|e| e.entity) {
            if let Some(position) = entity.location.and_then(|l| l.position) {
                println!(
                    "Entity {} at location: {}, {}",
                    entity.entity_id, position.latitude_degrees, position.longitude_degrees
                );
            }
        }
    }
    println!("End of stream reached.");

    Ok(())
}

```

If you're directly invoking the [`streamEntities`](/reference/rest/entities/stream-entities) REST API using `curl` or another
similar tool, you must list the components using `camelCase`: `locationUncertainty`.

If successful, you receive a real-time stream of entities with `aliases` populated:

```bash maxLines=3
Entity: ADS-B: N113PF
Entity: DIVE
Entity: FISHING VESSEL (37958)
```

## What's next?

* Learn more about the Entities API in [REST](/reference/rest/entities/publish-entity) and
  [gRPC](/reference/grpc/anduril-entitymanager-v-1/entity-manager-api/anduril-entitymanager-v-1-publish-entity).
* Learn how to [Publish entities to Lattice](/guides/entities/publish).
* Check out the Lattice [sample apps](/samples/overview).

> Monitoring entities in Lattice using the Lattice SDK