ShydaSHYDAPLATFORM
Vega SDK — Execution API

Execution API

Types and services used during a single Action invocation: request, execution context, result, cancellation token, progress reporter, and physical termination status.

Import

python
from vega_capability_sdk import (
    CancellationReason,
    CancellationToken,
    CapabilityExecutionContext,
    CapabilityExecutionRequest,
    CapabilityExecutionResult,
    PhysicalTerminationStatus,
    ProgressReporter,
)

Also exported from vega_capability_sdk.execution and vega_capability_sdk.contracts.

CapabilityExecutionRequest

python
@dataclass(frozen=True, slots=True)
class CapabilityExecutionRequest:
    execution_id: str
    capability_id: str
    provider_id: str
    action_id: str
    action_version: str
    inputs: Mapping[str, Any]
    deadline: datetime | None = None
    idempotency_key: str | None = None
    mission_id: str | None = None
    task_graph_id: str | None = None
    node_id: str | None = None
    trace_id: str | None = None
    metadata: Mapping[str, Any] = {}

Immutable request for one Action invocation. The five identity fields must be non-blank.inputs and metadata are deeply frozen JSON-compatible mappings. If supplied, deadline must be timezone-aware and is normalized to UTC. Construction raises ValueError for blank identities or a naive deadline, and TypeErrorfor non-JSON values or non-string mapping keys.

TermMeaning
execution_id / capability_id / provider_id / action_id / action_versionRequired identity fields; must not be blank.
inputsRequired JSON-compatible input mapping; recursively frozen.
deadlineOptional UTC-aware absolute deadline; not enforced by the request object.
idempotency_keyCorrelation value. The SDK does not deduplicate requests.
mission_id / task_graph_id / node_id / trace_idCorrelation-only fields; no Task or Session objects are exposed.
metadataOptional JSON-compatible correlation mapping; recursively frozen.

CapabilityExecutionContext

python
@dataclass(frozen=True, slots=True)
class CapabilityExecutionContext:
    execution_id: str
    cancellation: CancellationToken
    progress: ProgressReporter
    deadline: datetime | None = None
    logger_name: str = "vega.capability"
    artifacts: ArtifactService | None = None
    logger: StructuredLogger = StructuredLogger()
    clock: Clock = SystemClock()
    trace: TraceContext = TraceContext()
    resource_leases: tuple[ResourceLeaseView, ...] = ()

Restricted services and facts for a single execution. Any deadline must be UTC-aware.validate_request(request) raises ValueError unlessrequest.execution_id equals context.execution_id;BaseCapabilityProvider.execute performs this check before callingon_execute.

CapabilityExecutionResult

python
@dataclass(frozen=True, slots=True)
class CapabilityExecutionResult:
    output: Mapping[str, Any]
    artifact_references: tuple[str | ArtifactReference, ...] = ()
    diagnostics: Mapping[str, Any] = {}
    physical_termination: PhysicalTerminationStatus | None = None


    def to_dict(self) -> dict[str, Any]: ...

Successful Action result. output and diagnostics are deeply frozen JSON-compatible mappings. to_dict() thaws mappings and serializes eachArtifactReference; string artifact references are retained.

Cancellation

CancellationReason is a StrEnum: user_request, mission_cancelled,timeout, task_timeout, runtime_shutdown, safety_stop,operator_request, parent_cancelled.

python
class CancellationToken:
    @property
    def is_cancel_requested(self) -> bool: ...
    @property
    def cancellation_reason(self) -> CancellationReason | None: ...
    @property
    def secondary_reasons(self) -> tuple[CancellationReason, ...]: ...
    def request_cancel(self, reason: CancellationReason) -> bool: ...
    async def wait_for_cancel(self) -> CancellationReason: ...
    def raise_if_cancelled(self) -> None: ...

The first cancellation reason becomes the primary reason; later distinct reasons are retained as secondary. request_cancel returns True only when it establishes the primary reason. raise_if_cancelled() raises CancellationCapabilityError. The token does not cancel an asyncio task, stop hardware, or invokeCancellableProvider.cancel.

ProgressReporter

python
class ProgressReporter:
    def __init__(
        self,
        *,
        execution_id: str,
        on_progress: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
        clock: Clock | None = None,
        trace: TraceContext | None = None,
        validate_payload: Callable[[dict[str, Any]], None] | None = None,
        max_payload_bytes: int = 64 * 1024,
    ) -> None: ...


    async def report(self, payload: Mapping[str, Any]) -> None: ...

report freezes and validates the payload, applies the byte limit to its JSON representation, and creates an event with execution_id, monotonically increasingsequence, ISO timestamp, trace_id, and payload. RaisesTypeError for a non-JSON payload and ValueError when the encoded payload exceeds max_payload_bytes.

PhysicalTerminationStatus

StrEnum with not_applicable, unknown, confirmed, andfailed_to_confirm. Use it only for factual knowledge about the underlying physical action; Python coroutine completion is not proof of a physical stop.

Example

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


# Inside on_execute:
context.validate_request(request)
context.cancellation.raise_if_cancelled()
await context.progress.report({"phase": "complete"})
return CapabilityExecutionResult(
    output={"text": request.inputs["text"]},
    physical_termination=PhysicalTerminationStatus.NOT_APPLICABLE,
)