Service Registration

Registering and handling services in Home Assistant Core

Services are the primary way to trigger actions in Home Assistant. Understanding how to register and handle services properly is essential for creating integrations that users can control through automations, scripts, and the UI.

Service Registry

The ServiceRegistry manages all available services in Home Assistant:

from homeassistant.core import HomeAssistant, ServiceCall

def example(hass: HomeAssistant):
    # Access service registry
    service_registry = hass.services
    
    # Check if service exists
    if service_registry.has_service("light", "turn_on"):
        # Call service
        await service_registry.async_call(
            "light",
            "turn_on",
            {"entity_id": "light.living_room"}
        )

Reference: homeassistant/core.py:2477

Registering Services

Basic Service Registration

from homeassistant.core import HomeAssistant, ServiceCall, callback
import voluptuous as vol
from homeassistant.helpers import config_validation as cv

DOMAIN = "my_integration"

# Define service schema for validation
SERVICE_MY_ACTION_SCHEMA = vol.Schema({
    vol.Required("target"): cv.string,
    vol.Optional("intensity", default=50): vol.All(
        vol.Coerce(int),
        vol.Range(min=0, max=100)
    ),
})

async def async_setup(hass: HomeAssistant, config: dict) -> bool:
    """Set up the integration."""
    
    async def handle_my_action(call: ServiceCall) -> None:
        """Handle the service call."""
        target = call.data["target"]
        intensity = call.data["intensity"]
        
        _LOGGER.info(f"Executing action on {target} with intensity {intensity}")
        # Perform action
        await perform_action(target, intensity)
    
    # Register the service
    hass.services.async_register(
        DOMAIN,
        "my_action",
        handle_my_action,
        schema=SERVICE_MY_ACTION_SCHEMA
    )
    
    return True

Reference: homeassistant/core.py:2570

Service with Response Data

Services can return data to callers:

from homeassistant.core import SupportsResponse, ServiceResponse
import voluptuous as vol

SERVICE_GET_INFO_SCHEMA = vol.Schema({
    vol.Required("device_id"): cv.string,
})

async def handle_get_info(call: ServiceCall) -> ServiceResponse:
    """Handle service that returns data."""
    device_id = call.data["device_id"]
    
    # Fetch device information
    device_info = await fetch_device_info(device_id)
    
    # Return response data
    return {
        "device_id": device_id,
        "name": device_info["name"],
        "status": device_info["status"],
        "temperature": device_info["temp"],
    }

hass.services.async_register(
    DOMAIN,
    "get_info",
    handle_get_info,
    schema=SERVICE_GET_INFO_SCHEMA,
    supports_response=SupportsResponse.ONLY  # Requires return_response=True
)

Reference: homeassistant/core.py:2533

Service Response Types

from homeassistant.core import SupportsResponse

# NONE - Service does not return data (default)
supports_response=SupportsResponse.NONE

# OPTIONAL - Service can return data if requested
supports_response=SupportsResponse.OPTIONAL  

# ONLY - Service must be called with return_response=True
supports_response=SupportsResponse.ONLY

Service Handler Types

Callback Handler

For synchronous operations:

from homeassistant.core import callback, HassJobType

@callback
def sync_service_handler(call: ServiceCall) -> None:
    """Handle service synchronously."""
    # Fast, synchronous work only
    state = hass.states.get(call.data["entity_id"])
    _LOGGER.info(f"Current state: {state.state}")

hass.services.async_register(
    DOMAIN,
    "sync_action",
    sync_service_handler,
    job_type=HassJobType.Callback  # Optional: pre-specify job type
)

Reference: homeassistant/core.py:287

Async Handler

For async operations (most common):

async def async_service_handler(call: ServiceCall) -> None:
    """Handle service asynchronously."""
    # Can perform async operations
    device_id = call.data["device_id"]
    await control_device(device_id)
    await asyncio.sleep(1)
    _LOGGER.info("Action completed")

hass.services.async_register(
    DOMAIN,
    "async_action",
    async_service_handler
)

Executor Handler

For blocking operations:

import time

def blocking_service_handler(call: ServiceCall) -> None:
    """Handle service with blocking code."""
    # This will automatically run in executor
    time.sleep(5)  # Blocking is OK here
    # Perform blocking I/O
    with open("/path/to/file") as f:
        data = f.read()

hass.services.async_register(
    DOMAIN,
    "blocking_action",
    blocking_service_handler
    # HassJobType.Executor detected automatically
)

Reference: homeassistant/core.py:347

Service Call Object

ServiceCall Properties

def handle_service(call: ServiceCall) -> None:
    """Handle service call."""
    # Service identification
    domain: str = call.domain          # "my_integration"
    service: str = call.service        # "my_action"
    
    # Service data (validated by schema)
    data: dict = call.data
    entity_id = call.data.get("entity_id")
    
    # Context (who triggered it)
    context: Context = call.context
    user_id = call.context.user_id
    
    # Return response indicator
    return_response: bool = call.return_response

Calling Services

Basic Service Call

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

Reference: homeassistant/core.py:2712

Fire and Forget

# Call service without waiting
await hass.services.async_call(
    "notify",
    "send_message",
    {"message": "Hello"},
    blocking=False  # Don't wait (default)
)

Service Call with Response

# Call service and get response data
response = await hass.services.async_call(
    "my_integration",
    "get_info",
    {"device_id": "abc123"},
    blocking=True,
    return_response=True
)

if response:
    device_name = response["name"]
    device_status = response["status"]

Reference: homeassistant/core.py:2752

Service Call with Target

Target allows specifying entities, devices, or areas:

# Target specific entities
await hass.services.async_call(
    "light",
    "turn_on",
    service_data={"brightness": 200},
    target={
        "entity_id": ["light.living_room", "light.bedroom"]
    }
)

# Target devices
await hass.services.async_call(
    "homeassistant",
    "reload_config_entry",
    target={
        "device_id": ["device_abc", "device_xyz"]
    }
)

# Target areas
await hass.services.async_call(
    "light",
    "turn_off",
    target={
        "area_id": "living_room"
    }
)

Entity Services

Entity services automatically route calls to entity methods:

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

# Define service schema
SERVICE_SET_SPEED_SCHEMA = vol.Schema({
    vol.Required("speed"): vol.All(
        vol.Coerce(int),
        vol.Range(min=1, max=10)
    )
})

async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
    """Set up platform with entity service."""
    
    # Register service that calls entity method
    await async_register_entity_service(
        hass,
        DOMAIN,
        "set_speed",
        SERVICE_SET_SPEED_SCHEMA,
        "async_set_speed",  # Method name on entity
    )

class MyEntity(Entity):
    """Entity with service handler."""
    
    async def async_set_speed(self, speed: int) -> None:
        """Handle set_speed service call."""
        self._speed = speed
        await self.async_update_ha_state()

Reference: homeassistant/helpers/service.py

Service Validation

Schema Validation

Use voluptuous schemas for robust validation:

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

SERVICE_SCHEMA = vol.Schema({
    vol.Required("entity_id"): cv.entity_id,
    vol.Optional("temperature"): vol.All(
        vol.Coerce(float),
        vol.Range(min=10.0, max=30.0)
    ),
    vol.Optional("mode"): vol.In(["auto", "cool", "heat"]),
    vol.Optional("duration", default=3600): cv.positive_int,
})

Common Validation Helpers

from homeassistant.helpers import config_validation as cv

# Entity validation
entity_id = cv.entity_id            # Validates format
entity_ids = cv.entity_ids          # List of entity IDs
entity_domain = cv.entity_domain("light")  # Must be in domain

# String validation  
string = cv.string                  # Basic string
slug = cv.slug                      # Lowercase alphanumeric + underscore
template = cv.template              # Template string

# Numeric validation
positive_int = cv.positive_int      # > 0
port = cv.port                      # 1-65535
percentage = cv.percentage          # 0-100

# Time validation
time_period = cv.time_period        # timedelta
time = cv.time                      # time object

# Boolean
boolean = cv.boolean                # Flexible bool parsing

Service Events

Listening for Service Calls

from homeassistant.const import EVENT_CALL_SERVICE

@callback
def handle_service_call(event: Event) -> None:
    """Monitor all service calls."""
    domain = event.data["domain"]
    service = event.data["service"]
    service_data = event.data["service_data"]
    
    _LOGGER.debug(f"Service called: {domain}.{service}")

hass.bus.async_listen(EVENT_CALL_SERVICE, handle_service_call)

Service Registration Events

from homeassistant.const import (
    EVENT_SERVICE_REGISTERED,
    EVENT_SERVICE_REMOVED
)

@callback
def handle_service_registered(event: Event) -> None:
    """Handle new service registration."""
    domain = event.data["domain"]
    service = event.data["service"]
    _LOGGER.info(f"Service registered: {domain}.{service}")

hass.bus.async_listen(EVENT_SERVICE_REGISTERED, handle_service_registered)
hass.bus.async_listen(EVENT_SERVICE_REMOVED, handle_service_removed)

Reference: homeassistant/core.py:2644

Removing Services

# Remove service when no longer needed
hass.services.async_remove(DOMAIN, "my_action")

# In cleanup/unload
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """Unload a config entry."""
    # Remove services
    hass.services.async_remove(DOMAIN, "my_action")
    hass.services.async_remove(DOMAIN, "another_action")
    
    return True

Reference: homeassistant/core.py:2655

Advanced Patterns

Service with Context Propagation

async def handle_complex_action(call: ServiceCall) -> None:
    """Handle service with context propagation."""
    # Propagate context to state changes
    hass.states.async_set(
        "sensor.triggered_by_service",
        "active",
        {},
        context=call.context  # Propagate context
    )
    
    # Call another service with same context
    await hass.services.async_call(
        "light",
        "turn_on",
        {"entity_id": "light.related"},
        context=call.context  # Chain context
    )

Dynamic Service Registration

class DynamicServiceManager:
    """Manage services dynamically."""
    
    def __init__(self, hass: HomeAssistant):
        self._hass = hass
        self._registered_services: set[str] = set()
    
    async def register_device_service(self, device_id: str):
        """Register service for specific device."""
        service_name = f"control_{device_id}"
        
        if service_name in self._registered_services:
            return
        
        async def handle_device_service(call: ServiceCall):
            await self.control_device(device_id, call.data)
        
        self._hass.services.async_register(
            DOMAIN,
            service_name,
            handle_device_service
        )
        self._registered_services.add(service_name)
    
    async def cleanup(self):
        """Remove all registered services."""
        for service_name in self._registered_services:
            self._hass.services.async_remove(DOMAIN, service_name)
        self._registered_services.clear()

Service with Multiple Response Types

from homeassistant.core import SupportsResponse

async def flexible_service(call: ServiceCall) -> ServiceResponse:
    """Service that optionally returns data."""
    device_id = call.data["device_id"]
    device_info = await get_device_info(device_id)
    
    # Always perform action
    await control_device(device_id, call.data.get("action"))
    
    # Only return data if requested
    if call.return_response:
        return {
            "device_id": device_id,
            "status": device_info["status"],
            "timestamp": dt_util.utcnow().isoformat()
        }
    
    return None

hass.services.async_register(
    DOMAIN,
    "flexible_action",
    flexible_service,
    supports_response=SupportsResponse.OPTIONAL
)

Best Practices

1. Always Validate Input

# GOOD - Schema validation
SERVICE_SCHEMA = vol.Schema({
    vol.Required("target"): cv.string,
})

hass.services.async_register(
    DOMAIN, "action", handler, schema=SERVICE_SCHEMA
)

# BAD - No validation
hass.services.async_register(
    DOMAIN, "action", handler  # Missing schema
)

2. Use Appropriate Handler Types

# GOOD - Callback for fast sync work
@callback
def quick_handler(call: ServiceCall):
    state = hass.states.get(call.data["entity_id"])

# GOOD - Async for I/O
async def async_handler(call: ServiceCall):
    await external_api_call()

# BAD - Async without awaiting
async def wasteful_handler(call: ServiceCall):
    state = hass.states.get(call.data["entity_id"])  # Wasteful

3. Handle Errors Gracefully

from homeassistant.exceptions import HomeAssistantError, ServiceValidationError

async def safe_handler(call: ServiceCall) -> None:
    """Service handler with error handling."""
    try:
        device_id = call.data["device_id"]
        result = await control_device(device_id)
        
        if not result:
            raise ServiceValidationError("Device control failed")
            
    except KeyError as err:
        raise ServiceValidationError(f"Missing required field: {err}")
    except asyncio.TimeoutError:
        raise HomeAssistantError("Device communication timeout")
    except Exception:
        _LOGGER.exception("Unexpected error in service")
        raise

4. Document Services

Create services.yaml in your integration:

my_action:
  name: My Action
  description: Performs a custom action on the device.
  fields:
    target:
      name: Target
      description: Target device identifier.
      required: true
      example: "device_123"
      selector:
        text:
    intensity:
      name: Intensity
      description: Action intensity level.
      required: false
      default: 50
      selector:
        number:
          min: 0
          max: 100
          unit_of_measurement: "%"

5. Clean Up on Unload

async def async_unload_entry(
    hass: HomeAssistant,
    entry: ConfigEntry
) -> bool:
    """Unload a config entry."""
    # Remove all services
    for service in ["action1", "action2", "action3"]:
        hass.services.async_remove(DOMAIN, service)
    
    return True

Common Pitfalls

• Don’t forget schema validation (security and UX)
• Don’t block the event loop in service handlers
• Always handle missing entities gracefully
• Clean up services when integration unloads
• Use context propagation for state changes
• Document all services in services.yaml

Performance Tips

1. Use @callback for fast handlers - Avoid task creation overhead
2. Batch operations - Group multiple entity updates
3. Validate early - Reject invalid calls quickly
4. Use entity services - More efficient for entity operations
5. Limit blocking operations - Use executor for I/O

Related Resources

• Event Handling
• Async Programming
• State Management