以下作品由安信可社區(qū)用戶

WT_0213制作

自從擁有了安信可AiPi-PalChatV1

Ai-M61+VC02做的語音控制器不香了

這是之前做的兩個(gè)版本

通過VC02控制HA燈光的小項(xiàng)目

Ai-M61+VC02語音控制HA設(shè)備?bbs.ai-thinker.com/forum.php?mod=viewthread&tid=45021

Ai-M61+VC02語音控制HA設(shè)備v1.1版?bbs.ai-thinker.com/forum.php?mod=viewthread&tid=45059

這兩個(gè)項(xiàng)目都是基于HA的自動(dòng)化來完成的。

參考：

[智能家居]MQTT控制HomeAssistant設(shè)備?bbs.ai-thinker.com/forum.php?mod=viewthread&tid=44644&fromuid=15918

現(xiàn)在都在鏈接智能化、大模型、MCP、AI，那就接入吧

不得不說蝦哥開源的這個(gè)小智真的太棒了

方案落地

下面介紹下如何通過AiPi-PalChatV1+MCP控制HA設(shè)備。

參考：

【AiPi-PalChatV1語音開發(fā)板】小智 MCP 接入 Home Assistant?bbs.ai-thinker.com/forum.php?mod=viewthread&tid=47027

1、下載部署小智MCP服務(wù)代碼

git clone https://gitee.com/lazy-ai/ai-pi-pal-chat-v1-ha.git

克隆代碼后，先運(yùn)行

pip install paho-mqtt
pip install -r requirements.txt
如果是mac
pip install paho-mqtt
pip install -r requirements_mac.txt

等待安裝完成即可

安裝好后如果報(bào) XXXX 模塊沒有找到，就繼續(xù)執(zhí)行 pip install xxx把缺失的部分安裝全就好了。

2、下載小智客戶端

https://github.com/huangjunsen0406/py-xiaozhi.git

github 地址可能間歇性的打不開，在gitee上面找了個(gè)

git clone https://gitee.com/tinytaro/py-xiaozhi0.git

克隆下來以后和上面基本一樣先安裝依賴

pip install -r requirements.txt
如果是mac
pip install -r requirements_mac.txt

這里需要注意的是有個(gè)包 **cv2**找不到

pip install opencv-python

還有一個(gè)修改點(diǎn)，就是將小智客戶端中的IoT相關(guān)代碼注釋掉，不然不走M(jìn)CP

Iot代碼在src下面iot目錄下

注釋 **application.py**第138行

# 初始化物聯(lián)網(wǎng)設(shè)備

# self._initialize_iot_devices()

這樣就好了然后進(jìn)到項(xiàng)目目錄打開命令行

執(zhí)行

python main.py

正常情況下他會(huì)播報(bào)一個(gè)設(shè)備碼，和AiPi-PalChatV1 配網(wǎng)時(shí)效果時(shí)是一樣的。將設(shè)備碼添加到智能體。

3、配置MCP接入點(diǎn)

打開 https://xiaozhi.me/

點(diǎn)擊**控制臺(tái)**, 登錄后

點(diǎn)擊**配置角色**，拉到屏幕最下方

右下角**MCP接入點(diǎn)**

復(fù)制接入點(diǎn)地址到第一步命令行

先執(zhí)行

export MCP_ENDPOINT=接入點(diǎn)地址

再執(zhí)行命令

python mcp_pipe.py switch_lamp.py

成功運(yùn)行輸出

(base) ? mcp git:(main) python mcp_pipe.py switch_lamp.py
2025-06-17 18:15:30,888 - MCP_PIPE - INFO - Connecting to WebSocket server...
2025-06-17 18:15:31,541 - MCP_PIPE - INFO - Successfully connected to WebSocket server
2025-06-17 18:15:31,548 - MCP_PIPE - INFO - Started switch_lamp.py process
/Users/tengyun1/AI/mcp-calculator/switch_lamp.py:30: DeprecationWarning: Callback API version 1 is deprecated, update to latest version
client = mqtt.Client() # 創(chuàng)建MQTT客戶端實(shí)例
[06/17/25 18:15:32] INFO Processing request of type server.py:523
ListToolsRequest
2025-06-17 18:15:32,397 - Home Assistant MCP - ERROR - MQTT連接: 0
ERROR MQTT連接: 0 switch_lamp.py:51
2025-06-17 18:15:32,845 - Home Assistant MCP - INFO - Connected with result code 0
INFO Connected with result code 0 switch_lamp.py:33

這樣就都跑起來了，然后就是測(cè)試

在會(huì)話框中輸入“開燈/關(guān)燈”,右側(cè)會(huì)同步限制MQTT接收到的信息。

下面是MCP 源代碼

# server.py
import sys
import json
import logging
import paho.mqtt.client as mqtt
import threading
from mcp.server.fastmcp import FastMCP
# 配置日志
logger = logging.getLogger('Home Assistant MCP')
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(http://logging.INFO)
# Fix UTF-8 encoding for Windows console
if sys.platform == 'win32':
sys.stderr.reconfigure(encoding='utf-8')
sys.stdout.reconfigure(encoding='utf-8')
# MQTT配置
MQTT_CONFIG = {
"broker_address": "http://broker.emqx.io",
"broker_port": 1883,
"username": None,
"password": None,
}
client = mqtt.Client() # 創(chuàng)建MQTT客戶端實(shí)例
def on_connect(client, userdata, flags, rc):
http://logger.info(f"Connected with result code {rc}")
client.publish("ha/ai", "Hello MQTT") # 連接后發(fā)布消息
client.subscribe("ha/ai") # 訂閱主題以接收消息
def on_message(client, userdata, message):
try:
payload = message.payload.decode()
except UnicodeDecodeError:
payload = f"<無法解碼的消息內(nèi)容: {message.payload}>"
http://logger.info(f"Received message: {payload} on topic {message.topic}")
def init_mqtt():
try:
client.on_connect = on_connect
client.on_message = on_message
if MQTT_CONFIG["username"] and MQTT_CONFIG["password"]:
client.username_pw_set(MQTT_CONFIG["username"], MQTT_CONFIG["password"])
ret = client.connect(MQTT_CONFIG["broker_address"], MQTT_CONFIG["broker_port"], 60)
logger.error(f"MQTT連接: {ret}")
client.loop_forever()
except Exception as e:
logger.error(f"MQTT連接失敗: {e}")
sys.exit(1)
# Create an MCP server
mcp = FastMCP("Home Assistant MCP")
# Add a lamp control tool
@mcp.tool()
def switchLamp(on_off: bool) -> dict:
"""這是一個(gè)用于控制HomeAssistant的燈的接口"""
result = "開燈成功" if on_off else "關(guān)燈成功"
http://logger.info(f"switchLamp formula: {on_off}, result: {result}")
status = 1 if on_off else 0
payload = json.dumps({"status": status})
client.publish("ha/ai", payload) # 連接后發(fā)布消息
return {"success": True, "result": result}
# Start the server
if __name__ == "__main__":
# 啟動(dòng) MQTT 客戶端在單獨(dú)的線程中
mqtt_thread = threading.Thread(target=init_mqtt, daemon=True)
mqtt_thread.start()
mcp.run(transport="stdio")


審核編輯 黃宇