Service Registration

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"}
        )

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

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
)

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

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
)

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
)

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
)

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

# 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
)

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

# 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"]

# 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"
    }
)

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()

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,
})

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

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)

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)

# 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

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
    )

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()

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
)

# 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
)

# 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

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

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: "%"

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