Runtime SDK reference

Applies to VAST Cluster 5.5.

The DataEngine Runtime SDK is the Python package your function code imports. It defines the two functions the runtime calls (init and handler), the Context object passed to both, the event classes, and helpers for tracing and metrics. This page lists them for VAST Cluster 5.5. For a walkthrough, see write a function. For the attributes each event carries, see events.

Packages and imports

Import everything from vast.dataengine.sdk:

from vast.dataengine.sdk import (
    Context,
    VastEvent,
    VastEventList,
    ElementTriggerVastEvent,
    ScheduleVastEvent,
    FunctionVastEvent,
    ManualVastEvent,
    event_batching_handler,
)
from vast.dataengine.sdk.tracing import span_wrapped, event_span_wrapped

The tracing decorators live in vast.dataengine.sdk.tracing. Importing them from vast.dataengine.sdk fails with the 5.5 package.

How the SDK gets into your image. You don't install it yourself. vastde functions build uses the builder image, which carries the SDK and the DataEngine runtime and installs them into every function image it builds. That's why the requirements.txt that vastde functions init creates doesn't list the SDK. List only your own dependencies there.

A local copy for your editor or unit tests. The getting-started instructions install the SDK with pip install vast-dataengine-sdk. The package isn't on the public Python Package Index, so that command works only if your pip is set up with an index that carries it. Ask your platform team whether they host one. You don't need a local copy to build, run or deploy a function.

Older examples import from vast_runtime (for example from vast_runtime.vast_event import VastEvent). The 5.5 runtime still accepts those imports, with a deprecation warning. Use vast.dataengine.sdk in new code.

Handler contract

Your handler file (main.py unless you pass -H to vastde functions init and vastde functions build) defines two top-level functions:

Define both. vastde functions build checks the handler file and refuses to build it without a top-level init and handler, even though the runtime itself would run without init.

from vast.dataengine.sdk import Context, VastEvent

def init(ctx: Context):
    ctx.logger.info("Initialized")

async def handler(ctx: Context, event: VastEvent):
    ctx.logger.info(f"Received {event.type} event {event.id}")
    return {"seen": event.id}

init runs once per instance; handler runs per event or batch; an exception means a retry, then the dead-letter topic.

init

Use init for one-time work: reading secrets, creating clients and metric instruments, and setting your own attributes on ctx (for example ctx.db = connect(...)) for the handler to use. Attributes you set in init are visible in every later handler call on that instance.

An instance doesn't receive events until init has returned. If init raises, the instance never becomes ready. So fail fast in init on missing configuration, and check the function's logs if a deployment never becomes ready.

init can be async def too.

Sync and async handlers

handler can be def or async def. Use async def for throughput.

• async def handlers run on the runtime's event loop. Don't call blocking code (blocking I/O, time.sleep, CPU-heavy work) directly in them: it stalls every event the instance is processing. Use async libraries, or asyncio.to_thread.
• def handlers run in a pool of worker threads, so several events can be in your handler at the same time. Anything you share through ctx or module globals must be safe to use from several threads. When all the threads are busy, the instance refuses further events until one finishes, and they're delivered again later.

Batch handlers

When the function deployment has batching configured (config.event_batching in the pipeline YAML), the runtime collects events until the batch is full or the batch timeout passes, then calls your handler once with all of them. Mark the handler with @event_batching_handler so the runtime passes a VastEventList instead of a single event:

from vast.dataengine.sdk import Context, VastEventList, event_batching_handler

def init(ctx: Context):
    ctx.logger.info("Initialized")

@event_batching_handler
def handler(ctx: Context, events: VastEventList):
    payloads = [event.get_data() for event in events]
    write_all(payloads)                      # your bulk operation
    return [{"stored": True} for _ in payloads]

The rule for batch handlers: return a list of exactly the same length as the batch. Each element is sent downstream as its own event, in the same order as the input. Returning anything else (a list of another length, or not a list) fails the whole batch.

VastEventList supports len(), iteration and indexing. To slice it, convert it first: list(events)[:10].

Context

ctx is a Context. The runtime creates one per instance and passes the same object to init and every handler call.

pipeline_triggers_map exists only in a deployed pipeline. Reading it in a local run (vastde functions localrun) raises FileNotFoundError.

Metrics

Create instruments in init and record values in handler:

def init(ctx):
    global processed, latency
    processed = ctx.counter("files.processed", description="Files processed")
    latency = ctx.histogram("file.latency", unit="ms", boundaries=[10, 50, 100, 500, 1000])

def handler(ctx, event):
    ...
    if processed:
        processed.add(1, {"bucket": "media"})
    if latency:
        latency.record(elapsed_ms)

Rules:

• Each helper returns None instead of an instrument when it can't create one, and logs the reason. Check the result before you use it, as above.
• Names follow the OpenTelemetry instrument-name syntax: they start with a letter, contain only letters, digits, _, ., - and /, and are at most 255 characters long.
• These names are reserved for the metrics DataEngine records automatically, and the helpers refuse them: dataengine.function.invocations, dataengine.function.errors, dataengine.function.invoke_duration, dataengine.event.total and dataengine.event.reception_latency. vastde metrics names <pipeline> lists the ones a pipeline has recorded.
• Calling a helper again with the same name returns the same instrument. A name that's already used by another instrument type is refused.
• boundaries must be a non-empty list of increasing numbers, or the call raises ValueError.

You can chart custom metrics in the DataEngine dashboard, or read them with vastde metrics get. See observability.

Secrets

ctx.secrets is a nested dictionary: the top-level key is the name of the secret, and its value maps each key in that secret to its value as a string. A secret named llm-creds with the keys API_KEY and ENDPOINT appears as {"llm-creds": {"API_KEY": "...", "ENDPOINT": "..."}}.

creds = ctx.secrets.get("llm-creds", {})
api_key = creds.get("API_KEY", "")
if not api_key:
    raise RuntimeError("secret llm-creds has no API_KEY")

If no secrets are attached, ctx.secrets is an empty dictionary. The dictionary is keyed by secret name, so ctx.secrets.get("secrets", {}) reads a secret named secrets. Indexing a secret that isn't attached (ctx.secrets["thumbnail-api"]) raises KeyError, which fails the event like any other exception.

Secrets are Kubernetes secrets in the pipeline's namespace, mounted into the function's container. You attach them by name in the pipeline YAML, at the pipeline level or per function deployment, but not both: naming the same secret at both levels is rejected when you create the pipeline. A secret that doesn't exist in the namespace is accepted when you create the pipeline and fails the deploy. See security and the error reference.

Log secret names and key names if you need to check what's attached, never values:

ctx.logger.info("secrets: %s", {name: sorted(keys) for name, keys in ctx.secrets.items()})

In a deployed pipeline with one attached secret, that logged secrets: {'thumbnail-api': ['api-token']}.

Locally, vastde functions localrun --config <file> mounts the secrets listed in the file, but only in the foreground. With --detach, the function sees ctx.secrets as {}.

Environment variables

Read environment variables with os.environ. You set them in the pipeline YAML, for the whole pipeline (manifest.config.environment_variables) or for one function deployment (config.environment_variables). A function-level value overrides a pipeline-level one with the same name. Values arrive as strings. Use them for settings that aren't sensitive and change between deployments, such as bucket names, model names and batch sizes.

The only environment variables you can rely on are the ones you set. DataEngine also sets its own variables in the container, but they're not part of the SDK contract.

Some names are reserved, and vastde pipelines create rejects the pipeline if you set them: names that start with VAST_, VASTPIPELINE_ or K_, and the names HOSTNAME, PERFORMANCE and PORT. The error lists them:

pipeline creation failed: The following environment variable names are reserved for internal use and cannot be set by users: VAST_ENDPOINT. Reserved prefixes: VASTPIPELINE_, K_, VAST_. Reserved names: HOSTNAME, PERFORMANCE, PORT, VAST_RUNTIME_HANDLER_THREADS

That includes the VAST_ENDPOINT, VAST_ACCESS_KEY, VAST_SECRET_KEY, VAST_BUCKET and VAST_SCHEMA names for database functions, so use other names, for example DB_ENDPOINT and DB_BUCKET, and keep the keys in a secret.

VastEvent

Every event your handler receives is a VastEvent, or one of its subclasses, chosen by the event's type. A VastEvent is a CloudEvents v1.0 event (it extends the cloudevents package's CloudEvent), so event["attribute"] and event.get_attributes() work as well. The attributes behind each property are listed in events.

Subclasses

Use the casting helpers when a function can receive more than one type:

def handler(ctx, event):
    try:
        element = event.as_element_event()
    except TypeError:
        ctx.logger.warning(f"Ignoring {event.type} event {event.id}")
        return None
    ctx.logger.info(f"New object s3://{element.bucket}/{element.object_key}")

A property that can't be read from the event raises ValueError("malformed vast event") for the base properties. The element and schedule properties return None instead, and log a malformed vast event error. That's what you see locally with vastde functions invoke --generate-event, whose sample event has no elementpath; see test events locally.

Conditional trigger labels

event.set_trigger_labels({"key": "value"}) attaches labels to the event that your handler returns. A link from this function with conditional_trigger_labels passes the result on only if every label on the link matches. The rules:

• Calls merge: a second call adds keys and updates existing ones.
• An empty dictionary raises ValueError. To send a result nowhere, set a label that no link matches.
• Routing is hop by hop: labels are matched only against links that start at this function.

Label names and values must match strict patterns; see pipeline YAML. Labels must be set on the event you received, before the handler returns. The runtime reads them from there, not from the value you return.

VastEventList

A VastEventList is a list of VastEvent objects, passed to batch handlers.

Tracing

The runtime creates a trace for each event, with a span for the handler call. Each instance's init call gets a trace of its own, so the first event on a new instance shows up next to an init trace in vastde traces list. Batches get their own trace, and each event in the batch keeps its own trace too. Add your own spans in three ways:

from vast.dataengine.sdk.tracing import span_wrapped, event_span_wrapped

# 1. A context manager
def handler(ctx, event):
    with ctx.tracer.start_as_current_span("parse") as span:
        span.set_attribute("payload.size", len(str(event.get_data())))

# 2. A span attached to a specific event's trace (useful in batch handlers)
def handle_one(ctx, event):
    with ctx.start_as_current_event_span(event, "handle-one"):
        ...

# 3. Decorators
@span_wrapped("transform")          # or bare @span_wrapped: the span is named after the function
def transform(data):
    return str(data).upper()

@event_span_wrapped("handle-upload", attributes={"stage": "ingest"})
def handle_upload(ctx, event):      # finds the parameter named "event"; change it with event_param="..."
    return transform(event.get_data())

• event_span_wrapped raises ValueError if it can't find the event parameter, and TypeError if that parameter isn't a VastEvent.
• Both decorators wrap ordinary functions. On an async def, the span ends before the coroutine runs, so in async code use with ctx.tracer.start_as_current_span(...) inside the function instead.
• Span attribute names that start with event., vastpipeline., vastfunction., service. or tenant. are reserved for DataEngine. Setting one raises ValueError.

S3Client

The SDK includes a thin wrapper around a boto3 S3 client that reads its configuration from a secret and environment variables. It isn't documented, so you may prefer to create a boto3 client yourself from ctx.secrets, as the KB's secrets page does.

from vast.dataengine.sdk.s3 import S3Client, S3ClientError

def init(ctx):
    ctx.s3 = S3Client(ctx.secrets)          # nothing is read until first use

def handler(ctx, event):
    element = event.as_element_event()
    obj = ctx.s3.get_object(Bucket=element.bucket, Key=element.object_key)
    ...

• Construct it with S3Client(ctx.secrets). There is no ctx.s3 until you set one, although the class's own docstring uses ctx.s3.
• It creates the boto3 client on first use. Call ctx.s3.init(access_key=..., secret_key=..., endpoint_url=..., region=...) to create it early or to override any of the settings.
• Every other attribute is passed through to the boto3 client, so ctx.s3.list_objects_v2(...) and the rest work as in boto3.
• boto3 isn't installed for you: add it to requirements.txt. Without it, the first call raises S3ClientError. So do a missing secret, key or endpoint.
• AsyncS3Client (same module) is the aioboto3 equivalent for async def handlers; it needs aioboto3 in requirements.txt.

The keys belong to a VAST user with access to the buckets your function reads and writes. See security.

Return values

What your handler returns is not sent back to whoever delivered the event: the delivery gets an empty success response. It becomes the data of a new Function event, which DataEngine sends along each link that starts at this function deployment.

• Return a value that serialises to JSON: a dict, list, str, number or None. The runtime wraps a dict or str in an event for you.
• If no link starts at this function, the return value is discarded.
• None is still sent downstream, as an event with no data. To stop an event, use a conditional trigger label that no link matches.
• A batch handler returns a list, and each element becomes its own event.
• The runtime writes each return value to the function's logs, at INFO in the runtime scope. Don't return secrets.

What the downstream function receives is listed in events.

Errors and retries

If your handler raises an exception, the event's delivery fails. The runtime logs the exception, marks the trace as an error, and counts it in dataengine.function.errors. The log record is at ERROR in the runtime scope and carries your exception's message:

2026-10-08 12:53:07.18 [faulty] [ERROR] [vast-runtime] Error in handler execution: cannot process fail/broken.txt

DataEngine then delivers the event again, up to the number of retries set on the link that delivered it (links[].config.retries in the pipeline YAML), with an increasing delay between attempts. When the retries run out, the event goes to the tenant's dead-letter topic, which receives "function failures, routing failures and dispatching failures". The dead-lettered record is the original event, with the error message added in a header; see retries and the dead-letter topic.

Things to know:

• A retry runs your handler again on the same event, so write handlers that can safely run twice.
• If the handler succeeds but the result can't be sent downstream, the delivery also fails and is retried, so your handler runs again.
• In a batch handler, an exception fails every event in the batch.
• The runtime doesn't time out your handler itself. The function deployment's timeout (resources.timeout, in seconds) applies, and retries also cover timeouts.

Logging

The runtime sends three kinds of output to DataEngine logs, correlated with the event's trace:

• Records from ctx.logger, with the usual levels (ctx.logger.info(...), ctx.logger.error(...)).
• print() output, as INFO records. Output printed to sys.stderr isn't captured.
• Records from standard Python loggers (logging.getLogger(__name__)), including your libraries' loggers.

The function deployment's log level (config.log_level in the pipeline YAML) decides which records are kept. Read the logs with vastde logs get or vastde logs tail, or in the Web UI. Your records have the scope user; the runtime's own records have the scope runtime (shown as vast-runtime in the UI and in vastde logs get output). See observability.

See also

• Events: the attributes of each event type
• Write a function
• Build and test locally
• Pipeline YAML: batching, retries, log level, secrets and environment variables
• KB: Runtime SDK Developer's Guide, including the class reference