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

# Manage streams

> List, retrieve, stop, and re-egress existing video streams.

Use the Lattice SDK to inspect active streams, retrieve a specific stream, archive a stream when its source goes offline, and re-publish over RTSP or SRT for downstream consumers.

## Before you begin

* To inspect and manage streams, [set up](/guides/getting-started/set-up) your Lattice environment.
* See an overview of the [Video Manager API and stream lifecycle](/guides/video/overview) in Lattice.
* If you haven't already, learn how to [stream a video](/guides/video/create-streams).

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

## Stream ownership

Your integration owns the lifecycle of the `IngressStream` records it creates. Streams are not automatically removed when a publisher disconnects, a stream becomes inactive, or your integration stops communicating with Lattice. A stream with status `STREAM_STATUS_INACTIVE` still exists; when the source or publisher reconnects, reuse the existing stream rather than creating another one.

Persist each stream's `ingress_id` with the asset or integration that owns it. On startup or reconnect, call `GetIngressStream` with the saved ID to check the stream. If you don't have the ID, use `ListIngressStreams` and inspect its pages to find the stream. Create a new stream only after checking that an existing one for the source does not already exist.

## List active streams

Call `ListIngressStreams` to retrieve the current page of active streams. Archived streams are not returned. Send an empty request to fetch the first page (default: 50, max: 100), then iterate while `next_page_token` is non-empty.

> **Pagination contract**
>
> Pass the previous response's `next_page_token` as `page_token` and keep every other request field identical. The server may reject requests that change other fields between pages.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to list active ingress streams.
package main

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

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

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

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

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

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

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

	ctx := context.Background()

	// Iterate every page of active streams. Keep all request fields identical between
	// pages other than pageToken; the server may reject a request that changes them.
	pageToken := ""
	for {
		response, err := LatticeClient.Video.ListIngressStreams(ctx, &Lattice.ListIngressStreamsRequest{
			PageToken: &pageToken,
		})
		if err != nil {
			fmt.Printf("Failed to list streams: %v\n", err)
			os.Exit(1)
		}

		for _, stream := range response.GetIngressStreams() {
			fmt.Printf("Stream %s status %s\n", *stream.GetIngressID(), *stream.GetStatus())
		}

		if response.GetNextPageToken() == nil || *response.GetNextPageToken() == "" {
			break
		}
		pageToken = *response.GetNextPageToken()
	}
}

```

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

import java.util.List;

import com.anduril.Lattice;
import com.anduril.resources.video.requests.ListIngressStreamsRequest;
import com.anduril.types.IngressStream;
import com.anduril.types.ListIngressStreamsResponse;

/**
 * Lists active ingress streams with the Lattice SDK.
 */
public class ListIngressStreams {
    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 and ensure URL has https://
        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 {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Iterate every page of active streams. Keep all request fields identical between
            // pages other than page_token; the server may reject a request that changes them.
            String pageToken = "";
            while (true) {
                ListIngressStreamsResponse response = client.video().listIngressStreams(
                    ListIngressStreamsRequest.builder()
                        .pageToken(pageToken)
                        .build());

                List<IngressStream> streams = response.getIngressStreams().orElse(List.of());
                for (IngressStream stream : streams) {
                    System.out.println("Stream " + stream.getIngressId().orElse("")
                        + " status " + stream.getStatus().map(Object::toString).orElse(""));
                }

                String nextPageToken = response.getNextPageToken().orElse("");
                if (nextPageToken.isEmpty()) {
                    break;
                }
                pageToken = nextPageToken;
            }
        } catch (Exception e) {
            System.err.println("Failed to list streams: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to list active ingress
# streams.

import os
import sys

from anduril import Lattice


def main():
    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)

    client = Lattice(
        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}"},
    )

    # Iterate every page of active streams. Keep all request fields identical between
    # pages other than page_token; the server may reject a request that changes them.
    page_token = ""
    while True:
        try:
            response = client.video.list_ingress_streams(page_token=page_token)
        except Exception as error:
            print(f"Failed to list streams: {error}", file=sys.stderr)
            sys.exit(1)

        for stream in response.ingress_streams or []:
            print(f"Stream {stream.ingress_id} status {stream.status}")

        if not response.next_page_token:
            break
        page_token = response.next_page_token


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
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,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

async function App() {
    try {
        // Iterate every page of active streams. Keep all request fields identical between
        // pages other than pageToken; the server may reject a request that changes them.
        let pageToken = "";
        while (true) {
            const response = await client.video.listIngressStreams({ pageToken });

            for (const stream of response.ingressStreams ?? []) {
                console.log(`Stream ${stream.ingressId} status ${stream.status}`);
            }

            if (!response.nextPageToken) {
                break;
            }
            pageToken = response.nextPageToken;
        }
    } catch (error) {
        console.log(`Encountered the following error while listing streams: ${error}`);
    }
}

App();

```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

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

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

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		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 := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// Iterate every page of active streams. Keep all request fields identical between
	// pages other than page_token; the server may reject a request that changes them.
	pageToken := ""
	for {
		resp, err := client.ListIngressStreams(ctx, &videomanagerv1.ListIngressStreamsRequest{
			PageToken: pageToken,
		})
		if err != nil {
			log.Fatalf("Failed to list streams: %v", err)
		}
		for _, stream := range resp.GetIngressStreams() {
			fmt.Printf("Stream %s status %s\n", stream.GetIngressId(), stream.GetStatus())
		}
		if resp.GetNextPageToken() == "" {
			break
		}
		pageToken = resp.GetNextPageToken()
	}
}

```

```py title={"Python (gRPC)"}
# 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.videomanager.v1.list_ingress_streams_request_pub_pb import (
    ListIngressStreamsRequest,
)
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)


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(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Iterate every page of active streams. Keep all request fields identical between
    # pages other than page_token; the server may reject a request that changes them.
    page_token = ""
    while True:
        try:
            response = client.list_ingress_streams(
                ListIngressStreamsRequest(page_token=page_token),
                timeout_ms=10000,
            )
        except Exception as error:
            print(f"Failed to list streams: {error}", file=sys.stderr)
            sys.exit(1)

        for stream in response.ingress_streams:
            print(f"Stream {stream.ingress_id} status {stream.status}")

        if not response.next_page_token:
            break
        page_token = response.next_page_token


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// 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::videomanager::v1::ListIngressStreamsRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

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

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

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

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

    let mut client = VideoManagerApiClient::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)
        },
    );

    // Iterate every page of active streams. Keep all request fields identical between
    // pages other than page_token; the server may reject a request that changes them.
    let mut page_token = String::new();
    loop {
        let request = Request::new(ListIngressStreamsRequest {
            page_size: 0,
            page_token: page_token.clone(),
        });
        let response = client.list_ingress_streams(request).await?.into_inner();

        for stream in &response.ingress_streams {
            println!(
                "Stream {} status {:?}",
                stream.ingress_id,
                stream.status()
            );
        }

        if response.next_page_token.is_empty() {
            break;
        }
        page_token = response.next_page_token;
    }

    Ok(())
}

```

## Get a single stream

Call `GetIngressStream` with an `ingress_id` to retrieve a single `IngressStream`. The response includes the current status, ingress configuration, and associated egress stream identifiers. For a description of every field, see [IngressStream object](/guides/video/overview#ingressstream-object).

> **MPEG-TS streams are read-only over the public internet**
>
> MPEG-TS ingress is supported only at the edge, in closed networks. If you connect to Lattice over the public internet, you can't start an MPEG-TS stream, but a stream created at the edge still appears in `ListIngressStreams` and can be inspected with `GetIngressStream`.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to retrieve a single ingress stream.
package main

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

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

// Ingress ID of the stream to get. Replace before running.
const ingressID = "<INGRESS-ID>"

func main() {
	// Get environment variables
	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")

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

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

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

	ctx := context.Background()

	response, err := LatticeClient.Video.GetIngressStream(ctx, &Lattice.GetIngressStreamRequest{
		IngressID: ingressID,
	})
	if err != nil {
		fmt.Printf("Failed to get stream: %v\n", err)
		os.Exit(1)
	}

	stream := response.GetIngressStream()
	if stream == nil {
		fmt.Printf("No ingress stream found for %s\n", ingressID)
		os.Exit(1)
	}
	fmt.Printf("Ingress ID: %s\n", *stream.GetIngressID())
	fmt.Printf("Title:      %q\n", *stream.GetTitle())
	fmt.Printf("Status:     %s\n", *stream.GetStatus())
}

```

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

import com.anduril.Lattice;
import com.anduril.types.GetIngressStreamResponse;
import com.anduril.types.IngressStream;

/**
 * Retrieves a single ingress stream with the Lattice SDK.
 */
public class GetIngressStream {
    // Ingress ID of the stream to get. Replace before running.
    private static final String INGRESS_ID = "<INGRESS-ID>";

    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 and ensure URL has https://
        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 {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            GetIngressStreamResponse response = client.video().getIngressStream(INGRESS_ID);
            IngressStream stream = response.getIngressStream().orElse(null);
            if (stream == null) {
                System.err.println("No ingress stream found for " + INGRESS_ID);
                System.exit(1);
            }
            System.out.println("Ingress ID: " + stream.getIngressId().orElse(""));
            System.out.println("Title:      " + stream.getTitle().orElse(""));
            System.out.println("Status:     " + stream.getStatus().map(Object::toString).orElse(""));
        } catch (Exception e) {
            System.err.println("Failed to get stream: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to retrieve a single ingress
# stream.

import os
import sys

from anduril import Lattice

# Ingress ID of the stream to get. Replace before running.
INGRESS_ID = "<INGRESS-ID>"


def main():
    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)

    client = Lattice(
        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}"},
    )

    try:
        response = client.video.get_ingress_stream(INGRESS_ID)
    except Exception as error:
        print(f"Failed to get stream: {error}", file=sys.stderr)
        sys.exit(1)

    stream = response.ingress_stream
    if not stream:
        print(f"No ingress stream found for {INGRESS_ID}", file=sys.stderr)
        sys.exit(1)
    print(f"Ingress ID: {stream.ingress_id}")
    print(f"Title:      {stream.title!r}")
    print(f"Status:     {stream.status}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
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,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Ingress ID of the stream to get. Replace before running.
const ingressId = "<INGRESS-ID>";

async function App() {
    try {
        const response = await client.video.getIngressStream({ ingressId });

        const stream = response.ingressStream;
        if (!stream) {
            console.log(`No ingress stream found for ${ingressId}`);
            return;
        }
        console.log(`Ingress ID: ${stream.ingressId}`);
        console.log(`Title:      ${JSON.stringify(stream.title)}`);
        console.log(`Status:     ${stream.status}`);
    } catch (error) {
        console.log(`Encountered the following error while fetching the stream: ${error}`);
    }
}

App();

```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

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

// Ingress ID of the stream to get. Replace before running.
const ingressID = "<INGRESS-ID>"

func main() {
	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 := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	resp, err := client.GetIngressStream(ctx, &videomanagerv1.GetIngressStreamRequest{
		IngressId: ingressID,
	})
	if err != nil {
		log.Fatalf("Failed to get stream: %v", err)
	}

	stream := resp.GetIngressStream()
	fmt.Printf("Ingress ID: %s\n", stream.GetIngressId())
	fmt.Printf("Title:      %q\n", stream.GetTitle())
	fmt.Printf("Status:     %s\n", stream.GetStatus())
}

```

```py title={"Python (gRPC)"}
# 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.videomanager.v1.get_ingress_stream_request_pub_pb import (
    GetIngressStreamRequest,
)
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Ingress ID of the stream to get. Replace before running.
INGRESS_ID = "<INGRESS-ID>"


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(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    try:
        response = client.get_ingress_stream(
            GetIngressStreamRequest(ingress_id=INGRESS_ID),
            timeout_ms=10000,
        )
    except Exception as error:
        print(f"Failed to get stream: {error}", file=sys.stderr)
        sys.exit(1)

    if not response.has_field("ingress_stream"):
        print(f"No ingress stream found for {INGRESS_ID}", file=sys.stderr)
        sys.exit(1)
    stream = response.ingress_stream
    print(f"Ingress ID: {stream.ingress_id}")
    print(f"Title:      {stream.title!r}")
    print(f"Status:     {stream.status}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// 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::videomanager::v1::GetIngressStreamRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Ingress ID of the stream to get. Replace before running.
const INGRESS_ID: &str = "<INGRESS-ID>";

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

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

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

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

    let mut client = VideoManagerApiClient::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)
        },
    );

    let request = Request::new(GetIngressStreamRequest {
        ingress_id: INGRESS_ID.to_string(),
    });
    let response = client.get_ingress_stream(request).await?.into_inner();

    match response.ingress_stream {
        Some(stream) => {
            println!("Ingress ID: {}", stream.ingress_id);
            println!("Title:      {:?}", stream.title);
            println!("Status:     {:?}", stream.status());
        }
        None => eprintln!("No ingress stream found for {}", INGRESS_ID),
    }

    Ok(())
}

```

## Stop an ingress stream

Stopping a stream archives it. The `IngressStream` record remains retrievable with `GetIngressStream`, but its status transitions to `STREAM_STATUS_ARCHIVED` and `ListIngressStreams` no longer returns the record.

> **Archive, not delete**
>
> `DeleteIngressStream` does not delete the underlying record. To remove the stale media reference from the camera entity, follow up with an `OverrideEntity` call as shown in the example below. The example reads the entity, filters out the stale `MediaItem`, and writes the surviving items back. A `field_path` of `media` replaces the entire `Media` component while leaving every other entity component untouched.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to stop (archive) an ingress stream
// and remove its stale media reference from the associated camera entity.
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"
)

const (
	// Ingress ID of the stream to stop. Replace before running.
	ingressID = "<DELETE-INGRESS-ID>"
	// Entity the video is attached to. Set to "" to only stop the stream without
	// editing any entity's media list.
	entityID = "<ENTITY-ID>"
)

func main() {
	// Get environment variables
	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")

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

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

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

	ctx := context.Background()

	fmt.Printf("Stopping ingress stream with ID: %s\n", ingressID)
	_, err := LatticeClient.Video.DeleteIngressStream(ctx, &Lattice.DeleteIngressStreamRequest{
		IngressID: ingressID,
	})
	if err != nil {
		fmt.Printf("Failed to stop ingress stream: %v\n", err)
		os.Exit(1)
	}

	// With no entity to update, stopping the stream is all there is to do.
	if entityID == "" {
		return
	}

	// Read the entity, filter the stale MediaItem out, and write the surviving items back.
	entity, err := LatticeClient.Entities.GetEntity(ctx, &Lattice.GetEntityRequest{
		EntityID: entityID,
	})
	if err != nil {
		fmt.Printf("Failed to fetch entity %s: %v\n", entityID, err)
		os.Exit(1)
	}

	var current []*Lattice.MediaItem
	if entity.GetMedia() != nil {
		current = entity.GetMedia().GetMedia()
	}
	remaining := make([]*Lattice.MediaItem, 0, len(current))
	for _, item := range current {
		if item.GetItemIdentifier() == nil || *item.GetItemIdentifier() != ingressID {
			remaining = append(remaining, item)
		}
	}

	if len(remaining) == len(current) {
		fmt.Printf("Video %s was not in entity %s media list; nothing to remove\n", ingressID, entityID)
		return
	}

	// A field_path of media replaces the entire Media component with the surviving items.
	_, err = LatticeClient.Entities.OverrideEntity(ctx, &Lattice.EntityOverride{
		EntityID:  entityID,
		FieldPath: "media",
		Entity: &Lattice.Entity{
			EntityID: Lattice.String(entityID),
			Media: &Lattice.Media{
				Media: remaining,
			},
		},
		Provenance: &Lattice.Provenance{
			IntegrationName:  Lattice.String("lattice-video-integration"),
			DataType:         Lattice.String("video-association"),
			SourceUpdateTime: Lattice.Time(time.Now().UTC()),
		},
	})
	if err != nil {
		fmt.Printf("Failed to override entity media: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Removed video %s from entity %s\n", ingressID, entityID)
}

```

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

import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;

import com.anduril.Lattice;
import com.anduril.resources.entities.requests.EntityOverride;
import com.anduril.types.Entity;
import com.anduril.types.Media;
import com.anduril.types.MediaItem;
import com.anduril.types.Provenance;

/**
 * Stops (archives) an ingress stream with the Lattice SDK and removes its stale media reference
 * from the associated camera entity.
 */
public class DeleteIngressStream {
    // Ingress ID of the stream to stop. Replace before running.
    private static final String INGRESS_ID = "<DELETE-INGRESS-ID>";
    // Entity the video is attached to. Set to "" to only stop the stream without
    // editing any entity's media list.
    private static final String ENTITY_ID = "<ENTITY-ID>";

    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 and ensure URL has https://
        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 {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            System.out.println("Stopping ingress stream with ID: " + INGRESS_ID);
            client.video().deleteIngressStream(INGRESS_ID);

            // With no entity to update, stopping the stream is all there is to do.
            if (ENTITY_ID.isEmpty()) {
                return;
            }

            // Read the entity, filter the stale MediaItem out, and write the surviving items back.
            Entity entity = client.entities().getEntity(ENTITY_ID);
            List<MediaItem> current = entity.getMedia().flatMap(Media::getMedia).orElse(List.of());
            List<MediaItem> remaining = new ArrayList<>();
            for (MediaItem item : current) {
                if (!item.getItemIdentifier().orElse("").equals(INGRESS_ID)) {
                    remaining.add(item);
                }
            }

            if (remaining.size() == current.size()) {
                System.out.println("Video " + INGRESS_ID + " was not in entity " + ENTITY_ID
                    + " media list; nothing to remove");
                return;
            }

            EntityOverride entityOverride = EntityOverride.builder()
                .entity(Entity.builder()
                    .entityId(ENTITY_ID)
                    .media(Media.builder()
                        .media(remaining)
                        .build())
                    .build())
                .provenance(Provenance.builder()
                    .integrationName("lattice-video-integration")
                    .dataType("video-association")
                    .sourceUpdateTime(OffsetDateTime.now(ZoneOffset.UTC))
                    .build())
                .build();
            client.entities().overrideEntity(ENTITY_ID, "media", entityOverride);

            System.out.println("Removed video " + INGRESS_ID + " from entity " + ENTITY_ID);
        } catch (Exception e) {
            System.err.println("Exception: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to stop (archive) an ingress
# stream and remove its stale media reference from the associated camera entity.

import os
import sys
from datetime import datetime, timezone

from anduril import Entity, Lattice, Media, Provenance

# Ingress ID of the stream to stop. Replace before running.
INGRESS_ID = "<DELETE-INGRESS-ID>"
# Entity the video is attached to. Set to "" to only stop the stream without
# editing any entity's media list.
ENTITY_ID = "<ENTITY-ID>"


def main():
    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)

    client = Lattice(
        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}"},
    )

    print(f"Stopping ingress stream with ID: {INGRESS_ID}")
    try:
        client.video.delete_ingress_stream(INGRESS_ID)
    except Exception as error:
        print(f"Failed to stop ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    # With no entity to update, stopping the stream is all there is to do.
    if not ENTITY_ID:
        return

    # Read the entity, filter the stale MediaItem out, and write the surviving items back.
    try:
        entity = client.entities.get_entity(ENTITY_ID)
    except Exception as error:
        print(f"Failed to fetch entity {ENTITY_ID}: {error}", file=sys.stderr)
        sys.exit(1)

    current = list(entity.media.media) if entity.media and entity.media.media else []
    remaining = [item for item in current if item.item_identifier != INGRESS_ID]

    if len(remaining) == len(current):
        print(
            f"Video {INGRESS_ID} was not in entity {ENTITY_ID} media list; "
            "nothing to remove"
        )
        return

    try:
        client.entities.override_entity(
            entity_id=ENTITY_ID,
            field_path="media",
            entity=Entity(
                entity_id=ENTITY_ID,
                media=Media(media=remaining),
            ),
            provenance=Provenance(
                integration_name="lattice-video-integration",
                data_type="video-association",
                source_update_time=datetime.now(timezone.utc),
            ),
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Removed video {INGRESS_ID} from entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
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,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Ingress ID of the stream to stop. Replace before running.
const ingressId = "<DELETE-INGRESS-ID>";
// Entity the video is attached to. Set to "" to only stop the stream without
// editing any entity's media list.
const entityId = "<ENTITY-ID>";

async function App() {
    try {
        console.log(`Stopping ingress stream with ID: ${ingressId}`);
        await client.video.deleteIngressStream({ ingressId });

        // With no entity to update, stopping the stream is all there is to do.
        if (!entityId) {
            return;
        }

        // Read the entity, filter the stale MediaItem out, and write the surviving items back.
        const entity = await client.entities.getEntity({ entityId });
        const current = entity.media?.media ?? [];
        const remaining = current.filter((item) => item.itemIdentifier !== ingressId);

        if (remaining.length === current.length) {
            console.log(`Video ${ingressId} was not in entity ${entityId} media list; nothing to remove`);
            return;
        }

        // A field_path of media replaces the entire Media component with the surviving items.
        await client.entities.overrideEntity({
            entityId,
            fieldPath: "media",
            entity: {
                entityId: entityId,
                media: { media: remaining }
            },
            provenance: {
                integrationName: "lattice-video-integration",
                dataType: "video-association",
                sourceUpdateTime: new Date().toISOString()
            }
        });
        console.log(`Removed video ${ingressId} from entity ${entityId}`);
    } catch (error) {
        console.log(`Encountered the following error while stopping the stream: ${error}`);
    }
}

App();

```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/entitymanager/v1/entitymanagerv1grpc"
	entitymanager "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/entitymanager/v1"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	// Ingress ID of the stream to stop. Replace before running.
	ingressID = "<DELETE-INGRESS-ID>"
	// Entity the video is attached to. Set to "" to only stop the stream without
	// editing any entity's media list.
	entityID = "<ENTITY-ID>"
)

func main() {
	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()

	videoClient := videomanagerv1grpc.NewVideoManagerAPIClient(conn)
	emClient := entitymanagerv1grpc.NewEntityManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	fmt.Printf("Stopping ingress stream with ID: %s\n", ingressID)
	_, err = videoClient.DeleteIngressStream(ctx, &videomanagerv1.DeleteIngressStreamRequest{
		IngressId: ingressID,
	})
	if err != nil {
		log.Fatalf("Failed to stop ingress stream: %v", err)
	}

	// With no entity to update, stopping the stream is all there is to do.
	if entityID == "" {
		return
	}

	// Read the entity, filter the stale MediaItem out, and write the surviving items back.
	getResp, err := emClient.GetEntity(ctx, &entitymanager.GetEntityRequest{EntityId: entityID})
	if err != nil {
		log.Fatalf("Failed to fetch entity %s: %v", entityID, err)
	}

	current := getResp.GetEntity().GetMedia().GetMedia()
	remaining := make([]*entitymanager.MediaItem, 0, len(current))
	for _, item := range current {
		if item.GetItemIdentifier() == ingressID {
			continue
		}
		remaining = append(remaining, item)
	}

	if len(remaining) == len(current) {
		fmt.Printf("Video %s was not in entity %s media list; nothing to remove\n", ingressID, entityID)
		return
	}

	_, err = emClient.OverrideEntity(ctx, &entitymanager.OverrideEntityRequest{
		Entity: &entitymanager.Entity{
			EntityId: entityID,
			Media:    &entitymanager.Media{Media: remaining},
		},
		FieldPath: []string{"media"},
		Provenance: &entitymanager.Provenance{
			IntegrationName:  "lattice-video-integration",
			DataType:         "video-association",
			SourceUpdateTime: timestamppb.Now(),
		},
	})
	if err != nil {
		log.Fatalf("Failed to override entity media: %v", err)
	}
	fmt.Printf("Removed video %s from entity %s\n", ingressID, entityID)
}

```

```py title={"Python (gRPC)"}
# 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 datetime import datetime, timezone
from bearer_auth import create_client
from protobuf.wkt import Timestamp

from anduril.entitymanager.v1.entity_pub_pb import Entity, Provenance
from anduril.entitymanager.v1.entity_manager_api_pub_pb import (
    GetEntityRequest,
    OverrideEntityRequest,
)
from anduril.entitymanager.v1.entity_manager_api_pub_connect import EntityManagerAPIClientSync
from anduril.entitymanager.v1.media_pub_pb import Media

from anduril.videomanager.v1.delete_ingress_stream_request_pub_pb import (
    DeleteIngressStreamRequest,
)
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Ingress ID of the stream to stop. Replace before running.
INGRESS_ID = "<DELETE-INGRESS-ID>"
# Entity the video is attached to. Set to "" to only stop the stream without
# editing any entity's media list.
ENTITY_ID = "<ENTITY-ID>"


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 clients over Connect gRPC with authentication
    video_client = create_client(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)
    em_client = create_client(EntityManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    print(f"Stopping ingress stream with ID: {INGRESS_ID}")
    try:
        video_client.delete_ingress_stream(
            DeleteIngressStreamRequest(ingress_id=INGRESS_ID),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to stop ingress stream: {error}", file=sys.stderr)
        sys.exit(1)

    # With no entity to update, stopping the stream is all there is to do.
    if not ENTITY_ID:
        return

    # Read the entity, filter the stale MediaItem out, and write the surviving items back.
    try:
        get_resp = em_client.get_entity(
            GetEntityRequest(entity_id=ENTITY_ID),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to fetch entity {ENTITY_ID}: {error}", file=sys.stderr)
        sys.exit(1)

    current = list(get_resp.entity.media.media)
    remaining = [item for item in current if item.item_identifier != INGRESS_ID]

    if len(remaining) == len(current):
        print(
            f"Video {INGRESS_ID} was not in entity {ENTITY_ID} media list; "
            "nothing to remove"
        )
        return

    now_ts = Timestamp.from_datetime(datetime.now(timezone.utc))

    try:
        em_client.override_entity(
            OverrideEntityRequest(
                entity=Entity(
                    entity_id=ENTITY_ID,
                    media=Media(media=remaining),
                ),
                field_path=["media"],
                provenance=Provenance(
                    integration_name="lattice-video-integration",
                    data_type="video-association",
                    source_update_time=now_ts,
                ),
            ),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to override entity media: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Removed video {INGRESS_ID} from entity {ENTITY_ID}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// 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::{
    Entity, GetEntityRequest, Media, OverrideEntityRequest, Provenance,
};
use anduril_lattice_sdk_community_neoeinstein_prost::anduril::videomanager::v1::DeleteIngressStreamRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::entitymanager::v1::tonic::entity_manager_api_client::EntityManagerApiClient;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use prost_types::Timestamp;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;
use std::time::{SystemTime, UNIX_EPOCH};

// Ingress ID of the stream to stop. Replace before running.
const INGRESS_ID: &str = "<DELETE-INGRESS-ID>";
// Entity the video is attached to. Set to "" to only stop the stream without
// editing any entity's media list.
const ENTITY_ID: &str = "<ENTITY-ID>";

fn now_timestamp() -> Timestamp {
    let d = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
    Timestamp {
        seconds: d.as_secs() as i64,
        nanos: d.subsec_nanos() as i32,
    }
}

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

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

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

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

    // Both clients share the same channel and inject the same auth metadata.
    let video_auth = auth_header.clone();
    let video_sandbox = sandbox_header.clone();
    let mut video_client = VideoManagerApiClient::with_interceptor(
        channel.clone(),
        move |mut req: Request<()>| {
            req.metadata_mut().insert("authorization", video_auth.clone());
            req.metadata_mut()
                .insert("anduril-sandbox-authorization", video_sandbox.clone());
            Ok(req)
        },
    );
    let mut em_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!("Stopping ingress stream with ID: {}", INGRESS_ID);
    video_client
        .delete_ingress_stream(Request::new(DeleteIngressStreamRequest {
            ingress_id: INGRESS_ID.to_string(),
        }))
        .await?;

    // With no entity to update, stopping the stream is all there is to do.
    if ENTITY_ID.is_empty() {
        return Ok(());
    }

    // Read the entity, filter the stale MediaItem out, and write the surviving items back.
    let get_resp = em_client
        .get_entity(Request::new(GetEntityRequest {
            entity_id: ENTITY_ID.to_string(),
        }))
        .await?
        .into_inner();

    let current = get_resp
        .entity
        .and_then(|e| e.media)
        .map(|m| m.media)
        .unwrap_or_default();
    let original_len = current.len();
    let remaining: Vec<_> = current
        .into_iter()
        .filter(|item| item.item_identifier != INGRESS_ID)
        .collect();

    if remaining.len() == original_len {
        println!(
            "Video {} was not in entity {} media list; nothing to remove",
            INGRESS_ID, ENTITY_ID
        );
        return Ok(());
    }

    // A field_path of media replaces the entire Media component with the surviving items.
    let override_req = Request::new(OverrideEntityRequest {
        entity: Some(Entity {
            entity_id: ENTITY_ID.to_string(),
            media: Some(Media { media: remaining }),
            ..Default::default()
        }),
        field_path: vec!["media".to_string()],
        provenance: Some(Provenance {
            integration_name: "lattice-video-integration".to_string(),
            data_type: "video-association".to_string(),
            source_update_time: Some(now_timestamp()),
            ..Default::default()
        }),
    });
    match em_client.override_entity(override_req).await {
        Ok(_) => println!("Removed video {} from entity {}", INGRESS_ID, ENTITY_ID),
        Err(e) => eprintln!("Failed to override entity media: {}", e),
    }

    Ok(())
}

```

## List active egress streams

Call `ListEgressStreams` to retrieve the current page of active egress streams. Each entry pairs an `egress_id` with the `ingress_id` it re-publishes, so you can see every downstream consumer of a source. Send an empty request to fetch the first page (default: 50, max: 100), then iterate while `next_page_token` is non-empty.

> **Pagination contract**
>
> Pass the previous response's `next_page_token` as `page_token` and keep every other request field identical. The server may reject requests that change other fields between pages.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to list active egress streams.
package main

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

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

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

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

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

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

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

	ctx := context.Background()

	// Iterate every page of active streams. Keep all request fields identical between
	// pages other than pageToken; the server may reject a request that changes them.
	pageToken := ""
	for {
		response, err := LatticeClient.Video.ListEgressStreams(ctx, &Lattice.ListEgressStreamsRequest{
			PageToken: &pageToken,
		})
		if err != nil {
			fmt.Printf("Failed to list egress streams: %v\n", err)
			os.Exit(1)
		}

		for _, stream := range response.GetEgressStreams() {
			fmt.Printf("Egress %s re-publishes ingress %s\n", *stream.GetEgressID(), *stream.GetIngressID())
		}

		if response.GetNextPageToken() == nil || *response.GetNextPageToken() == "" {
			break
		}
		pageToken = *response.GetNextPageToken()
	}
}

```

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

import java.util.List;

import com.anduril.Lattice;
import com.anduril.resources.video.requests.ListEgressStreamsRequest;
import com.anduril.types.EgressStream;
import com.anduril.types.ListEgressStreamsResponse;

/**
 * Lists active egress streams with the Lattice SDK.
 */
public class ListEgressStreams {
    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 and ensure URL has https://
        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 {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Iterate every page of active streams. Keep all request fields identical between
            // pages other than page_token; the server may reject a request that changes them.
            String pageToken = "";
            while (true) {
                ListEgressStreamsResponse response = client.video().listEgressStreams(
                    ListEgressStreamsRequest.builder()
                        .pageToken(pageToken)
                        .build());

                List<EgressStream> streams = response.getEgressStreams().orElse(List.of());
                for (EgressStream stream : streams) {
                    System.out.println("Egress " + stream.getEgressId().orElse("")
                        + " re-publishes ingress " + stream.getIngressId().orElse(""));
                }

                String nextPageToken = response.getNextPageToken().orElse("");
                if (nextPageToken.isEmpty()) {
                    break;
                }
                pageToken = nextPageToken;
            }
        } catch (Exception e) {
            System.err.println("Failed to list egress streams: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to list active egress
# streams.

import os
import sys

from anduril import Lattice


def main():
    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)

    client = Lattice(
        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}"},
    )

    # Iterate every page of active streams. Keep all request fields identical between
    # pages other than page_token; the server may reject a request that changes them.
    page_token = ""
    while True:
        try:
            response = client.video.list_egress_streams(page_token=page_token)
        except Exception as error:
            print(f"Failed to list egress streams: {error}", file=sys.stderr)
            sys.exit(1)

        for stream in response.egress_streams or []:
            print(f"Egress {stream.egress_id} re-publishes ingress {stream.ingress_id}")

        if not response.next_page_token:
            break
        page_token = response.next_page_token


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
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,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

async function App() {
    try {
        // Iterate every page of active streams. Keep all request fields identical between
        // pages other than pageToken; the server may reject a request that changes them.
        let pageToken = "";
        while (true) {
            const response = await client.video.listEgressStreams({ pageToken });

            for (const stream of response.egressStreams ?? []) {
                console.log(`Egress ${stream.egressId} re-publishes ingress ${stream.ingressId}`);
            }

            if (!response.nextPageToken) {
                break;
            }
            pageToken = response.nextPageToken;
        }
    } catch (error) {
        console.log(`Encountered the following error while listing streams: ${error}`);
    }
}

App();

```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

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

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

	if latticeEndpoint == "" || environmentToken == "" || sandboxesToken == "" {
		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 := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// Iterate every page of active streams. Keep all request fields identical between
	// pages other than page_token; the server may reject a request that changes them.
	pageToken := ""
	for {
		resp, err := client.ListEgressStreams(ctx, &videomanagerv1.ListEgressStreamsRequest{
			PageToken: pageToken,
		})
		if err != nil {
			log.Fatalf("Failed to list egress streams: %v", err)
		}
		for _, stream := range resp.GetEgressStreams() {
			fmt.Printf("Egress %s re-publishes ingress %s\n", stream.GetEgressId(), stream.GetIngressId())
		}
		if resp.GetNextPageToken() == "" {
			break
		}
		pageToken = resp.GetNextPageToken()
	}
}

```

```py title={"Python (gRPC)"}
# 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.videomanager.v1.list_egress_streams_request_pub_pb import (
    ListEgressStreamsRequest,
)
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)


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(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Iterate every page of active streams. Keep all request fields identical between
    # pages other than page_token; the server may reject a request that changes them.
    page_token = ""
    while True:
        try:
            response = client.list_egress_streams(
                ListEgressStreamsRequest(page_token=page_token),
                timeout_ms=10000,
            )
        except Exception as error:
            print(f"Failed to list egress streams: {error}", file=sys.stderr)
            sys.exit(1)

        for stream in response.egress_streams:
            print(f"Egress {stream.egress_id} re-publishes ingress {stream.ingress_id}")

        if not response.next_page_token:
            break
        page_token = response.next_page_token


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// 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::videomanager::v1::ListEgressStreamsRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

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

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

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

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

    let mut client = VideoManagerApiClient::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)
        },
    );

    // Iterate every page of active streams. Keep all request fields identical between
    // pages other than page_token; the server may reject a request that changes them.
    let mut page_token = String::new();
    loop {
        let request = Request::new(ListEgressStreamsRequest {
            page_size: 0,
            page_token: page_token.clone(),
        });
        let response = client.list_egress_streams(request).await?.into_inner();

        for stream in &response.egress_streams {
            println!(
                "Egress {} re-publishes ingress {}",
                stream.egress_id,
                stream.ingress_id
            );
        }

        if response.next_page_token.is_empty() {
            break;
        }
        page_token = response.next_page_token;
    }

    Ok(())
}

```

## Get a single egress stream

Call `GetEgressStream` with an `egress_id` to retrieve a single `EgressStream`. The response includes the `ingress_id` it re-publishes and the protocol-specific connection details, either an `RtspEgress` or an `SrtEgress`, that a downstream consumer uses to pull the stream.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to retrieve a single egress stream.
package main

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

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

// Egress ID of the stream to get. Replace before running.
const egressID = "<EGRESS-ID>"

func main() {
	// Get environment variables
	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")

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

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

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

	ctx := context.Background()

	response, err := LatticeClient.Video.GetEgressStream(ctx, &Lattice.GetEgressStreamRequest{
		EgressID: egressID,
	})
	if err != nil {
		fmt.Printf("Failed to get egress stream: %v\n", err)
		os.Exit(1)
	}

	stream := response.GetEgressStream()
	if stream == nil {
		fmt.Printf("No egress stream found for %s\n", egressID)
		os.Exit(1)
	}
	fmt.Printf("Egress ID:  %s\n", *stream.GetEgressID())
	fmt.Printf("Ingress ID: %s\n", *stream.GetIngressID())
	// Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
	if stream.GetRtsp() != nil {
		fmt.Printf("RTSP URL:   %s\n", *stream.GetRtsp().GetURL())
	} else if stream.GetSrt() != nil {
		fmt.Printf("SRT URL:    %s\n", *stream.GetSrt().GetURL())
		fmt.Printf("Session ID: %s\n", *stream.GetSrt().GetSessionID())
	}
}

```

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

import com.anduril.Lattice;
import com.anduril.types.EgressStream;
import com.anduril.types.GetEgressStreamResponse;
import com.anduril.types.RtspEgress;
import com.anduril.types.SrtEgress;

/**
 * Retrieves a single egress stream with the Lattice SDK.
 */
public class GetEgressStream {
    // Egress ID of the stream to get. Replace before running.
    private static final String EGRESS_ID = "<EGRESS-ID>";

    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 and ensure URL has https://
        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 {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            GetEgressStreamResponse response = client.video().getEgressStream(EGRESS_ID);
            EgressStream stream = response.getEgressStream().orElse(null);
            if (stream == null) {
                System.err.println("No egress stream found for " + EGRESS_ID);
                System.exit(1);
            }
            System.out.println("Egress ID:  " + stream.getEgressId().orElse(""));
            System.out.println("Ingress ID: " + stream.getIngressId().orElse(""));
            // Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
            RtspEgress rtsp = stream.getRtsp().orElse(null);
            SrtEgress srt = stream.getSrt().orElse(null);
            if (rtsp != null) {
                System.out.println("RTSP URL:   " + rtsp.getUrl().orElse(""));
            } else if (srt != null) {
                System.out.println("SRT URL:    " + srt.getUrl().orElse(""));
                System.out.println("Session ID: " + srt.getSessionId().orElse(""));
            }
        } catch (Exception e) {
            System.err.println("Failed to get egress stream: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to retrieve a single egress
# stream.

import os
import sys

from anduril import Lattice

# Egress ID of the stream to get. Replace before running.
EGRESS_ID = "<EGRESS-ID>"


def main():
    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)

    client = Lattice(
        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}"},
    )

    try:
        response = client.video.get_egress_stream(EGRESS_ID)
    except Exception as error:
        print(f"Failed to get egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    stream = response.egress_stream
    if not stream:
        print(f"No egress stream found for {EGRESS_ID}", file=sys.stderr)
        sys.exit(1)
    print(f"Egress ID:  {stream.egress_id}")
    print(f"Ingress ID: {stream.ingress_id}")
    # Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
    if stream.rtsp:
        print(f"RTSP URL:   {stream.rtsp.url}")
    elif stream.srt:
        print(f"SRT URL:    {stream.srt.url}")
        print(f"Session ID: {stream.srt.session_id}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
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,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Egress ID of the stream to get. Replace before running.
const egressId = "<EGRESS-ID>";

async function App() {
    try {
        const response = await client.video.getEgressStream({ egressId });

        const stream = response.egressStream;
        if (!stream) {
            console.log(`No egress stream found for ${egressId}`);
            return;
        }
        console.log(`Egress ID:  ${stream.egressId}`);
        console.log(`Ingress ID: ${stream.ingressId}`);
        // Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
        if (stream.rtsp) {
            console.log(`RTSP URL:   ${stream.rtsp.url}`);
        } else if (stream.srt) {
            console.log(`SRT URL:    ${stream.srt.url}`);
            console.log(`Session ID: ${stream.srt.sessionId}`);
        }
    } catch (error) {
        console.log(`Encountered the following error while fetching the stream: ${error}`);
    }
}

App();

```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

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

// Egress ID of the stream to get. Replace before running.
const egressID = "<EGRESS-ID>"

func main() {
	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 := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	resp, err := client.GetEgressStream(ctx, &videomanagerv1.GetEgressStreamRequest{
		EgressId: egressID,
	})
	if err != nil {
		log.Fatalf("Failed to get egress stream: %v", err)
	}

	stream := resp.GetEgressStream()
	fmt.Printf("Egress ID:  %s\n", stream.GetEgressId())
	fmt.Printf("Ingress ID: %s\n", stream.GetIngressId())
	// Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
	if rtsp := stream.GetRtsp(); rtsp != nil {
		fmt.Printf("RTSP URL:   %s\n", rtsp.GetUrl())
	} else if srt := stream.GetSrt(); srt != nil {
		fmt.Printf("SRT URL:    %s\n", srt.GetUrl())
		fmt.Printf("Session ID: %s\n", srt.GetSessionId())
	}
}

```

```py title={"Python (gRPC)"}
# 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.videomanager.v1.get_egress_stream_request_pub_pb import (
    GetEgressStreamRequest,
)
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Egress ID of the stream to get. Replace before running.
EGRESS_ID = "<EGRESS-ID>"


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(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    try:
        response = client.get_egress_stream(
            GetEgressStreamRequest(egress_id=EGRESS_ID),
            timeout_ms=10000,
        )
    except Exception as error:
        print(f"Failed to get egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    if not response.has_field("egress_stream"):
        print(f"No egress stream found for {EGRESS_ID}", file=sys.stderr)
        sys.exit(1)
    stream = response.egress_stream
    print(f"Egress ID:  {stream.egress_id}")
    print(f"Ingress ID: {stream.ingress_id}")
    # Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
    if stream.has_field("rtsp"):
        print(f"RTSP URL:   {stream.egress.value.url}")
    elif stream.has_field("srt"):
        print(f"SRT URL:    {stream.egress.value.url}")
        print(f"Session ID: {stream.egress.value.session_id}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// 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::videomanager::v1::{
    egress_stream::Egress, GetEgressStreamRequest,
};
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Egress ID of the stream to get. Replace before running.
const EGRESS_ID: &str = "<EGRESS-ID>";

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

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

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

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

    let mut client = VideoManagerApiClient::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)
        },
    );

    let request = Request::new(GetEgressStreamRequest {
        egress_id: EGRESS_ID.to_string(),
    });
    let response = client.get_egress_stream(request).await?.into_inner();

    match response.egress_stream {
        Some(stream) => {
            println!("Egress ID:  {}", stream.egress_id);
            println!("Ingress ID: {}", stream.ingress_id);
            // Exactly one of rtsp or srt is set, matching the transport the egress stream uses.
            match stream.egress {
                Some(Egress::Rtsp(rtsp)) => println!("RTSP URL:   {}", rtsp.url),
                Some(Egress::Srt(srt)) => {
                    println!("SRT URL:    {}", srt.url);
                    println!("Session ID: {}", srt.session_id);
                }
                None => {}
            }
        }
        None => eprintln!("No egress stream found for {}", EGRESS_ID),
    }

    Ok(())
}

```

## Stop an egress stream

Call `DeleteEgressStream` with an `egress_id` to stop delivering a stream to its downstream consumers. Deleting an egress stream stops only that downstream delivery; the underlying ingress stream and any other egress streams consuming it are unaffected.

> **Egress streams stop with their source**
>
> Deleting an `IngressStream` automatically stops every egress stream consuming it, so you don't need to delete egress streams individually before stopping their source. `DeleteEgressStream` returns `NOT_FOUND` if no matching active egress stream exists.

```go title={"Go (REST)"}
// This Go example uses the Lattice REST SDK (lattice-sdk-go) to stop an egress stream,
// which halts delivery to its downstream consumers.
package main

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

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

// Egress ID of the stream to stop. Replace before running.
const egressID = "<DELETE-EGRESS-ID>"

func main() {
	// Get environment variables
	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")

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

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

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

	ctx := context.Background()

	// Deleting an egress stream stops only that downstream delivery. The underlying ingress
	// stream and any other egress streams consuming it are unaffected. Deleting an egress
	// stream that does not exist returns NOT_FOUND.
	fmt.Printf("Stopping egress stream with ID: %s\n", egressID)
	_, err := LatticeClient.Video.DeleteEgressStream(ctx, &Lattice.DeleteEgressStreamRequest{
		EgressID: egressID,
	})
	if err != nil {
		fmt.Printf("Failed to stop egress stream: %v\n", err)
		os.Exit(1)
	}

	fmt.Printf("Stopped egress stream %s\n", egressID)
}

```

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

import com.anduril.Lattice;

/**
 * Stops an egress stream with the Lattice SDK, which halts delivery to its downstream consumers.
 */
public class DeleteEgressStream {
    // Egress ID of the stream to stop. Replace before running.
    private static final String EGRESS_ID = "<DELETE-EGRESS-ID>";

    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 and ensure URL has https://
        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 {
            // Create Lattice client with Sandbox authentication
            Lattice client = Lattice.builder()
                .url(endpoint)
                .token(environmentToken)
                .addHeader("Anduril-Sandbox-Authorization", "Bearer " + sandboxesToken)
                .build();

            // Deleting an egress stream stops only that downstream delivery. The underlying ingress
            // stream and any other egress streams consuming it are unaffected, and no entity or
            // media reference is touched. Deleting an egress stream that does not exist returns
            // NOT_FOUND.
            System.out.println("Stopping egress stream with ID: " + EGRESS_ID);
            client.video().deleteEgressStream(EGRESS_ID);

            System.out.println("Stopped egress stream " + EGRESS_ID);
        } catch (Exception e) {
            System.err.println("Failed to stop egress stream: " + e.getMessage());
            System.exit(1);
        }
    }
}

```

```py title={"Python (REST)"}
# This Python example uses the Lattice REST SDK (anduril-lattice-sdk) to stop an egress stream,
# which halts delivery to its downstream consumers.

import os
import sys

from anduril import Lattice

# Egress ID of the stream to stop. Replace before running.
EGRESS_ID = "<DELETE-EGRESS-ID>"


def main():
    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)

    client = Lattice(
        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}"},
    )

    # Deleting an egress stream stops only that downstream delivery. The underlying ingress
    # stream and any other egress streams consuming it are unaffected. Deleting an egress
    # stream that does not exist returns NOT_FOUND.
    print(f"Stopping egress stream with ID: {EGRESS_ID}")
    try:
        client.video.delete_egress_stream(EGRESS_ID)
    except Exception as error:
        print(f"Failed to stop egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Stopped egress stream {EGRESS_ID}")


if __name__ == "__main__":
    main()

```

```ts title={"Typescript (REST)"}
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,
        // Remove the following if you are not developing on Sandboxes.
        headers: {  "Anduril-Sandbox-Authorization": `Bearer ${sandboxesToken}`  }
    }
);

// Egress ID of the stream to stop. Replace before running.
const egressId = "<DELETE-EGRESS-ID>";

async function App() {
    try {
        // Deleting an egress stream stops only that downstream delivery. The underlying ingress
        // stream and any other egress streams consuming it are unaffected, and no entity's media
        // list is touched. Deleting an egress stream that does not exist returns NOT_FOUND.
        console.log(`Stopping egress stream with ID: ${egressId}`);
        await client.video.deleteEgressStream({ egressId });
        console.log(`Stopped egress stream ${egressId}`);
    } catch (error) {
        console.log(`Encountered the following error while stopping the stream: ${error}`);
    }
}

App();

```

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

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

	"schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/grpc/go/anduril/videomanager/v1/videomanagerv1grpc"
	videomanagerv1 "schema-registry.developer.anduril.com/gen/go/anduril/lattice-sdk/protocolbuffers/go/anduril/videomanager/v1"

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

// Egress ID of the stream to stop. Replace before running.
const egressID = "<DELETE-EGRESS-ID>"

func main() {
	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 := videomanagerv1grpc.NewVideoManagerAPIClient(conn)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// Deleting an egress stream stops only that downstream delivery. The underlying
	// ingress stream and any other egress streams consuming it are unaffected, and no
	// entity or media is touched. Deleting an egress stream that does not exist returns
	// NOT_FOUND.
	fmt.Printf("Stopping egress stream with ID: %s\n", egressID)
	_, err = client.DeleteEgressStream(ctx, &videomanagerv1.DeleteEgressStreamRequest{
		EgressId: egressID,
	})
	if err != nil {
		log.Fatalf("Failed to stop egress stream: %v", err)
	}
	fmt.Printf("Stopped egress stream %s\n", egressID)
}

```

```py title={"Python (gRPC)"}
# 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.videomanager.v1.delete_egress_stream_request_pub_pb import (
    DeleteEgressStreamRequest,
)
from anduril.videomanager.v1.video_manager_api_pub_connect import (
    VideoManagerAPIClientSync,
)

# Egress ID of the stream to stop. Replace before running.
EGRESS_ID = "<DELETE-EGRESS-ID>"


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(VideoManagerAPIClientSync, lattice_endpoint, environment_token, sandboxes_token)

    # Deleting an egress stream stops only that downstream delivery. The underlying ingress
    # stream and any other egress streams consuming it are unaffected, and no entity or
    # media is touched. Deleting an egress stream that does not exist returns NOT_FOUND.
    print(f"Stopping egress stream with ID: {EGRESS_ID}")
    try:
        client.delete_egress_stream(
            DeleteEgressStreamRequest(egress_id=EGRESS_ID),
            timeout_ms=30000,
        )
    except Exception as error:
        print(f"Failed to stop egress stream: {error}", file=sys.stderr)
        sys.exit(1)

    print(f"Stopped egress stream {EGRESS_ID}")


if __name__ == "__main__":
    main()

```

```rs title={"Rust (gRPC)"}
// 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::videomanager::v1::DeleteEgressStreamRequest;
use anduril_lattice_sdk_community_neoeinstein_tonic::anduril::videomanager::v1::tonic::video_manager_api_client::VideoManagerApiClient;
use bearer_auth::EnvironmentTokenAuth;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::Request;

use std::env;

// Egress ID of the stream to stop. Replace before running.
const EGRESS_ID: &str = "<DELETE-EGRESS-ID>";

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

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

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

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

    let mut client = VideoManagerApiClient::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)
        },
    );

    // Deleting an egress stream stops only that downstream delivery. Unlike deleting an
    // ingress stream, it never touches an entity or its media; the underlying ingress
    // stream and any other egress streams consuming it are unaffected. Deleting an egress
    // stream that does not exist returns NOT_FOUND.
    println!("Stopping egress stream with ID: {}", EGRESS_ID);
    client
        .delete_egress_stream(Request::new(DeleteEgressStreamRequest {
            egress_id: EGRESS_ID.to_string(),
        }))
        .await?;

    println!("Stopped egress stream {}", EGRESS_ID);

    Ok(())
}

```

## What's next

* [Overview](/guides/video/overview): Review the API surface and stream lifecycle.
* [Create streams](/guides/video/create-streams): Publish an asset and start an ingress stream.