Provider API — Vega SDK Reference

Import

python

from vega_capability_sdk import (
    BaseCapabilityProvider,
    CancellableProvider,
    CapabilityProvider,
    PausableProvider,
    ProviderLifecycleError,
    ProviderLifecycleState,
    ReloadableProvider,
    ResumableProvider,
)

The same symbols are exported from vega_capability_sdk.provider.

BaseCapabilityProvider

python

class BaseCapabilityProvider:
    async def initialize(self, context: ProviderInitializationContext) -> None: ...
    async def execute(
        self,
        request: CapabilityExecutionRequest,
        context: CapabilityExecutionContext,
    ) -> CapabilityExecutionResult: ...
    async def health_check(self) -> CapabilityHealthResult: ...
    async def shutdown(self) -> None: ...

Convenience base class for a Capability Provider. Override its on_* hooks, not its public lifecycle methods, unless an integration has a specific reason to replace the standard lifecycle behavior. The constructor takes no arguments; a subclass that defines__init__ should call super().__init__().

Properties

Term	Meaning
lifecycle_state	Current base-provider ProviderLifecycleState.
init_context	ProviderInitializationContext from the most recent successful initialization. Raises RuntimeError before initialization or after shutdown.

Override hooks

Term	Meaning
on_initialize(context) -> None	Initialize Provider-owned dependencies.
on_execute(request, context) -> CapabilityExecutionResult	Implement an Action execution. Default raises NotImplementedError.
on_health_check() -> CapabilityHealthResult	Return a side-effect-free health result. Default is CapabilityHealthResult.healthy().
on_shutdown() -> None	Release Provider-owned resources. Default does nothing.

Lifecycle method behavior

Term	Meaning
initialize(context)	Allowed from new, stopped, or failed. On success stores context and moves to ready; an exception moves to failed and is re-raised.
execute(request, context)	Requires ready. Validates matching execution IDs. Re-raises CapabilityError; wraps other Exception values in ProviderInternalError.
health_check()	Returns unknown when not ready. Exceptions from the health hook are contained and reported as unknown.
shutdown()	Idempotent after stopped. Cancels tasks created by create_background_task, calls the shutdown hook, clears initialization context, and ends in stopped.

create_background_task

python

def create_background_task(
    self,
    coroutine: Coroutine[Any, Any, T],
    *,
    name: str | None = None,
) -> asyncio.Task[T]: ...

Creates a Provider-managed task while the Provider is ready. The task is cancelled and awaited during shutdown. Calling it outside ready closes the provided coroutine and raises ProviderLifecycleError. Completed tasks are removed from the managed set.

Provider protocols

Protocols are runtime-checkable structural interfaces. A class need not inherit from them; it must provide the required members. CapabilityProvider is the minimum contract (initialize, execute, health_check, shutdown).

Term	Meaning
CancellableProvider	async cancel(execution_id: str) -> None
PausableProvider	async pause(execution_id: str) -> None
ResumableProvider	async resume(execution_id: str) -> None
ReloadableProvider	async reload_configuration(context: ProviderInitializationContext) -> None

These protocols declare Python capabilities only. They do not add transport, scheduling, or Runtime behavior.

Lifecycle types

ProviderLifecycleState is a StrEnum with new, initializing,ready, stopping, stopped, and failed.ProviderLifecycleError is a RuntimeError subclass raised when an operation is invoked in an invalid Provider lifecycle state.

Example

python

from vega_capability_sdk import (
    BaseCapabilityProvider,
    CapabilityExecutionResult,
    CapabilityHealthResult,
)


class EchoProvider(BaseCapabilityProvider):
    async def on_initialize(self, context) -> None:
        self._prefix = context.configuration.get("prefix", "")

    async def on_execute(self, request, context):
        context.cancellation.raise_if_cancelled()
        return CapabilityExecutionResult(
            output={"text": f"{self._prefix}{request.inputs['text']}"}
        )

    async def on_health_check(self):
        return CapabilityHealthResult.healthy()

    async def on_shutdown(self) -> None:
        self._prefix = ""