Service Calls

Understanding Home Assistant’s service registry and service call execution system

Services are the primary way to trigger actions in Home Assistant. The Service Registry manages registration, validation, and execution of callable services across all domains.

ServiceRegistry Overview

The ServiceRegistry provides a centralized system for managing services:

homeassistant/core.py

class ServiceRegistry:
    """Offer the services over the eventbus."""
    
    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize a service registry."""
        self._services: dict[str, dict[str, Service]] = {}
        self._hass = hass

Services are organized in a two-level dictionary structure: domain -> service_name -> Service. This enables O(1) lookups for service execution.

Service Object Structure

Each registered service is represented by a Service object:

homeassistant/core.py

class Service:
    """Representation of a callable service."""
    
    def __init__(
        self,
        func: Callable[[ServiceCall], Coroutine[Any, Any, ServiceResponse] | ServiceResponse | None],
        schema: VolSchemaType | None,
        domain: str,
        service: str,
        context: Context | None = None,
        supports_response: SupportsResponse = SupportsResponse.NONE,
        job_type: HassJobType | None = None,
        description_placeholders: Mapping[str, str] | None = None,
    ) -> None:
        self.job = HassJob(func, f"service {domain}.{service}", job_type=job_type)
        self.schema = schema
        self.supports_response = supports_response
        self.description_placeholders = description_placeholders

Service Components

job

HassJob wrapping the service function for optimized execution

schema

Voluptuous schema for validating service data

supports_response

Whether the service can return data to the caller

description_placeholders

Translation placeholders for service descriptions

ServiceCall Object

When a service is called, it receives a ServiceCall object:

homeassistant/core.py

class ServiceCall:
    """Representation of a call to a service."""
    
    def __init__(
        self,
        hass: HomeAssistant,
        domain: str,
        service: str,
        data: dict[str, Any] | None = None,
        context: Context | None = None,
        return_response: bool = False,
    ) -> None:
        self.hass = hass
        self.domain = domain
        self.service = service
        self.data = ReadOnlyDict(data or {})
        self.context = context or Context()
        self.return_response = return_response

ServiceCall Attributes

• domain: The domain of the service (e.g., “light”)
• service: The service name (e.g., “turn_on”)
• data: Read-only dictionary of service parameters
• context: Context tracking who/what triggered the call
• return_response: Whether the caller expects return data

Registering Services

Basic Registration

import voluptuous as vol
from homeassistant.core import HomeAssistant, ServiceCall

async def async_handle_my_service(call: ServiceCall) -> None:
    """Handle the my_service call."""
    entity_id = call.data.get("entity_id")
    value = call.data.get("value")
    
    _LOGGER.info("Service called with entity_id=%s, value=%s", entity_id, value)
    # Perform the service action
    await do_something(entity_id, value)

# Define validation schema
SERVICE_SCHEMA = vol.Schema({
    vol.Required("entity_id"): cv.entity_id,
    vol.Required("value"): vol.Coerce(int),
})

# Register the service
hass.services.async_register(
    "my_domain",
    "my_service",
    async_handle_my_service,
    schema=SERVICE_SCHEMA
)

1. Service Function Defined

   Create an async function that accepts a ServiceCall parameter.

2. Schema Created

   Define a voluptuous schema to validate service data.

3. Service Registered

   Call async_register to register the service with the registry.

4. Event Fired

   The registry fires EVENT_SERVICE_REGISTERED to notify listeners.

Registration with Response Support

from homeassistant.core import SupportsResponse, ServiceResponse

async def async_get_data(call: ServiceCall) -> ServiceResponse:
    """Service that returns data to the caller."""
    entity_id = call.data["entity_id"]
    
    # Fetch data
    data = await fetch_entity_data(entity_id)
    
    # Return dict (must be JSON-serializable)
    return {
        "entity_id": entity_id,
        "value": data.value,
        "timestamp": data.timestamp.isoformat(),
    }

hass.services.async_register(
    "my_domain",
    "get_data",
    async_get_data,
    schema=GET_DATA_SCHEMA,
    supports_response=SupportsResponse.ONLY
)

Response Support Modes

homeassistant/core.py

class SupportsResponse(enum.StrEnum):
    """Service call response configuration."""
    
    NONE = "none"        # No response data (default)
    OPTIONAL = "optional" # Response data when requested
    ONLY = "only"        # Must always request response

• NONE: Traditional services that perform actions without returning data
• OPTIONAL: Services that can optionally return data when return_response=True
• ONLY: Read-only services that must be called with return_response=True

Calling Services

Basic Service Call

# Async context (non-blocking)
await hass.services.async_call(
    "light",
    "turn_on",
    {"entity_id": "light.living_room", "brightness": 255},
    blocking=False
)

# Blocking call (wait for completion)
await hass.services.async_call(
    "light",
    "turn_on",
    {"entity_id": "light.living_room", "brightness": 255},
    blocking=True
)

Service Call with Response

# Request response data
response = await hass.services.async_call(
    "my_domain",
    "get_data",
    {"entity_id": "sensor.temperature"},
    blocking=True,
    return_response=True
)

print(f"Temperature: {response['value']}")

Services can only return response data when called with blocking=True and return_response=True. Attempting to use return_response=True with blocking=False will raise a ServiceValidationError.

Service Call Flow

homeassistant/core.py

async def async_call(
    self,
    domain: str,
    service: str,
    service_data: dict[str, Any] | None = None,
    blocking: bool = False,
    context: Context | None = None,
    target: dict[str, Any] | None = None,
    return_response: bool = False,
) -> ServiceResponse:
    """Call a service."""
    context = context or Context()
    service_data = service_data or {}
    
    # Lookup service handler
    try:
        handler = self._services[domain][service]
    except KeyError:
        domain = domain.lower()
        service = service.lower()
        try:
            handler = self._services[domain][service]
        except KeyError:
            raise ServiceNotFound(domain, service) from None
    
    # Validate response support
    if return_response and handler.supports_response is SupportsResponse.NONE:
        raise ServiceValidationError("Service does not support response")
    
    # Merge target into service_data
    if target:
        service_data.update(target)
    
    # Validate service data
    if handler.schema:
        processed_data = handler.schema(service_data)
    else:
        processed_data = service_data
    
    # Create ServiceCall object
    service_call = ServiceCall(
        self._hass, domain, service, processed_data, context, return_response
    )
    
    # Fire call_service event
    self._hass.bus.async_fire_internal(
        EVENT_CALL_SERVICE,
        {
            ATTR_DOMAIN: domain,
            ATTR_SERVICE: service,
            ATTR_SERVICE_DATA: service_data,
        },
        context=context,
    )
    
    # Execute service
    coro = self._execute_service(handler, service_call)
    if not blocking:
        # Run in background
        self._hass.async_create_task_internal(
            self._run_service_call_catch_exceptions(coro, service_call),
            f"service call background {service_call.domain}.{service_call.service}",
            eager_start=True,
        )
        return None
    
    # Wait for completion
    response_data = await coro
    if not return_response:
        return None
    return response_data

1. Handler Lookup

   The service handler is retrieved from the registry.

2. Response Validation

   Response support flags are validated against the request.

3. Data Validation

   Service data is validated against the service schema.

4. Event Fired

   EVENT_CALL_SERVICE is fired to notify listeners.

5. Service Executed

   The service handler is executed (blocking or background).

Service Execution

Services are executed based on their job type:

homeassistant/core.py

async def _execute_service(
    self, handler: Service, service_call: ServiceCall
) -> ServiceResponse:
    """Execute a service."""
    job = handler.job
    target = job.target
    
    if job.job_type is HassJobType.Coroutinefunction:
        # Async function - await directly
        return await target(service_call)
    
    if job.job_type is HassJobType.Callback:
        # Callback - call immediately
        return target(service_call)
    
    # Executor job - run in thread pool
    return await self._hass.async_add_executor_job(target, service_call)

The job type is determined automatically when the service is registered, enabling optimal execution without runtime checks.

Service Schema Validation

Schemas use the Voluptuous library for validation:

import voluptuous as vol
from homeassistant.helpers import config_validation as cv

SERVICE_SCHEMA = vol.Schema({
    # Required fields
    vol.Required("entity_id"): cv.entity_ids,
    
    # Optional fields with defaults
    vol.Optional("brightness", default=255): vol.All(
        vol.Coerce(int),
        vol.Range(min=0, max=255)
    ),
    
    # Conditional fields
    vol.Optional("color_temp"): vol.All(
        vol.Coerce(int),
        vol.Range(min=153, max=500)
    ),
    
    # Custom validation
    vol.Optional("transition"): vol.All(
        vol.Coerce(float),
        vol.Range(min=0)
    ),
})

Common Validators

Entity Validators

import homeassistant.helpers.config_validation as cv

# Single entity ID
schema = vol.Schema({"entity_id": cv.entity_id})

# List of entity IDs
schema = vol.Schema({"entity_id": cv.entity_ids})

# Entity ID from specific domain
schema = vol.Schema({"entity_id": cv.entity_domain("light")})

Checking Service Availability

# Check if a service exists
if hass.services.has_service("light", "turn_on"):
    await hass.services.async_call("light", "turn_on", ...)

# Check response support
response_support = hass.services.supports_response("my_domain", "get_data")
if response_support == SupportsResponse.ONLY:
    # Must use return_response=True
    response = await hass.services.async_call(
        "my_domain",
        "get_data",
        ...,
        blocking=True,
        return_response=True
    )

Removing Services

Unregister a service when your component unloads:

# Async context
hass.services.async_remove("my_domain", "my_service")

# Fires EVENT_SERVICE_REMOVED

Always remove services during component cleanup to prevent orphaned service registrations that reference unloaded code.

Service Events

Service operations trigger events:

EVENT_SERVICE_REGISTERED

Fired when a service is registered:

{
    "domain": "my_domain",
    "service": "my_service"
}

EVENT_CALL_SERVICE

Fired when a service is called:

{
    "domain": "light",
    "service": "turn_on",
    "service_data": {
        "entity_id": "light.living_room",
        "brightness": 255
    }
}

EVENT_SERVICE_REMOVED

Fired when a service is unregistered:

{
    "domain": "my_domain",
    "service": "my_service"
}

Error Handling

Service calls can raise several exceptions:

from homeassistant.exceptions import (
    ServiceNotFound,
    ServiceValidationError,
    Unauthorized,
)

try:
    await hass.services.async_call(
        "my_domain",
        "my_service",
        service_data,
        blocking=True,
    )
except ServiceNotFound:
    _LOGGER.error("Service not found")
except ServiceValidationError as err:
    _LOGGER.error("Invalid service data: %s", err)
except Unauthorized:
    _LOGGER.error("Unauthorized to call service")

When blocking=False, exceptions are caught and logged automatically by _run_service_call_catch_exceptions. Only blocking calls propagate exceptions to the caller.

Best Practices

Use async handlers: Always prefer async service handlers for better performance
Define schemas: Always define validation schemas to catch errors early
Return JSON-serializable data: Response data must be JSON-serializable
Handle contexts: Propagate contexts to maintain audit trails
Document services: Use service descriptions and description_placeholders

Service Description Files

Services are documented in services.yaml:

my_service:
  name: My Service
  description: Performs a custom action
  fields:
    entity_id:
      description: Entity to control
      example: "light.living_room"
      required: true
      selector:
        entity:
          domain: light
    value:
      description: Value to set
      example: 50
      required: true
      selector:
        number:
          min: 0
          max: 100
          mode: slider

This enables automatic UI generation and API documentation.

Performance Considerations

1. Use callbacks: Decorate simple services with @callback when possible
2. Avoid blocking I/O: Use async operations or executor jobs for I/O
3. Validate early: Schema validation prevents invalid data from reaching handlers
4. Background execution: Use blocking=False for fire-and-forget operations
5. Batch operations: Consider batching multiple entity operations in a single service