Implementing SiLA features

For each feature that is added to a SiLA Server/Client package, a file [featureidentifier] is generated in the feature_implementations directory. It contains a class [FeatureIdentifier]Impl, which extends the generated base class [FeatureIdentifier]Base. This base class extends the class FeatureImplementationBase.

All code examples follow the features implemented in the example server.

Unobservable property

For each unobservable property, a method get_[PropertyIdentifier](self, *, metadata) is generated. It is called when a SiLA Client requests the current property value.

# GreetingProvider.StartYear
def get_StartYear(self, *, metadata) -> int:
    return 2022

Unobservable command

For each unobservable command, a method [CommandIdentifier](self, Param1, Param2, ..., *, metadata) is generated. It is called when a SiLA Client requests the execution of this command. Each parameter from the feature definition is translated into one method argument (e.g. Param1). You have to return a [CommandIdentifier]_Responses object, which is a NamedTuple instance with one field per command response specified in the feature definition. Example: The command SayHello has one response “Greeting”. To implement this method, use return SayHello_Responses(Greeting=...).

# GreetingProvider.SayHello
def SayHello(self, Name: str, *, metadata) -> SayHello_Responses:
    return SayHello_Responses(Greeting="Hello " + Name)

Observable property

When a SiLA Client requests to subscribe to an observable property, the method [PropertyIdentifier]_on_subscription(self, *, metadata) is called. It is defined in the Base class. If you want to specify what happens on subscriptions, override this method, and don’t forget to call the super method:

# TimerProvider.CurrentTime: print a message when a client subscribes
def CurrentTime_on_subscription(self, *, metadata) -> None:
    print("A client subscribed to MyProperty!")

To update the property value, use the method update_MyProperty(new_value), which is also defined in the Base class. When this method is called, the new value is sent to all currently subscribed clients, and it is stored and sent immediately when a clients subscribes to the property.

You can call this method from anywhere, e.g. in the __init__ method of the class to set an initial value, and when other methods are called that change the value of the property.

If you regularly want to update the value, you can use the method FeatureImplementationBase.run_periocially().

from datetime import datetime, time, timezone

# TimerProvider.CurrentTime: update observable property every second
class TimerProviderImpl(TimerProviderBase):
    def __init__(self, parent_server: Server) -> None:

        def update_current_time() -> None:
            # get current time
            current_time: time =

            # update observable property

        self.run_periodically(update_current_time, delay_seconds=1)

Observable command

Observable commands are implemented similarly to unobservable commands. A method [CommandIdentifier](self, Param1, Param2, ..., *, metadata, instance) is generated, which is called when a SiLA Client requests to execute the command. Like command responses, intermediate responses are also represented by :py:class:~typing.NamedTuple` classes.

There are two additional things to keep in mind:

Command instance

The implementation method receives the additional parameter instance, which is an ObservableCommandInstance object. It can be used to send updates to the client while the command is running (progress, estimated_remaining_time, lifetime_of_execution, and send_intermediate_response()).

Lifetime of execution

Instances of observable commands have a “lifetime of execution”. After its lifetime, client cannot subscribe to the status or intermediate responses of the instance. The lifetime can be extended, but not reduced.


When the lifetime is None, it is infinite. This means that the command instance is stored in the server memory until the server is shut down.

class TimerProviderImpl(TimerProviderBase):
    def __init__(self, parent_server: Server) -> None:
        self.Countdown_default_lifetime_of_execution = datetime.timedelta(seconds=2)

    def Countdown(
        self, N: int, Message: str, *, metadata, instance: ObservableCommandInstanceWithIntermediateResponses[Countdown_IntermediateResponses]
    ) -> Countdown_Responses:
        # change status from `waiting` to `running`

        instance.progress = 0
        instance.estimated_remaining_time = datetime.timedelta(seconds=N)
        instance.lifetime_of_exection = datetime.timedelta(2 + N)

        for i in range(N, 0, -1):
            instance.send_intermediate_response(Countdown_IntermediateResponses(i))  # send intermediate responses
            instance.progress = (N - i) / N * 100
            instance.estimated_remaining_time = datetime.timedelta(seconds=i)


        return Countdown_Responses(Message,

Error handling

Defined Execution Errors

If the feature definition contains defined execution errors, DefinedExecutionError subclasses are generated. You can import them from the generated feature submodule and raise them like normal Python exceptions.

from ..generated.timerprovider import CountdownTooLong

class TimerProviderImpl(TimerProviderBase):
    def Countdown(self, N: int, ...):
        if N > 9000:
            raise CountdownTooLong

Undefined Execution Errors

If your code raises an uncaught exception, the SDK will wrap it in an UndefinedExecutionError and sent that to the client.

Validation Errors

These are raised by the SDK automatically if a client sends parameters that violate constraints specified in the feature definition. In such cases, your feature implementation code will not be called.


Receiving metadata in command/property implementations

The property access/subscription and also the command methods have a parameter metadata, which is a MetadataDict. When SiLA clients send metadata along with a request for property access or command execution, this metadata will be included in that dict-like object. The keys are sila2.framework.Metadata objects and the values are the metadata values sent by the client.

Metadata Interceptors

Usually, metadata affects multiple commands and properties and should be handled in the same way. This can be achieved by implementing a MetadataInterceptor and adding it to the Server class.

# new file feature_implementations/
from typing import Any, Dict

from sila2.framework import FullyQualifiedIdentifier
from sila2.server import MetadataDict, MetadataInterceptor

from ..generated.delayprovider import DelayProviderFeature, DelayTooLong

class DelayInterceptor(MetadataInterceptor):
    def __init__(self):
        # this interceptor affects the Delay metadata in the DelayProvider feature

    def intercept(self, parameters, metadata: MetadataDict, target_call: FullyQualifiedIdentifier) -> None:
        delay: int = metadata[DelayProviderFeature["Delay"]]

        if delay > 10_000:
            raise DelayTooLong("Maximum delay is 10 seconds")  # defined execution error

        time.sleep(delay/1000)  # delay is given in milliseconds
# in
from .feature_implementations.delay_importceptor import DelayInterceptor

class Server(SilaServer):
    def __init__(...):

        # add interceptor to server

start() and stop(): Setup and Shutdown

If you want to execute code on server startup or shutdown, you can override the start() and stop() methods inherited from FeatureImplementationBase. These methods are called by the server, do not call them yourself.

Make sure to call the super methods:

def start(self) -> None:
    print("Feature implementation started")

def stop(self) -> None:
    print("Feature implementation stopped")