Utilities

Validation

Use vega-capability validate or validate_capability_package to validate a package without importing the Provider. Use CapabilityPackageValidator.validate_payload when a test or integration boundary needs JSON Schema payload validation.

from vega_capability_sdk.manifests import validate_capability_package

definition = validate_capability_package(".")
action = definition.action_by_id("example.echo.run")

Logging, configuration, and secrets

StructuredLogger accepts structured fields and recursively redacts values marked as secrets. ProviderConfigurationView.require retrieves a typed required value. SecretValue redacts on string conversion but its raw value is still sensitive after get_secret_value.

token = self.init_context.secrets.get("api_token")
context.logger.info("service configured", token=token)

Use configuration.get() for an optional value and require() for a required, typed value:

endpoint = self.init_context.configuration.require("endpoint", str)
timeout = self.init_context.configuration.get("timeout_seconds", 10)

SecretAccessor.get() raises KeyError when a secret is unavailable or not authorized. Never include the raw value in logs, errors, Progress, results, diagnostics, or Artifact metadata. Bind stable fields once instead of repeating them:

logger = context.logger.bind(component="speech_backend")
logger.info("request accepted", action_id=request.action_id)

Clock, trace, and serialization

Use context.clock.now() rather than direct wall-clock access when time affects behavior or tests. The SDK requires timezone-aware UTC datetimes for deadlines and health timestamps. TraceContext.as_headers() can provide traceparent and baggage for trusted downstream clients. JSON-facing values accept only null, booleans, numbers, strings, mappings with string keys, and lists/tuples.

An execution deadline is a fact, not an automatic timer. Compare it using the injected clock or rely on an explicit cancellation signal according to your Provider contract:

if request.deadline is not None and context.clock.now() >= request.deadline:
    raise TimeoutCapabilityError("Execution deadline has passed")

Only forward trace headers to a trusted downstream service. The SDK formats provided values; it does not validate W3C identifier syntax or escape baggage. SDK-facing JSON values are deeply frozen — build a new result rather than attempting to mutate request input.

Artifacts and resources

Use an Artifact service for bytes or large data, then return the reference:

from vega_capability_sdk import DependencyUnavailableError

if context.artifacts is None:
    raise DependencyUnavailableError("Artifact storage is unavailable")

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

return CapabilityExecutionResult(output={"ok": True}, artifact_references=(ref,))

resource_leases are read-only Runtime-provided views. They do not replace device-level validation or safety checks. Do not return local absolute file paths as portable Action output — an Artifact reference is the public handoff for stored bytes.

Testing helpers

• TestClock for deterministic time control.
• ProgressCollector to capture emitted progress reports.
• InMemoryArtifactStore for artifact assertions.
• fake_initialization_context and fake_execution_context for isolated Provider tests.
• ProviderContractHarness to exercise the full Provider lifecycle.

from datetime import timedelta

from vega_capability_sdk.testing import (
    ProgressCollector,
    TestClock,
    fake_execution_context,
)

clock = TestClock()
collector = ProgressCollector()
context = fake_execution_context(clock=clock, collector=collector)

clock.advance(timedelta(seconds=5))
await context.progress.report({"phase": "working"})

assert collector.events[0]["timestamp"] == clock.now().isoformat()

Pass an explicit collector when the test needs its events. If omitted, the factory creates an internal collector that is not returned separately.

Choose the appropriate utility

Common pitfalls

• Expecting the manifest validator to import or run the Provider.
• Treating a redacted wrapper as permission to expose its raw secret.
• Using wall-clock functions when an injected clock is available.
• Assuming a deadline automatically stops execution.
• Returning bytes, vendor objects, datetimes, or local paths in JSON output.
• Assuming Artifact storage is always configured.
• Treating a resource lease as a hardware safety check.
• Creating a test Progress collector without retaining it for assertions.