Services API — Vega SDK Reference

Import

python

from vega_capability_sdk import (
    ArtifactReference,
    ArtifactService,
    ArtifactWriter,
    Clock,
    ResourceLeaseView,
    StructuredLogger,
    SystemClock,
    TraceContext,
)

These names are also exported from vega_capability_sdk.services.require_utc and redact are lower-level module helpers and are not exported from the package-level services.__all__.

Artifacts — ArtifactReference

python

@dataclass(frozen=True, slots=True)
class ArtifactReference:
    artifact_id: str
    mime_type: str
    logical_role: str
    size_bytes: int
    checksum_sha256: str
    uri: str | None = None

    def to_dict(self) -> dict[str, str | int | None]: ...

Portable description of an Artifact. to_dict() returns every field using the same snake-case names. The dataclass does not validate identifier format, checksum format, negative sizes, or URI syntax. The Artifact service implementation is responsible for supplying consistent values.

ArtifactWriter

Runtime-checkable asynchronous protocol:

python

async def __aenter__() -> Self: ...
async def __aexit__(exc_type, exc, traceback) -> None: ...
async def write(data: bytes) -> None: ...
async def finalize() -> ArtifactReference: ...
async def abort() -> None: ...

Use finalize() to obtain a reference, or abort() to discard staged data. The protocol does not prescribe whether repeated finalize, abort, orwrite calls are accepted; consult the injected implementation.

ArtifactService

python

def create(*, mime_type: str, logical_role: str) -> ArtifactWriter: ...

An Artifact service is injected as an optional context field. The SDK does not provide a production storage implementation. mime_type and logical_role are keyword-only. create() returns a staged writer immediately; writing and finalization are asynchronous.

Time — Clock, SystemClock, require_utc

Term	Meaning
Clock	Runtime-checkable protocol with now() -> datetime. Must return a timezone-aware UTC timestamp.
SystemClock	Concrete clock whose now() returns the current UTC time.
require_utc(value, *, field_name)	Raises ValueError for a naive datetime; otherwise returns value normalized to UTC. Importable from vega_capability_sdk.services.clock.

Structured Logging — StructuredLogger

python

class StructuredLogger:
    def __init__(self, logger: logging.Logger | None = None, *, bindings=None) -> None: ...
    def bind(self, **values: Any) -> StructuredLogger: ...
    def debug(self, message: str, **fields: Any) -> None: ...
    def info(self, message: str, **fields: Any) -> None: ...
    def warning(self, message: str, **fields: Any) -> None: ...
    def error(self, message: str, **fields: Any) -> None: ...
    def exception(self, message: str, **fields: Any) -> None: ...

bind returns a new logger with merged bindings. Each logging method submits the merged binding and field mapping as extra={"vega": ...} to the Python logger. exception also records exception information. Values identified asSecretValue, and secrets nested in mappings, lists, or tuples, are redacted.

redact(value: Any) -> Any is additionally public fromvega_capability_sdk.services.logging. It returns **redacted** for values marked with __vega_secret__, recursively converts mappings to dictionaries and lists or tuples to lists. Redaction recognizes secret wrappers; it cannot detect a raw string obtained from get_secret_value().

Resource Leases — ResourceLeaseView

python

@dataclass(frozen=True, slots=True)
class ResourceLeaseView:
    resource_id: str
    mode: str
    lease_id: str | None = None
    expires_at: datetime | None = None

Read-only resource lease information supplied on an execution context. No constructor validation is performed. In particular, mode is a string rather than the manifestResourceMode enum, and expires_at is not normalized by this dataclass.

Tracing — TraceContext

python

@dataclass(frozen=True, slots=True)
class TraceContext:
    trace_id: str | None = None
    span_id: str | None = None
    trace_flags: str | None = None
    baggage: Mapping[str, str] = {}

    def as_headers(self) -> Mapping[str, str]: ...

baggage is copied into an immutable mapping. as_headers() addstraceparent only when both trace and span IDs are present, using flags00 when trace_flags is absent. It adds baggage as comma-separated key=value pairs when baggage is non-empty. The returned mapping is immutable.

The class does not validate W3C trace identifier lengths, hexadecimal syntax, trace flags, or baggage escaping. Only propagate trusted, already-normalized values.

Examples

python

if context.artifacts is None:
    raise RuntimeError("Artifact service is unavailable")

async with context.artifacts.create(
    mime_type="text/plain",
    logical_role="transcript",
) as writer:
    await writer.write(b"hello")
    reference = await writer.finalize()

python

logger = context.logger.bind(component="speech")
logger.info("request accepted", execution_id=context.execution_id)

headers = context.trace.as_headers()