VSS VST Tool Implementations for Timeline, Sensor List, and Duration Coverage

The Video Search & Summarization (VSS) agent exposes three async LangChain-wrapped tools—vst.timeline, vst.sensor_list, and vst.duration—that query the Video Streaming Toolkit (VST) to retrieve temporal boundaries, discover available sensors, and calculate video durations through dedicated pydantic models and resilient HTTP utilities.

The NVIDIA AI Blueprint for video-search-and-summarization provides a modular agent framework that interfaces with the Video Streaming Toolkit (VST) through specialized tool implementations. These tools enable LLM-driven workflows to dynamically discover camera streams, retrieve temporal metadata, and calculate exact video durations before performing analysis. This article examines the concrete implementations of the timeline, sensor list, and duration tools located in the agent/src/vss_agents/tools/vst directory.

Timeline Tool: Retrieving Video Stream Boundaries

The vst.timeline tool provides the exact start and end timestamps for any video stream stored in VST. This functionality is essential for framing temporal queries and ensuring summarization requests stay within valid time windows.

Core Implementation in timeline.py

The implementation resides in agent/src/vss_agents/tools/vst/timeline.py and follows a strict separation of concerns:

• Config class: VSTTimelineConfig holds static parameters including vst_internal_url, which defaults to the VST_INTERNAL_URL environment variable.
• Input model: VSTTimelineInput requires a sensor_id parameter that accepts either a human-readable sensor name or a canonical stream ID.
• Output model: VSTTimelineOutput returns ISO-8601 formatted start_timestamp and end_timestamp strings.
• Core function: _vst_timeline is the async generator that orchestrates the lookup.

The execution flow delegates HTTP communication to agent/src/vss_agents/tools/vst/utils.py. The get_stream_id helper resolves ambiguous sensor names to canonical stream IDs by querying /vst/api/v1/sensor/streams. Subsequently, get_timeline calls /vst/api/v1/storage/timelines, extracts the first entry’s temporal boundaries, and validates that the duration is at least one second to prevent degenerate cases.

# From agent/src/vss_agents/tools/vst/timeline.py

@register_function(
    config_class=VSTTimelineConfig,
    framework_wrappers=[LLMFrameworkEnum.LANGCHAIN],
    response_tags=[TABULAR_RESPONSE_TAG],
)
async def _vst_timeline(config: VSTTimelineConfig, builder: Builder):
    """Fetch start and end timestamps for a VST stream."""
    # Implementation yields a single_fn that accepts VSTTimelineInput

The get_timeline utility implements exponential backoff via create_retry_strategy (from vss_agents.utils.retry), ensuring resilience against transient VST unavailability.

Sensor List Tool: Discovering Available Cameras

The vst.sensor_list tool enables dynamic discovery of all cameras currently registered with the VST service. This supports UI population and automated sensor selection workflows where the agent must present valid options to users.

Implementation in sensor_list.py

Located in agent/src/vss_agents/tools/vst/sensor_list.py, this tool requires no input parameters, as reflected by the empty VSTSensorListInput model.

# From agent/src/vss_agents/tools/vst/sensor_list.py

class VSTSensorListInput(BaseModel):
    """No inputs required for listing sensors."""
    pass

The core logic relies on get_name_to_stream_id_map from utils.py, which queries /vst/api/v1/sensor/streams and constructs a mapping of sensor names to stream IDs. The tool sorts these names alphabetically and returns them via VSTSensorListOutput, making it predictable for downstream prompt construction.

This tool is frequently invoked before timeline or duration calls to validate that a requested sensor actually exists in the VST registry.

Duration Tool: Calculating Video Length

The vst.duration tool computes the total length of a video stream in seconds by leveraging the same temporal metadata retrieved by the timeline tool. It acts as a thin computational wrapper around VST’s timeline data.

Implementation in duration.py

The agent/src/vss_agents/tools/vst/duration.py file defines VSTDurationConfig, VSTDurationInput (requiring a sensor_id), and VSTDurationOutput (returning a duration float).

The async function _vst_duration reuses get_stream_id and get_timeline from the shared utilities to fetch the start and end timestamps. It converts the ISO-8601 strings to Python datetime objects, computes the delta, and returns the total seconds as a float. By inheriting the retry logic from get_timeline, this tool maintains the same reliability guarantees without duplicating HTTP handling code.

# Conceptual flow from duration.py implementation

start = datetime.fromisoformat(timeline_data["startTime"])
end = datetime.fromisoformat(timeline_data["endTime"])
duration_seconds = (end - start).total_seconds()

Shared Infrastructure and Utilities

All three tools rely on agent/src/vss_agents/tools/vst/utils.py for low-level VST interaction.

HTTP Helpers and Retry Logic

The utils.py module provides three critical functions:

• get_name_to_stream_id_map: Queries the VST sensor streams endpoint and returns a dictionary mapping sensor names to stream IDs.
• get_stream_id: Resolves a potentially ambiguous identifier to a canonical stream ID, raising VSTError if resolution fails.
• get_timeline: Performs the actual HTTP GET to /vst/api/v1/storage/timelines, applies retry logic via create_retry_strategy, validates the JSON response, and returns the raw timeline data.

The retry strategy applies exponential backoff for transient HTTP failures, preventing the agent from failing on momentary network blips.

LangChain Registration

Each tool is registered with the Nat-builder framework using the @register_function decorator, specifying LLMFrameworkEnum.LANGCHAIN in the framework_wrappers list. This registration makes the tools automatically discoverable as LangChain Tool objects, allowing LLM agents to invoke them through standard ainvoke or run methods.

Practical Implementation Examples

Direct Async Usage

The following example demonstrates invoking all three tools directly using their async generators:

import asyncio
from vss_agents.tools.vst.timeline import _vst_timeline, VSTTimelineConfig, VSTTimelineInput
from vss_agents.tools.vst.sensor_list import _vst_sensor_list, VSTSensorListConfig
from vss_agents.tools.vst.duration import _vst_duration, VSTDurationConfig, VSTDurationInput
from nat.builder.builder import Builder

async def demo_vst_tools():
    builder = Builder()
    base_url = "http://localhost:30888"
    
    # 1. Discover all sensors

    sensor_cfg = VSTSensorListConfig(vst_internal_url=base_url)
    async for fn_info in _vst_sensor_list(sensor_cfg, builder):
        sensors = await fn_info.single_fn(input_data=None)
        print(f"Available sensors: {sensors.sensor_names}")
    
    # 2. Get timeline for specific sensor

    timeline_cfg = VSTTimelineConfig(vst_internal_url=base_url)
    async for fn_info in _vst_timeline(timeline_cfg, builder):
        timeline = await fn_info.single_fn(VSTTimelineInput(sensor_id="Camera_01"))
        print(f"Timeline: {timeline.start_timestamp} to {timeline.end_timestamp}")
    
    # 3. Calculate duration

    duration_cfg = VSTDurationConfig(vst_internal_url=base_url)
    async for fn_info in _vst_duration(duration_cfg, builder):
        dur = await fn_info.single_fn(VSTDurationInput(sensor_id="Camera_01"))
        print(f"Duration: {dur.duration} seconds")

asyncio.run(demo_vst_tools())

LangChain Agent Integration

When running inside the Nat-builder runtime, these tools can be wrapped for LangChain agents:

from langchain.agents import initialize_agent, Tool
from langchain.chat_models import ChatOpenAI
from nat.builder.builder import Builder

builder = Builder()  # Provided by Nat runtime

# Wrappers automatically handle async invocation

tools = [
    Tool(
        name="vst.sensor_list",
        func=lambda _: builder.get_tool("vst.sensor_list").ainvoke(input={}),
        description="Returns a sorted list of available sensor names."
    ),
    Tool(
        name="vst.timeline",
        func=lambda sid: builder.get_tool("vst.timeline").ainvoke(input={"sensor_id": sid}),
        description="Returns start and end timestamps for a given sensor."
    ),
    Tool(
        name="vst.duration",
        func=lambda sid: builder.get_tool("vst.duration").ainvoke(input={"sensor_id": sid}),
        description="Returns video duration in seconds for a given sensor."
    )
]

agent = initialize_agent(
    tools=tools,
    llm=ChatOpenAI(model="gpt-4o-mini"),
    agent="zero-shot-react-description",
    verbose=True
)

# The agent can now autonomously discover sensors and analyze durations

agent.run("How long is the video from the WarehouseCam_7 sensor?")

Summary

• vst.timeline (in agent/src/vss_agents/tools/vst/timeline.py) retrieves ISO-8601 start and end timestamps by wrapping get_timeline from utils.py, which queries /vst/api/v1/storage/timelines.
• vst.sensor_list (in agent/src/vss_agents/tools/vst/sensor_list.py) returns alphabetically sorted sensor names using get_name_to_stream_id_map, which queries /vst/api/v1/sensor/streams.
• vst.duration (in agent/src/vss_agents/tools/vst/duration.py) calculates video length in seconds by reusing timeline data and computing datetime deltas.
• All tools use pydantic models for type-safe inputs and outputs, async/await for non-blocking I/O, and exponential backoff retry strategies for resilience.
• Registration via @register_function with LLMFrameworkEnum.LANGCHAIN makes these tools immediately available to LangChain agents orchestrated by the Nat-builder framework.

Frequently Asked Questions

How does the VSS agent resolve sensor names to stream IDs?

The get_stream_id function in agent/src/vss_agents/tools/vst/utils.py queries the VST endpoint /vst/api/v1/sensor/streams to build a mapping of names to canonical IDs. If the provided identifier matches a name in this mapping, the corresponding stream ID is returned; otherwise, the identifier is assumed to already be a valid stream ID.

What happens if the VST service returns a timeline shorter than one second?

The get_timeline helper explicitly validates that the duration between startTime and endTime is at least one second. If VST returns timestamps indicating a shorter or invalid duration, the function raises a VSTError, preventing the agent from processing degenerate video segments.

Can these tools be used outside of the Nat-builder framework?

Yes. While the tools are registered via @register_function for Nat-builder discovery, the underlying async functions (_vst_timeline, _vst_sensor_list, _vst_duration) can be imported directly and invoked by providing a Builder instance and the appropriate config objects, as shown in the direct async usage example.

How is the VST service URL configured across environments?

Each tool’s config class (VSTTimelineConfig, VSTSensorListConfig, VSTDurationConfig) accepts a vst_internal_url parameter. If not explicitly provided, these classes default to the VST_INTERNAL_URL environment variable, allowing the same code to target local development instances or production VST deployments without modification.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}