Building a Smart Voice Assistant: Integrating Home Assistant MCP with LangGraph

Voice control and natural language interaction are becoming essential in the smart home era. This article details how to integrate Home Assistant's MCP (Model Context Protocol) service into a smart voice assistant using the LangGraph framework, enabling true natural language home automation. The focus is on practical code and step-by-step guidance.

GitHub: mawwalker/Moss — Star and join the discussion!

Introduction

Traditional smart home control often relies on fixed commands, limiting user experience. With the rise of large language models (LLMs) and frameworks like LangGraph, we can empower voice assistants to understand natural language and deeply integrate with Home Assistant via the MCP protocol for flexible, powerful home control.

MCP (Model Context Protocol) is an open standard that allows LLMs to securely connect to external data sources and tools. In this project, the Moss assistant uses an MCP server to directly invoke Home Assistant features, enabling device status queries, scene control, and more.

1. Architecture Overview

The core goal is to let a LangGraph Agent communicate seamlessly with Home Assistant via MCP. The architecture includes:

• LangGraph Agent: Handles natural language understanding and reasoning
• MCP Tool Integration: Uses langchain_mcp_tools to expose Home Assistant as standard tools
• Async Resource Management: Uses async context managers to ensure proper MCP connection lifecycle

2. Integrating Home Assistant MCP Tools

1. Building the HASS MCP Tool

We use the langchain_mcp_tools package to convert Home Assistant's MCP service into LangChain-compatible tools. Core code:

from langchain_mcp_tools import convert_mcp_to_langchain_tools
from loguru import logger
import os
from config.conf import hass_config, PROJECT_ROOT

async def build_hass_tools() -> tuple:
    try:
        hass_mcp_config = {
            "homeassistant": {
                "transport": "stdio",
                "command": "uv",
                "args": [
                    "run",
                    "mcp",
                    "run",
                    "hass-mcp/app/server.py",
                ],
                "env": {
                    "HA_URL": hass_config["hass_url"],
                    "HA_TOKEN": hass_config["token"],
                    "PYTHONPATH": os.path.join(PROJECT_ROOT, "hass-mcp"),
                }
            }
        }
        tools, cleanup = await convert_mcp_to_langchain_tools(hass_mcp_config)
        return tools, cleanup
    except Exception as e:
        logger.error(f"Failed to build HASS tools: {e}")
        return [], lambda: None

Notes:

• The hass_mcp_config specifies how to launch the Home Assistant MCP service and required environment variables (HA_URL, HA_TOKEN, etc).
• convert_mcp_to_langchain_tools automatically exposes all MCP features as LangChain tools.
• The returned tools can be used directly in the agent; cleanup is for resource release.

2. Tool Initialization and Agent Integration

When initializing the agent, dynamically load HASS tools and support automatic resource cleanup:

from agent_core.hass_mcp import build_hass_tools
from config.conf import hass_config
from loguru import logger

async def init_tools():
    tools = []
    cleanup_funcs = []
    if hass_config["enable"]:
        logger.debug("Initializing HASS tools...")
        try:
            hass_tools, cleanup = await build_hass_tools()
            tools.extend(hass_tools)
            cleanup_funcs.append(cleanup)
        except Exception as e:
            logger.error(f"Failed to initialize HASS tools: {e}")
    logger.info(f"Initialized tools: {[tool.name for tool in tools]}")
    return tools, cleanup_funcs

3. LangGraph Agent and MCP Tool Collaboration

MossAgent uses an async context manager to ensure safe MCP connection lifecycle. The agent loads HASS tools at startup and supports streaming responses and dynamic system prompts.

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from agent_core.tools import init_tools
from agent_core.prompts import agent_additional_prompts
from langchain_core.runnables import RunnableConfig
from loguru import logger
import asyncio
from datetime import datetime

class MossAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="your-model",
            temperature=0.1,
            api_key="your-api-key",
            base_url="your-base-url",
        )
        self.agent = None
        self._initialized = False
        self._cleanup_funcs = []
    async def _initialize(self):
        if not self._initialized:
            tools, cleanup_funcs = await init_tools()
            self._cleanup_funcs.extend(cleanup_funcs)
            self.agent = create_react_agent(
                model=self.llm,
                tools=tools,
                prompt=self.dynamic_system_prompt
            )
            self._initialized = True
    async def cleanup(self):
        for cleanup_func in self._cleanup_funcs:
            try:
                if asyncio.iscoroutinefunction(cleanup_func):
                    await cleanup_func()
                else:
                    cleanup_func()
            except Exception as e:
                logger.error(f"Error during cleanup: {e}")
        self._cleanup_funcs.clear()
    async def __aenter__(self):
        await self._initialize()
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.cleanup()
    @staticmethod
    def dynamic_system_prompt(state, config):
        current_date = config.get("configurable", {}).get("current_date", "")
        language = config.get("configurable", {}).get("language", "en")
        system_msg = agent_additional_prompts.format(
            current_date=current_date,
            language=language
        )
        return [{"role": "system", "content": system_msg}] + state["messages"]
    async def run(self, messages, config=None):
        await self._initialize()
        result = self.agent.astream(
            {"messages": messages},
            stream_mode=["updates", "messages", "custom"],
            config=config,
        )
        async for stream_mode, chunk in result:
            if stream_mode == "messages":
                content = chunk[0].content
                yield content

Highlights:

• Async context management with automatic MCP cleanup
• Dynamic system prompts for multi-language and context adaptation
• Streaming responses for better user experience

4. Usage Example

Assuming Home Assistant and MCP are configured, here's a simple usage example:

import asyncio
from datetime import datetime
from langchain_core.runnables import RunnableConfig

messages = [
    {"role": "user", "content": "Turn off the bedside lamp"},
]

async def main():
    configurable = {
        "current_date": datetime.now().strftime("%Y-%m-%d"),
        "language": "English"
    }
    async with MossAgent() as agent:
        async for chunk in agent.run(messages, config=RunnableConfig(configurable=configurable)):
            print(chunk)

asyncio.run(main())

5. MCP Integration Advantages

1. Standardized Interface: MCP provides a unified protocol between LLMs and Home Assistant, greatly simplifying integration.
2. Real-time Communication: Supports bidirectional real-time communication; device state changes are instantly reflected.
3. Security: Built-in security mechanisms ensure safe data and operations.
4. High Extensibility: To add new Home Assistant features, just update the MCP server—no need to modify agent code.

6. Conclusion

By deeply integrating LangGraph and the MCP protocol, the Moss smart voice assistant achieves true natural language home control. Whether it's device switching, scene activation, or status queries, everything can be done smoothly via natural language. MCP makes the system more standard, secure, and extensible.

If you're building a smart home voice assistant, LangGraph + MCP is highly recommended. Feel free to reach out, ask questions, or star the project!