Vega Python SDK Core APIs — Shyda Platform SDKs

Concepts

Term	Meaning
Provider	Lifecycle-aware implementation of a Capability's Actions.
Initialization context	Read-only configuration and services supplied once after a Provider is loaded.
Execution request	Immutable identity, inputs, optional deadline, and correlation data for one Action invocation.
Execution context	Restricted services for one invocation.
Outcome	A successful result or a structured Capability error.

Guide versus API Reference

Term	Meaning
When to use a Provider hook	Every Provider method and protocol signature
How request, context, result, and error work together	Every field, type, default, and validation rule
How to choose progress, Artifacts, logging, or errors	Complete members of each supporting class
Common implementation patterns and boundaries	Exhaustive enum values and serialization details

1. Start with a Provider, not a Runtime client

BaseCapabilityProvider is the usual entry point. It represents a loaded Capability Provider; it is not a client for connecting to Vega Runtime. The Runtime calls its public lifecycle methods. Your code normally overrides four hooks:

python

class Provider(BaseCapabilityProvider):
    async def on_initialize(self, context):
        ...

    async def on_execute(self, request, context):
        ...

    async def on_health_check(self):
        ...

    async def on_shutdown(self):
        ...

Use the hooks because the base class enforces lifecycle state, retains a successful initialization context, converts unexpected execution exceptions to ProviderInternalError, returns unknown health on health-check exceptions, and cancels base-managed background tasks during shutdown. Do not connect to hardware at module import time.

2. Treat requests as immutable facts

CapabilityExecutionRequest identifies one Action execution. Its required identity fields, inputs, and metadata are deeply frozen. Use context.validate_request(request)first, then read inputs without modifying them.

python

context.validate_request(request)
text = request.inputs.get("text")
if not isinstance(text, str) or not text.strip():
    raise InvalidInputError("text must be a non-empty string")

Mission, task-graph, and node fields are Runtime correlation hints — not public SDK Task or Session APIs. The SDK has no Task, Session, Event Bus, or remote streaming abstraction.

3. Use the execution context as the capability boundary

Term	Meaning
Stop safely	context.cancellation — conveys a cooperative cancellation request and reason.
Show meaningful work state	context.progress — associates structured progress with the execution.
Record diagnostic facts	context.logger — binds structured fields and redacts secret wrappers.
Measure or compare time	context.clock — makes time-dependent code testable.
Correlate downstream calls	context.trace — carries portable trace headers and baggage.
Produce bytes or large output	context.artifacts — creates portable Artifact references.
Inspect assigned resources	context.resource_leases — exposes read-only Runtime lease facts.

Do not treat a lease as device-level safety confirmation, and do not assume an Artifact service is present without checking for None.

4. Report progress for observable long-running work

python

await context.progress.report({"phase": "starting"})
await context.progress.report({"phase": "working", "percentComplete": 50})

The SDK freezes the payload, enforces a default 64 KiB size limit, and adds execution ID, an increasing sequence, timestamp, and trace ID. Progress is a callback into the current integration — not a WebSocket, gRPC, or general Event API.

5. Return results; raise expected failures

python

return CapabilityExecutionResult(
    output={"spoken": True},
    diagnostics={"backend": "mock"},
    physical_termination=PhysicalTerminationStatus.NOT_APPLICABLE,
)

Term	Meaning
Input shape is valid but a semantic value is unusable	InvalidInputError
Device or request precondition is absent	PreconditionFailedError
Required service is unavailable	DependencyUnavailableError
Runtime-assigned resource is unavailable	ResourceUnavailableError
Device operation fails	HardwareError
External API communication fails	CommunicationError
Operation exceeds a Provider-observed deadline	TimeoutCapabilityError
Policy must reject work	SafetyRejectionError
Provider cannot implement requested Action	UnsupportedOperationError

Do not catch these errors only to turn them into a generic failure. Let them propagate. The base Provider converts unexpected ordinary exceptions to ProviderInternalError and logs the exception type.

An error's retryable value is a hint, not an instruction to retry inside the Provider. Runtime owns retry and continuation decisions.

6. Use initialization context for stable dependencies

python

async def on_initialize(self, context):
    endpoint = context.configuration.require("endpoint", str)
    token = context.secrets.get("service_token")
    self._client = await Client.connect(endpoint, token.get_secret_value())

The secret wrapper redacts when logged or formatted. Its raw value is still sensitive after get_secret_value; never put it in errors, diagnostics, or logs. Configuration is deeply immutable.

7. Use health as a safe operational fact

python

async def on_health_check(self):
    if self._client is None:
        return CapabilityHealthResult.unhealthy("Client is not initialized")
    if await self._client.ping():
        return CapabilityHealthResult.healthy(message="Service ready")
    return CapabilityHealthResult.degraded("Service did not respond")

Complete Provider pattern

python

from vega_capability_sdk import (
    BaseCapabilityProvider,
    CapabilityExecutionResult,
    CapabilityHealthResult,
    DependencyUnavailableError,
    InvalidInputError,
    PhysicalTerminationStatus,
)

class EchoProvider(BaseCapabilityProvider):
    def __init__(self, adapter) -> None:
        super().__init__()
        self._adapter = adapter

    async def on_initialize(self, context) -> None:
        await self._adapter.connect(context.configuration)

    async def on_execute(self, request, context):
        context.validate_request(request)
        text = request.inputs.get("text")
        if not isinstance(text, str) or not text.strip():
            raise InvalidInputError("text must be a non-empty string")
        context.cancellation.raise_if_cancelled()
        context.logger.info("echo started", action_id=request.action_id)
        await context.progress.report({"phase": "working"})
        try:
            value = await self._adapter.echo(text)
        except OSError as exc:
            raise DependencyUnavailableError("Echo backend unavailable", cause=exc) from exc
        context.cancellation.raise_if_cancelled()
        return CapabilityExecutionResult(output={"echo": value})

    async def on_health_check(self):
        return (
            CapabilityHealthResult.healthy()
            if await self._adapter.health()
            else CapabilityHealthResult.degraded("Echo backend unavailable")
        )

    async def on_shutdown(self) -> None:
        await self._adapter.close()

The example intentionally keeps the adapter outside SDK types. Vendor clients, ROS nodes, device protocols, and AI services belong behind Capability-owned interfaces.

Test the composition

Test Provider hooks through public lifecycle methods so the test includes base state enforcement and error conversion:

python

from vega_capability_sdk import CapabilityExecutionRequest
from vega_capability_sdk.testing import ProviderContractHarness

request = CapabilityExecutionRequest(
    execution_id="test-execution",
    capability_id="example.echo",
    provider_id="example.echo.python",
    action_id="example.echo.run",
    action_version="1.0.0",
    inputs={"text": "hello"},
)

report = await ProviderContractHarness().run(provider, request)
report.raise_for_failures()

Pass the manifest ActionDefinition to the harness when output-schema validation should be part of the test.

Best practices

Override on_* hooks and preserve the base public lifecycle.
Establish stable dependencies during initialization, not module import.
Keep one execution's mutable state isolated from other executions.
Observe cancellation before and after external I/O.
Use context services instead of hidden Runtime imports or global state.
Return only facts the backend can support.
Keep errors, Progress, diagnostics, and logs bounded and free of secrets.

Common pitfalls

Returning a dictionary instead of CapabilityExecutionResult.
Mutating request inputs, configuration, metadata, or other frozen mappings.
Assuming deadline automatically interrupts Provider code.
Treating optional correlation IDs as SDK Task or Session objects.
Catching CapabilityError and replacing it with a generic exception.
Reporting confirmed physical termination without backend acknowledgement.
Accessing context.artifacts without checking for None.
Starting unmanaged background tasks that survive shutdown.