call_service_tool

Instructions

Implementation Reference

• app/server.py:943-971 (handler)

  The `call_service_tool` handler function that executes the tool logic. It calls the underlying HA service via `call_service()` (imported from app.hass) and wraps the result in a dict with 'success', 'domain', 'service', and 'affected_entities' keys. Registered with @mcp.tool() decorator and @async_handler('call_service') logging decorator.

  @mcp.tool()
  @async_handler("call_service")
  async def call_service_tool(domain: str, service: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """
      Call any Home Assistant service (low-level API access)
      
      Args:
          domain: The domain of the service (e.g., 'light', 'switch', 'automation')
          service: The service to call (e.g., 'turn_on', 'turn_off', 'toggle')
          data: Optional data to pass to the service (e.g., {'entity_id': 'light.living_room'})
      
      Returns:
          A dictionary with success status, the domain/service called, and the
          list of affected entity states returned by Home Assistant.
  
      Examples:
          domain='light', service='turn_on', data={'entity_id': 'light.x', 'brightness': 255}
          domain='automation', service='reload'
          domain='fan', service='set_percentage', data={'entity_id': 'fan.x', 'percentage': 50}
  
      """
      logger.info(f"Calling Home Assistant service: {domain}.{service} with data: {data}")
      affected_entities = await call_service(domain, service, data or {})
      return {
          "success": True,
          "domain": domain,
          "service": service,
          "affected_entities": affected_entities,
      }

• app/hass.py:356-378 (helper)

  The underlying `call_service()` helper in app/hass.py that makes the actual HTTP POST to the Home Assistant REST API at `/api/services/{domain}/{service}`. Returns the list of affected entity states from the API response.

  @handle_api_errors
  async def call_service(domain: str, service: str, data: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
      """Call a Home Assistant service.
  
      Returns:
          List of affected entity states (may be empty for services like reload).
      """
      if data is None:
          data = {}
      
      client = await get_client()
      response = await client.post(
          f"{HA_URL}/api/services/{domain}/{service}", 
          headers=get_ha_headers(),
          json=data
      )
      response.raise_for_status()
      
      # Invalidate cache after service calls as they might change entity states
      global _entities_timestamp
      _entities_timestamp = 0
      
      return response.json()

• app/server.py:1-48 (registration)

  The `call_service` function from app.hass is imported at the top of server.py (line 18), making it available for the tool handler to call.

  import functools
  import logging
  import json
  import httpx
  from typing import List, Dict, Any, Optional, Callable, Awaitable, TypeVar, cast
  
  # Set up logging
  logging.basicConfig(
      level=logging.INFO,
      format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
      handlers=[
          logging.StreamHandler()
      ]
  )
  logger = logging.getLogger(__name__)
  
  from app.hass import (
      get_hass_version, get_entity_state, call_service, get_entities,
      get_automations, restart_home_assistant,
      cleanup_client, filter_fields, summarize_domain, get_system_overview,
      get_hass_error_log, get_entity_history, get_entity_history_range,
      get_entity_statistics, get_entity_statistics_range,
  )
  
  # Type variable for generic functions
  T = TypeVar('T')
  
  # Create an MCP server
  import os
  
  from mcp.server.fastmcp import FastMCP, Context, Image
  import mcp.types as types
  
  # When MCP_TRANSPORT is "streamable-http", configure for stateless operation
  # so the server works behind load balancers and on hosts like Smithery that
  # scale horizontally. Stateful mode is still the right default for local stdio.
  _http_mode = os.environ.get("MCP_TRANSPORT") == "streamable-http"
  
  mcp = FastMCP(
      "Hass-MCP",
      # Bind localhost by default. Override with MCP_HOST when running in Docker
      # or behind a reverse proxy. PORT (without prefix) is honored for Smithery
      # and other PaaS conventions.
      host=os.environ.get("MCP_HOST", "127.0.0.1"),
      port=int(os.environ.get("PORT", os.environ.get("MCP_PORT", "8000"))),
      stateless_http=_http_mode,
      json_response=_http_mode,
  )

• app/server.py:39-48 (registration)

  The tool is registered with MCP via the @mcp.tool() decorator on line 943, which is part of the FastMCP server instance created here.

  mcp = FastMCP(
      "Hass-MCP",
      # Bind localhost by default. Override with MCP_HOST when running in Docker
      # or behind a reverse proxy. PORT (without prefix) is honored for Smithery
      # and other PaaS conventions.
      host=os.environ.get("MCP_HOST", "127.0.0.1"),
      port=int(os.environ.get("PORT", os.environ.get("MCP_PORT", "8000"))),
      stateless_http=_http_mode,
      json_response=_http_mode,
  )