diff --git a/.gitignore b/.gitignore
index fac29de..cf5ce3b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,3 +47,4 @@ __pycache__
 .mi.json
 xiaozhi.json
 open-xiaoai.node
+examples/hass/config.json
diff --git a/examples/hass/Cargo.toml b/examples/hass/Cargo.toml
new file mode 100644
index 0000000..1b1fa4f
--- /dev/null
+++ b/examples/hass/Cargo.toml
@@ -0,0 +1,18 @@
+[package]
+name = "open_xiaoai_server"
+version = "1.0.0"
+edition = "2021"
+
+[lib]
+name = "open_xiaoai_server"
+crate-type = ["cdylib"]
+
+[dependencies]
+pyo3 = { version = "0.24.0", features = ["extension-module", "abi3-py38"] }
+pyo3-async-runtimes = { version = "0.24", features = ["attributes", "tokio-runtime"] }
+open-xiaoai = { path = "../../packages/client-rust" }
+tokio = { version = "1.32.0", features = ["full"] }
+tokio-tungstenite = "0.26.2"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
diff --git a/examples/hass/Dockerfile b/examples/hass/Dockerfile
new file mode 100644
index 0000000..2f64c01
--- /dev/null
+++ b/examples/hass/Dockerfile
@@ -0,0 +1,53 @@
+FROM python:3.12-slim AS builder
+
+SHELL ["/bin/bash", "-o", "pipefail", "-c"]
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    curl build-essential pkg-config libssl-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
+
+ENV BASH_ENV=/root/.bash_env
+RUN touch "$BASH_ENV"
+RUN echo '. "$BASH_ENV"' >> "$HOME/.bashrc"
+RUN echo '[ -s "$HOME/.cargo/env" ] && . "$HOME/.cargo/env"' >> "$BASH_ENV"
+
+WORKDIR /app
+
+COPY examples/hass/hass ./hass
+COPY examples/hass/src ./src
+COPY examples/hass/Cargo.toml ./Cargo.toml
+COPY examples/hass/pyproject.toml ./pyproject.toml
+COPY examples/hass/README.md ./README.md
+COPY examples/hass/requirements.txt ./requirements.txt
+COPY examples/hass/hass_assistant.py ./hass_assistant.py
+COPY examples/hass/config.example.json ./config.example.json
+COPY packages/client-rust ./client-rust
+
+RUN python -m venv /opt/venv
+ENV VIRTUAL_ENV=/opt/venv
+ENV PATH="/opt/venv/bin:${PATH}"
+
+RUN pip install --upgrade pip
+RUN pip install -r requirements.txt
+RUN pip install maturin
+
+RUN sed -i 's/\.\.\/\.\.\/packages\///g' Cargo.toml \
+    && maturin develop --release
+
+
+
+FROM python:3.12-slim
+
+ENV VIRTUAL_ENV=/opt/venv
+ENV PATH="/opt/venv/bin:${PATH}"
+
+WORKDIR /app
+
+COPY --from=builder /opt/venv /opt/venv
+COPY --from=builder /app/hass /app/hass
+COPY --from=builder /app/hass_assistant.py /app/hass_assistant.py
+COPY --from=builder /app/config.example.json /app/config.example.json
+
+CMD ["python", "hass_assistant.py"]
diff --git a/examples/hass/README.md b/examples/hass/README.md
new file mode 100644
index 0000000..7568627
--- /dev/null
+++ b/examples/hass/README.md
@@ -0,0 +1,125 @@
+# Open-XiaoAI x Home Assistant Assist
+
+该示例用于将小爱音箱接入 Home Assistant 的对话/语音助手能力，支持文本对话与上下文会话，并提供可选的语音输入管线对接能力。
+
+## 功能说明
+
+- 文本对话：对接 Home Assistant `conversation.process` API，支持自然语言控制设备/查询信息
+- 实时打断：接收到新用户消息时先中断当前播报，再处理新输入，避免并发播报
+- 连续对话会话管理：默认开启连续对话；当命中结束词或语音静默超时时自动结束并重建 `conversation_id`
+- 区域上下文注入：支持客户端 IP 到 Home Assistant 区域映射，命中后在每次请求前注入区域前置提示词
+- 小爱音箱 I/O：从小爱音箱事件中提取最终识别文本（ASR），将回复通过小爱本机 TTS 播放
+- 可选音频能力（扩展点）：支持接收小爱录音流 `record`，用于后续对接 Assist pipeline（STT→Intent→TTS）
+
+## 使用场景
+
+- 用中文/英文自然语言控制 Home Assistant：开灯、关灯、调亮度、问温度/天气、查询传感器状态
+- 把小爱音箱作为 Home Assistant 的“外置麦克风 + 扬声器”（局域网内运行）
+## 架构概览
+
+- 小爱音箱 →（WebSocket 4399）→ Rust 网关（`open_xiaoai_server`）→ Python 回调
+- 文本链路（默认，低延迟）：
+  - 设备事件 `instruction/NewLine` → 抽取 `SpeechRecognizer.RecognizeResult` 文本
+  - 调用 HA `/api/conversation/process`
+  - 把返回的 `response.speech.plain.speech` 交给小爱 TTS 播报
+
+## 目录结构
+
+本目录包含：
+
+- `hass_assistant.py`：主入口
+- `config.example.json`：配置模板
+- `requirements.txt`：Python 依赖
+- `src/`：Rust(PyO3) 扩展模块（小爱音箱 I/O 网关）
+- `hass/`：Python 实现（HA 客户端、对话编排、上下文记忆）
+- `tests/`：单元测试
+- `scripts/`：集成测试脚本
+
+## 安装与运行
+
+### 前置条件
+
+- 你的小爱音箱已刷机并运行 Rust 客户端补丁（否则服务端收不到事件/音频流）
+- Home Assistant 已启用 `conversation`（默认配置一般已启用）并生成 Long-Lived Access Token
+- 本机具备 Python（建议 3.10+）与 Rust 工具链（用于编译 PyO3 扩展）
+- 可选：如果你希望使用 webrtcvad（更好的语音端点检测），Windows 需要安装 MSVC Build Tools
+
+### 1) 准备配置
+
+复制配置模板：
+
+```bash
+cd examples/hass
+cp config.example.json config.json
+```
+
+编辑 `config.json`：
+
+- `homeassistant.url`：HA 地址（例如 `http://192.168.1.10:8123`）
+- `homeassistant.access_token`：HA Long-Lived Access Token
+- `homeassistant.assistant_entity_id`：对话助理 ID（对应 `agent_id`，默认 `home_assistant`）
+
+### 配置新增项说明
+
+`xiaoai` 段新增以下关键配置：
+
+- `interrupt_on_new_input`：是否在收到新消息时立即执行打断，默认 `true`
+- `interrupt_command`：打断命令，默认使用 `ubus call mibrain ai_service` 触发中断
+- `continuous_conversation_enabled`：是否开启连续会话，默认 `true`
+- `session_idle_timeout_seconds`：连续会话静默超时秒数，默认 `15`
+- `session_end_keywords`：结束词列表，默认包含 `再见` `拜拜` `滚` `退下`，按包含匹配
+- `client_ip_area_mapping`：客户端 IP 到区域名称映射，仅在命中时注入区域提示词
+
+区域提示词注入内容为：
+
+`当前用户对话所在区域：<区域名称>，如后续对话未明确指定区域，则默认为此区域`
+
+### 2) 安装依赖
+
+```bash
+pip install -r requirements.txt
+```
+
+### 3) 编译 Rust 扩展（open_xiaoai_server）
+
+推荐使用 maturin：
+
+```bash
+pip install maturin
+maturin develop --release
+```
+
+### 4) 运行
+
+```bash
+python hass_assistant.py
+```
+
+## API 调用示例
+
+### Home Assistant conversation.process（curl）
+
+```bash
+curl -X POST \
+  -H "Authorization: Bearer YOUR_LONG_LIVED_ACCESS_TOKEN" \
+  -H "Content-Type: application/json" \
+  http://HOME_ASSISTANT:8123/api/conversation/process \
+  -d '{"text":"打开客厅灯","language":"zh-cn","agent_id":"home_assistant"}'
+```
+
+### Python 调用（不连接小爱，仅验证 HA）
+
+可参考 `scripts/integration_text.py`。
+## 故障排除
+
+- 401/403：检查 `access_token` 是否正确、是否完整复制
+- 连接超时：检查 `homeassistant.url` 是否可达、是否被防火墙/反代阻断
+- 无语音输入：确认小爱已运行客户端补丁且能连到本机 4399 端口
+- 无语音输出：确认小爱系统 `ubus`/`mibrain` 可用（或改用 `play_url` 播放 URL）
+- 对话不连贯：确认 `state.json` 能写入（用于持久化 `conversation_id`），或删除它重置会话
+- 打断不生效：检查 `xiaoai.interrupt_command` 是否可在设备上执行，以及日志中是否出现 `xiaoai_interrupt_failed`
+- 区域提示词未注入：检查 `client_ip_area_mapping` 是否包含当前客户端 IP，并查看 `area_mapping_hit/area_mapping_miss` 日志
+## 部署说明
+
+- 建议仅在局域网内部署运行，避免将 4399 暴露到公网
+- 如果 HA 与运行本示例的机器不在同一网段，确保 HA 的 URL 在设备侧可访问（尤其是播放 URL 时）
diff --git a/examples/hass/config.example.json b/examples/hass/config.example.json
new file mode 100644
index 0000000..31fddc9
--- /dev/null
+++ b/examples/hass/config.example.json
@@ -0,0 +1,36 @@
+{
+  "homeassistant": {
+    "url": "http://homeassistant.local:8123",
+    "access_token": "YOUR_LONG_LIVED_ACCESS_TOKEN",
+    "assistant_entity_id": "home_assistant",
+    "language": "zh-cn",
+    "conversation_id": null,
+    "timeout_seconds": 2.5
+  },
+  "xiaoai": {
+    "tts_blocking": false,
+    "shell_timeout_ms": 600000,
+    "interrupt_on_new_input": true,
+    "interrupt_command": "ubus call mibrain ai_service '{\"nlp\":1,\"nlp_text\":\"\"}'",
+    "continuous_conversation_enabled": true,
+    "session_idle_timeout_seconds": 15,
+    "session_end_keywords": ["再见", "拜拜", "滚", "退下"],
+    "client_ip_area_mapping": {
+      "192.168.1.101": "客厅",
+      "192.168.1.102": "主卧"
+    }
+  },
+  "runtime": {
+    "concurrency": 4,
+    "log_level": "INFO",
+    "persist_state_path": "state.json"
+  },
+  "voice": {
+    "enabled": false,
+    "mode": "record_vad_pipeline",
+    "vad_aggressiveness": 2,
+    "min_silence_ms": 800,
+    "max_utterance_ms": 15000,
+    "sample_rate": 16000
+  }
+}
diff --git a/examples/hass/hass/__init__.py b/examples/hass/hass/__init__.py
new file mode 100644
index 0000000..ced2eec
--- /dev/null
+++ b/examples/hass/hass/__init__.py
@@ -0,0 +1,10 @@
+__all__ = [
+    "config_loader",
+    "errors",
+    "ha_client",
+    "memory",
+    "typing",
+    "vad",
+    "xiaoai_bridge",
+]
+
diff --git a/examples/hass/hass/config_loader.py b/examples/hass/hass/config_loader.py
new file mode 100644
index 0000000..494e9ab
--- /dev/null
+++ b/examples/hass/hass/config_loader.py
@@ -0,0 +1,10 @@
+import json
+from pathlib import Path
+
+from hass.typing import AppConfig
+
+
+def load_config(path: str) -> AppConfig:
+    data = json.loads(Path(path).read_text(encoding="utf-8"))
+    return AppConfig.model_validate(data)
+
diff --git a/examples/hass/hass/errors.py b/examples/hass/hass/errors.py
new file mode 100644
index 0000000..2a47523
--- /dev/null
+++ b/examples/hass/hass/errors.py
@@ -0,0 +1,19 @@
+class HassAssistantError(Exception):
+    pass
+
+
+class HomeAssistantRequestError(HassAssistantError):
+    pass
+
+
+class HomeAssistantAuthError(HomeAssistantRequestError):
+    pass
+
+
+class HomeAssistantResponseError(HomeAssistantRequestError):
+    pass
+
+
+class XiaoAiBridgeError(HassAssistantError):
+    pass
+
diff --git a/examples/hass/hass/ha_client.py b/examples/hass/hass/ha_client.py
new file mode 100644
index 0000000..a258fa7
--- /dev/null
+++ b/examples/hass/hass/ha_client.py
@@ -0,0 +1,161 @@
+import asyncio
+from typing import Optional
+
+import aiohttp
+
+from hass.errors import (
+    HomeAssistantAuthError,
+    HomeAssistantRequestError,
+    HomeAssistantResponseError,
+)
+from hass.memory import ConversationMemory
+from hass.typing import AssistantReply, HomeAssistantConfig
+
+
+class HomeAssistantClient:
+    def __init__(
+        self, config: HomeAssistantConfig, memory: ConversationMemory, concurrency: int = 4
+    ) -> None:
+        self._config = config
+        self._memory = memory
+        self._session: Optional[aiohttp.ClientSession] = None
+        self._lock = asyncio.Lock()
+        self._semaphore = asyncio.Semaphore(max(1, int(concurrency)))
+
+    async def start(self) -> None:
+        if self._session is not None:
+            return
+        headers = {"Authorization": f"Bearer {self._config.access_token}"}
+        timeout = aiohttp.ClientTimeout(total=self._config.timeout_seconds)
+        self._session = aiohttp.ClientSession(headers=headers, timeout=timeout)
+
+    async def close(self) -> None:
+        if self._session is None:
+            return
+        await self._session.close()
+        self._session = None
+
+    async def reset_conversation(self) -> None:
+        self._memory.conversation_id = None
+        await self._memory.save()
+
+    async def process_text(self, text: str, prefix_prompt: Optional[str] = None) -> AssistantReply:
+        if self._session is None:
+            raise HomeAssistantRequestError("HomeAssistantClient is not started")
+
+        request_text = text
+        if prefix_prompt:
+            request_text = f"{prefix_prompt}\n用户请求：{text}"
+
+        payload = {"text": request_text}
+        if self._config.language:
+            payload["language"] = self._config.language
+        if self._config.agent_id:
+            payload["agent_id"] = self._config.agent_id
+
+        conversation_id = self._memory.conversation_id or self._config.conversation_id
+        if conversation_id:
+            payload["conversation_id"] = conversation_id
+
+        url = str(self._config.url).rstrip("/") + "/api/conversation/process"
+        async with self._semaphore:
+            async with (
+                self._lock
+                if (self._memory.conversation_id or self._config.conversation_id)
+                else _NullAsyncContext()
+            ):
+                data = None
+                reset_attempts = 2 if conversation_id else 1
+                for reset_index in range(reset_attempts):
+                    if reset_index == 1:
+                        payload.pop("conversation_id", None)
+                        if self._memory.conversation_id:
+                            self._memory.conversation_id = None
+                            await self._memory.save()
+                    attempts = 3
+                    delay = 0.5
+                    for i in range(attempts):
+                        try:
+                            timeout_seconds = (
+                                self._config.timeout_seconds
+                                if i == 0
+                                else max(self._config.timeout_seconds, 8.0)
+                            )
+                            request_timeout = aiohttp.ClientTimeout(total=timeout_seconds)
+                            async with self._session.post(
+                                url,
+                                json=payload,
+                                timeout=request_timeout,
+                            ) as resp:
+                                if resp.status in (401, 403):
+                                    raise HomeAssistantAuthError(
+                                        f"Home Assistant auth failed: {resp.status}"
+                                    )
+                                if resp.status >= 500:
+                                    raise HomeAssistantRequestError(
+                                        f"Home Assistant server error: {resp.status}"
+                                    )
+                                if resp.status >= 400:
+                                    raise HomeAssistantRequestError(
+                                        f"Home Assistant request failed: {resp.status}"
+                                    )
+                                data = await resp.json()
+                                break
+                        except asyncio.TimeoutError:
+                            if i < attempts - 1:
+                                await asyncio.sleep(delay)
+                                delay *= 2
+                                continue
+                            error = HomeAssistantRequestError(
+                                f"Home Assistant request timeout, timeout_seconds={max(self._config.timeout_seconds, 8.0)}"
+                            )
+                        except aiohttp.ClientError as e:
+                            if i < attempts - 1:
+                                await asyncio.sleep(delay)
+                                delay *= 2
+                                continue
+                            error = HomeAssistantRequestError(str(e))
+                        else:
+                            error = None
+                        if error is not None:
+                            if reset_index == reset_attempts - 1:
+                                raise error
+                            break
+                    if data is not None:
+                        break
+
+        if not isinstance(data, dict):
+            raise HomeAssistantResponseError("Unexpected response payload type")
+
+        new_conversation_id = data.get("conversation_id")
+        if new_conversation_id:
+            self._memory.conversation_id = new_conversation_id
+            await self._memory.save()
+
+        response = data.get("response", {}) if isinstance(data.get("response", {}), dict) else {}
+        speech = (
+            response.get("speech", {}).get("plain", {}).get("speech", "")
+            if isinstance(response.get("speech", {}), dict)
+            else ""
+        )
+        if not isinstance(speech, str):
+            raise HomeAssistantResponseError("Unexpected response speech type")
+        if not speech:
+            response_type = response.get("response_type")
+            speech = "好的" if response_type in ("action_done", "query_answer") else "出错了"
+
+        continue_conversation = bool(data.get("continue_conversation", False))
+        return AssistantReply(
+            speech=speech,
+            continue_conversation=continue_conversation,
+            conversation_id=new_conversation_id,
+            raw=data,
+        )
+
+
+class _NullAsyncContext:
+    async def __aenter__(self):
+        return None
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
diff --git a/examples/hass/hass/memory.py b/examples/hass/hass/memory.py
new file mode 100644
index 0000000..2867770
--- /dev/null
+++ b/examples/hass/hass/memory.py
@@ -0,0 +1,22 @@
+import json
+from pathlib import Path
+from typing import Optional
+
+
+class ConversationMemory:
+    def __init__(self, state_path: str) -> None:
+        self._path = Path(state_path)
+        self.conversation_id: Optional[str] = None
+
+    async def load(self) -> None:
+        if not self._path.exists():
+            return
+        data = json.loads(self._path.read_text(encoding="utf-8"))
+        self.conversation_id = data.get("conversation_id")
+
+    async def save(self) -> None:
+        self._path.write_text(
+            json.dumps({"conversation_id": self.conversation_id}, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
diff --git a/examples/hass/hass/typing.py b/examples/hass/hass/typing.py
new file mode 100644
index 0000000..299f94a
--- /dev/null
+++ b/examples/hass/hass/typing.py
@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, Field, HttpUrl, model_validator
+
+
+class HomeAssistantConfig(BaseModel):
+    url: HttpUrl
+    access_token: str
+    agent_id: str = "home_assistant"
+    assistant_entity_id: Optional[str] = None
+    language: Optional[str] = None
+    conversation_id: Optional[str] = None
+    timeout_seconds: float = 2.5
+
+    @model_validator(mode="after")
+    def _normalize_agent_id(self):
+        if self.assistant_entity_id:
+            self.agent_id = self.assistant_entity_id
+        return self
+
+
+class XiaoAiConfig(BaseModel):
+    tts_blocking: bool = False
+    shell_timeout_ms: int = 10 * 60 * 1000
+    interrupt_on_new_input: bool = True
+    interrupt_command: str = "ubus call mibrain ai_service '{\"nlp\":1,\"nlp_text\":\"\"}'"
+    continuous_conversation_enabled: bool = True
+    session_idle_timeout_seconds: int = 15
+    session_end_keywords: list[str] = Field(
+        default_factory=lambda: ["再见", "拜拜", "滚", "退下"]
+    )
+    client_ip_area_mapping: dict[str, str] = Field(default_factory=dict)
+
+
+class RuntimeConfig(BaseModel):
+    concurrency: int = 4
+    log_level: str = "INFO"
+    persist_state_path: str = "state.json"
+
+
+class VoiceConfig(BaseModel):
+    enabled: bool = False
+    mode: Literal["record_vad_pipeline"] = "record_vad_pipeline"
+    vad_aggressiveness: int = 2
+    min_silence_ms: int = 800
+    max_utterance_ms: int = 15000
+    sample_rate: int = 16000
+
+
+class AppConfig(BaseModel):
+    homeassistant: HomeAssistantConfig
+    xiaoai: XiaoAiConfig = Field(default_factory=XiaoAiConfig)
+    runtime: RuntimeConfig = Field(default_factory=RuntimeConfig)
+    voice: VoiceConfig = Field(default_factory=VoiceConfig)
+
+
+class AssistantReply(BaseModel):
+    speech: str
+    continue_conversation: bool = False
+    conversation_id: Optional[str] = None
+    raw: dict = Field(default_factory=dict)
+
+
+class RecognizedSpeechInput(BaseModel):
+    text: str
+    client_ip: Optional[str] = None
diff --git a/examples/hass/hass/vad.py b/examples/hass/hass/vad.py
new file mode 100644
index 0000000..12cea6b
--- /dev/null
+++ b/examples/hass/hass/vad.py
@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+try:
+    import webrtcvad
+except Exception:
+    webrtcvad = None
+
+
+@dataclass(frozen=True)
+class VadConfig:
+    aggressiveness: int = 2
+    min_silence_ms: int = 800
+    max_utterance_ms: int = 15000
+    sample_rate: int = 16000
+    energy_threshold: int = 450
+
+
+class VadSegmenter:
+    def __init__(self, config: VadConfig) -> None:
+        self._config = config
+        self._vad = webrtcvad.Vad(int(config.aggressiveness)) if webrtcvad else None
+        self._frame_ms = 20
+        self._frame_bytes = int(config.sample_rate * self._frame_ms / 1000) * 2
+        self._buffer = bytearray()
+        self._in_speech = False
+        self._speech = bytearray()
+        self._silence_ms = 0
+        self._speech_ms = 0
+
+    def feed(self, pcm: bytes) -> list[bytes]:
+        if not pcm:
+            return []
+
+        self._buffer.extend(pcm)
+        utterances: list[bytes] = []
+
+        while len(self._buffer) >= self._frame_bytes:
+            frame = bytes(self._buffer[: self._frame_bytes])
+            del self._buffer[: self._frame_bytes]
+
+            if self._vad is not None:
+                is_speech = self._vad.is_speech(frame, self._config.sample_rate)
+            else:
+                samples = memoryview(frame).cast("h")
+                avg = sum(abs(x) for x in samples) / max(1, len(samples))
+                is_speech = avg >= self._config.energy_threshold
+            if is_speech:
+                if not self._in_speech:
+                    self._in_speech = True
+                    self._speech.clear()
+                    self._silence_ms = 0
+                    self._speech_ms = 0
+                self._speech.extend(frame)
+                self._speech_ms += self._frame_ms
+                self._silence_ms = 0
+            else:
+                if self._in_speech:
+                    self._speech.extend(frame)
+                    self._speech_ms += self._frame_ms
+                    self._silence_ms += self._frame_ms
+
+            if self._in_speech and (
+                self._silence_ms >= self._config.min_silence_ms
+                or self._speech_ms >= self._config.max_utterance_ms
+            ):
+                utterances.append(bytes(self._speech))
+                self._in_speech = False
+                self._speech.clear()
+                self._silence_ms = 0
+                self._speech_ms = 0
+
+        return utterances
+
+    def flush(self) -> list[bytes]:
+        if not self._in_speech or not self._speech:
+            return []
+        utterance = bytes(self._speech)
+        self._in_speech = False
+        self._speech.clear()
+        self._silence_ms = 0
+        self._speech_ms = 0
+        return [utterance]
diff --git a/examples/hass/hass/xiaoai_bridge.py b/examples/hass/hass/xiaoai_bridge.py
new file mode 100644
index 0000000..2223a7b
--- /dev/null
+++ b/examples/hass/hass/xiaoai_bridge.py
@@ -0,0 +1,243 @@
+import asyncio
+import json
+import logging
+import re
+from collections.abc import Awaitable, Callable
+from typing import Optional
+import time
+
+from hass.errors import XiaoAiBridgeError
+from hass.typing import RecognizedSpeechInput, XiaoAiConfig
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class XiaoAiBridge:
+    def __init__(self, config: XiaoAiConfig) -> None:
+        self._config = config
+        self._on_text: Optional[Callable[[RecognizedSpeechInput], Awaitable[None]]] = None
+        self._on_audio: Optional[Callable[[bytes], None]] = None
+        self._ready = asyncio.Event()
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._last_text: Optional[str] = None
+        self._last_client_ip: Optional[str] = None
+        self._last_text_ts: float = 0.0
+
+    def set_on_text(self, fn: Callable[[RecognizedSpeechInput], Awaitable[None]]) -> None:
+        self._on_text = fn
+
+    def set_on_audio(self, fn: Callable[[bytes], None]) -> None:
+        self._on_audio = fn
+
+    async def start(self) -> None:
+        try:
+            import open_xiaoai_server
+        except Exception as e:
+            raise XiaoAiBridgeError(
+                "open_xiaoai_server module is not available (Rust extension not built)"
+            ) from e
+
+        self._open_xiaoai_server = open_xiaoai_server
+        self._loop = asyncio.get_running_loop()
+        self._register_callbacks()
+        asyncio.ensure_future(self._open_xiaoai_server.start_server())
+        self._ready.set()
+
+    def _register_callbacks(self) -> None:
+        def on_event(event_json: str) -> None:
+            # Rust 侧会把 Event 结构序列化成 JSON 字符串传入，这里做反序列化与容错解析
+            try:
+                event = json.loads(event_json)
+            except Exception:
+                return
+
+            if event.get("event") != "instruction":
+                return
+
+            data = event.get("data") or {}
+            line_text = data.get("NewLine")
+            if not line_text:
+                return
+
+            # NewLine 是设备端透传的 JSON 文本，包含语音识别结果等指令
+            try:
+                line = json.loads(line_text)
+            except Exception:
+                return
+
+            header = line.get("header") or {}
+            if header.get("namespace") != "SpeechRecognizer":
+                return
+            if header.get("name") != "RecognizeResult":
+                return
+
+            payload = line.get("payload") or {}
+            if not payload.get("is_final"):
+                return
+
+            results = payload.get("results") or []
+            if not results or not isinstance(results, list):
+                return
+
+            text = results[0].get("text") if isinstance(results[0], dict) else None
+            if not text or not isinstance(text, str):
+                return
+
+            text = text.strip()
+            if not text:
+                return
+
+            if self._on_text is None or self._loop is None:
+                return
+
+            recognized = RecognizedSpeechInput(
+                text=text,
+                client_ip=self._extract_client_ip(event=event, data=data, line=line, payload=payload),
+            )
+
+            now = time.monotonic()
+            if (
+                self._last_text == recognized.text
+                and self._last_client_ip == (recognized.client_ip or "")
+                and (now - self._last_text_ts) < 1.2
+            ):
+                _LOGGER.info(
+                    "xiaoai_duplicate_text_skipped text=%s client_ip=%s",
+                    recognized.text,
+                    recognized.client_ip or "",
+                )
+                return
+
+            self._last_text = recognized.text
+            self._last_client_ip = recognized.client_ip or ""
+            self._last_text_ts = now
+
+            async def _dispatch() -> None:
+                if self._config.interrupt_on_new_input:
+                    await self.interrupt_playback(reason="new_input")
+                await self._on_text(recognized)
+
+            if not recognized.client_ip:
+                _LOGGER.info("xiaoai_client_ip_missing")
+
+            # 回调来自 Rust 的 tokio 线程，需切回 Python 主事件循环执行异步对话逻辑
+            asyncio.run_coroutine_threadsafe(_dispatch(), self._loop)
+
+        def on_input_data(data: bytes) -> None:
+            if self._on_audio is None:
+                return
+            self._on_audio(data)
+
+        self._open_xiaoai_server.register_fn("on_event", on_event)
+        self._open_xiaoai_server.register_fn("on_input_data", on_input_data)
+
+    @staticmethod
+    def _extract_client_ip(
+        event: dict,
+        data: dict,
+        line: dict,
+        payload: dict,
+    ) -> Optional[str]:
+        def _normalize_ip(value: object) -> Optional[str]:
+            if not isinstance(value, str):
+                return None
+            text = value.strip()
+            if not text:
+                return None
+            if ":" in text:
+                host, _, maybe_port = text.rpartition(":")
+                if host and maybe_port.isdigit():
+                    text = host
+            match = re.search(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", text)
+            if match:
+                return match.group(0)
+            return None
+
+        def _walk(obj: object) -> Optional[str]:
+            if isinstance(obj, dict):
+                for key, value in obj.items():
+                    key_text = str(key).lower()
+                    if any(token in key_text for token in ("ip", "addr", "remote")):
+                        normalized = _normalize_ip(value)
+                        if normalized:
+                            return normalized
+                    nested = _walk(value)
+                    if nested:
+                        return nested
+                return None
+            if isinstance(obj, list):
+                for item in obj:
+                    nested = _walk(item)
+                    if nested:
+                        return nested
+                return None
+            return _normalize_ip(obj)
+
+        candidates = [
+            data.get("ClientIP"),
+            data.get("client_ip"),
+            data.get("ip"),
+            data.get("remote_addr"),
+            line.get("client_ip"),
+            payload.get("client_ip"),
+            payload.get("remote_addr"),
+            event.get("client_ip"),
+            event.get("remote_addr"),
+            event.get("peer"),
+            event.get("addr"),
+            event,
+        ]
+        for candidate in candidates:
+            normalized = _walk(candidate)
+            if normalized:
+                return normalized
+        return None
+
+    @staticmethod
+    def _escape_single_quotes(text: str) -> str:
+        # run_shell 侧通常会交给 shell 执行，单引号中出现单引号需要转义
+        return text.replace("'", "'\\''")
+
+    async def interrupt_playback(self, reason: str) -> None:
+        await self._ready.wait()
+        command = self._config.interrupt_command.strip()
+        if not command:
+            _LOGGER.info("xiaoai_interrupt_skipped reason=%s detail=empty_command", reason)
+            return
+        _LOGGER.info("xiaoai_interrupt_triggered reason=%s", reason)
+        try:
+            await self._open_xiaoai_server.run_shell(
+                command, float(self._config.shell_timeout_ms)
+            )
+        except Exception as exc:
+            _LOGGER.warning("xiaoai_interrupt_failed reason=%s error=%s", reason, exc)
+
+    async def speak(self, text: str) -> None:
+        if not text:
+            return
+        await self._ready.wait()
+        if self._config.tts_blocking:
+            safe_text = self._escape_single_quotes(text)
+            script = f"/usr/sbin/tts_play.sh '{safe_text}'"
+            await self._open_xiaoai_server.run_shell(
+                script, float(self._config.shell_timeout_ms)
+            )
+            return
+
+        payload = json.dumps({"text": text, "save": 0}, ensure_ascii=False)
+        payload = self._escape_single_quotes(payload)
+        script = f"ubus call mibrain text_to_speech '{payload}'"
+        await self._open_xiaoai_server.run_shell(script, float(self._config.shell_timeout_ms))
+
+    async def play_url(self, url: str) -> None:
+        if not url:
+            return
+        await self._ready.wait()
+        payload = json.dumps({"url": url, "type": 1}, ensure_ascii=False)
+        payload = self._escape_single_quotes(payload)
+        script = f"ubus call mediaplayer player_play_url '{payload}'"
+        await self._open_xiaoai_server.run_shell(script, float(self._config.shell_timeout_ms))
+
+    async def wait_forever(self) -> None:
+        await self._ready.wait()
+        await asyncio.Event().wait()
diff --git a/examples/hass/hass_assistant.py b/examples/hass/hass_assistant.py
new file mode 100644
index 0000000..97e9d61
--- /dev/null
+++ b/examples/hass/hass_assistant.py
@@ -0,0 +1,164 @@
+import asyncio
+import logging
+import os
+from pathlib import Path
+from collections.abc import Awaitable, Callable
+from typing import Optional
+
+from hass.config_loader import load_config
+from hass.errors import HomeAssistantRequestError
+from hass.ha_client import HomeAssistantClient
+from hass.memory import ConversationMemory
+from hass.typing import RecognizedSpeechInput
+from hass.xiaoai_bridge import XiaoAiBridge
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ConversationSession:
+    def __init__(
+        self,
+        *,
+        enabled: bool,
+        idle_timeout_seconds: int,
+        end_keywords: list[str],
+    ) -> None:
+        self._enabled = enabled
+        self._idle_timeout_seconds = max(1, int(idle_timeout_seconds))
+        self._end_keywords = [keyword.strip() for keyword in end_keywords if keyword and keyword.strip()]
+        self._idle_task: Optional[asyncio.Task] = None
+
+    def match_end_keyword(self, text: str) -> Optional[str]:
+        for keyword in self._end_keywords:
+            if keyword in text:
+                return keyword
+        return None
+
+    @staticmethod
+    def build_area_prefix(area: str) -> str:
+        return f"当前用户对话所在区域：{area}，如后续对话未明确指定区域，则默认为此区域"
+
+    def resolve_area(self, client_ip: Optional[str], mapping: dict[str, str]) -> Optional[str]:
+        if not client_ip:
+            return None
+        return mapping.get(client_ip)
+
+    def restart_idle_timer(self, on_timeout: Callable[[], Awaitable[None]]) -> None:
+        if not self._enabled:
+            return
+        self.cancel_idle_timer()
+
+        async def _timeout_worker() -> None:
+            try:
+                await asyncio.sleep(self._idle_timeout_seconds)
+                await on_timeout()
+            except asyncio.CancelledError:
+                return
+
+        self._idle_task = asyncio.create_task(_timeout_worker())
+
+    def cancel_idle_timer(self) -> None:
+        if self._idle_task and not self._idle_task.done():
+            self._idle_task.cancel()
+        self._idle_task = None
+
+    async def close(self) -> None:
+        self.cancel_idle_timer()
+
+
+async def run(config_path: str) -> None:
+    config = load_config(config_path)
+    logging.basicConfig(level=getattr(logging, config.runtime.log_level.upper(), logging.INFO))
+
+    memory = ConversationMemory(config.runtime.persist_state_path)
+    await memory.load()
+
+    ha = HomeAssistantClient(config.homeassistant, memory, concurrency=config.runtime.concurrency)
+    await ha.start()
+
+    bridge = XiaoAiBridge(config.xiaoai)
+    await bridge.start()
+
+    session = ConversationSession(
+        enabled=config.xiaoai.continuous_conversation_enabled,
+        idle_timeout_seconds=config.xiaoai.session_idle_timeout_seconds,
+        end_keywords=config.xiaoai.session_end_keywords,
+    )
+    session_end_lock = asyncio.Lock()
+
+    async def end_session(reason: str) -> None:
+        async with session_end_lock:
+            await ha.reset_conversation()
+            _LOGGER.info("session_ended reason=%s", reason)
+
+    async def on_idle_timeout() -> None:
+        await end_session("idle_timeout")
+
+    async def on_text(recognized: RecognizedSpeechInput) -> None:
+        text = recognized.text.strip()
+        if not text:
+            return
+
+        session.restart_idle_timer(on_idle_timeout)
+
+        matched_keyword = session.match_end_keyword(text)
+        if matched_keyword:
+            await end_session(f"keyword:{matched_keyword}")
+            await bridge.speak("好的，当前会话已结束")
+            return
+
+        area = session.resolve_area(recognized.client_ip, config.xiaoai.client_ip_area_mapping)
+        prefix_prompt = None
+        if area:
+            prefix_prompt = session.build_area_prefix(area)
+            _LOGGER.info(
+                "area_mapping_hit client_ip=%s area=%s",
+                recognized.client_ip,
+                area,
+            )
+        else:
+            _LOGGER.info("area_mapping_miss client_ip=%s", recognized.client_ip or "")
+
+        try:
+            if not config.xiaoai.continuous_conversation_enabled:
+                await end_session("continuous_disabled_new_round")
+            reply = await ha.process_text(text, prefix_prompt=prefix_prompt)
+            await bridge.speak(reply.speech)
+        except HomeAssistantRequestError as exc:
+            _LOGGER.warning(
+                "ha_process_failed text=%s client_ip=%s error=%s",
+                text,
+                recognized.client_ip or "",
+                exc,
+            )
+            if prefix_prompt:
+                try:
+                    _LOGGER.info("ha_process_retry_without_prefix client_ip=%s", recognized.client_ip)
+                    reply = await ha.process_text(text, prefix_prompt=None)
+                    await bridge.speak(reply.speech)
+                    return
+                except HomeAssistantRequestError as retry_exc:
+                    _LOGGER.warning(
+                        "ha_process_retry_failed client_ip=%s error=%s",
+                        recognized.client_ip or "",
+                        retry_exc,
+                    )
+            await bridge.speak("Home Assistant 连接异常")
+
+    bridge.set_on_text(on_text)
+    try:
+        await bridge.wait_forever()
+    finally:
+        await session.close()
+        await ha.close()
+
+
+def main() -> None:
+    config_path = os.environ.get("HASS_ASSISTANT_CONFIG", "config.json")
+    if not Path(config_path).exists():
+        config_path = "config.example.json"
+    asyncio.run(run(config_path))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/hass/pyproject.toml b/examples/hass/pyproject.toml
new file mode 100644
index 0000000..75b4086
--- /dev/null
+++ b/examples/hass/pyproject.toml
@@ -0,0 +1,19 @@
+[project]
+name = "open-xiaoai-hass"
+version = "1.0.0"
+description = "小爱音箱接入 Home Assistant Assist"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+  "aiohttp>=3.9.0",
+  "websockets>=12.0",
+  "pydantic>=2.6.0",
+  "maturin>=1.8.3"
+]
+
+[build-system]
+requires = ["maturin>=1.0,<2.0"]
+build-backend = "maturin"
+
+[tool.maturin]
+features = ["pyo3/extension-module"]
diff --git a/examples/hass/requirements.txt b/examples/hass/requirements.txt
new file mode 100644
index 0000000..c3afeab
--- /dev/null
+++ b/examples/hass/requirements.txt
@@ -0,0 +1,8 @@
+aiohttp>=3.9.0
+websockets>=12.0
+webrtcvad>=2.0.10; platform_system != "Windows"
+pydantic>=2.6.0
+
+pytest>=8.0.0
+pytest-asyncio>=0.23.0
+aioresponses>=0.7.7
diff --git a/examples/hass/scripts/__init__.py b/examples/hass/scripts/__init__.py
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/examples/hass/scripts/__init__.py
@@ -0,0 +1 @@
+
diff --git a/examples/hass/scripts/integration_device_speak.py b/examples/hass/scripts/integration_device_speak.py
new file mode 100644
index 0000000..6a12e9c
--- /dev/null
+++ b/examples/hass/scripts/integration_device_speak.py
@@ -0,0 +1,21 @@
+import asyncio
+import sys
+
+from hass.config_loader import load_config
+from hass.xiaoai_bridge import XiaoAiBridge
+
+
+async def main() -> None:
+    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
+    text = sys.argv[2] if len(sys.argv) > 2 else "已连接"
+
+    config = load_config(config_path)
+    bridge = XiaoAiBridge(config.xiaoai)
+    await bridge.start()
+    await bridge.speak(text)
+    await asyncio.sleep(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
diff --git a/examples/hass/scripts/integration_device_text.py b/examples/hass/scripts/integration_device_text.py
new file mode 100644
index 0000000..96bf030
--- /dev/null
+++ b/examples/hass/scripts/integration_device_text.py
@@ -0,0 +1,10 @@
+import asyncio
+import sys
+
+from hass_assistant import run
+
+
+if __name__ == "__main__":
+    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
+    asyncio.run(run(config_path))
+
diff --git a/examples/hass/scripts/integration_text.py b/examples/hass/scripts/integration_text.py
new file mode 100644
index 0000000..8cca19b
--- /dev/null
+++ b/examples/hass/scripts/integration_text.py
@@ -0,0 +1,30 @@
+import asyncio
+import sys
+
+from hass.config_loader import load_config
+from hass.ha_client import HomeAssistantClient
+from hass.memory import ConversationMemory
+
+
+async def main() -> None:
+    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
+    text = sys.argv[2] if len(sys.argv) > 2 else "打开客厅灯"
+
+    config = load_config(config_path)
+    memory = ConversationMemory(config.runtime.persist_state_path)
+    await memory.load()
+
+    client = HomeAssistantClient(
+        config.homeassistant, memory, concurrency=config.runtime.concurrency
+    )
+    await client.start()
+    try:
+        reply = await client.process_text(text)
+        print(reply.speech)
+    finally:
+        await client.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+
diff --git a/examples/hass/src/lib.rs b/examples/hass/src/lib.rs
new file mode 100644
index 0000000..fae9507
--- /dev/null
+++ b/examples/hass/src/lib.rs
@@ -0,0 +1,112 @@
+use open_xiaoai::services::connect::message::MessageManager;
+use open_xiaoai::services::connect::rpc::RPC;
+use open_xiaoai::services::audio::config::AudioConfig;
+use pyo3::prelude::*;
+use pyo3::types::PyBytes;
+use serde_json::json;
+use server::AppServer;
+
+pub mod macros;
+pub mod python;
+pub mod server;
+
+#[pyfunction]
+fn on_output_data(py: Python, data: Py<PyBytes>) -> PyResult<Bound<PyAny>> {
+    let bytes = data.as_bytes(py).to_vec();
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let _ = MessageManager::instance()
+            .send_stream("play", bytes, None)
+            .await;
+        Ok(())
+    })
+}
+
+#[pyfunction]
+fn start_server(py: Python) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async {
+        AppServer::run().await;
+        Ok(())
+    })
+}
+
+#[pyfunction]
+fn run_shell(py: Python, script: String, timeout_millis: f64) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let res = RPC::instance()
+            .call_remote("run_shell", Some(json!(script)), Some(timeout_millis as u64))
+            .await;
+        let result = match res {
+            Err(e) => format!("run_shell error: {}", e),
+            Ok(res) => match res.data {
+                Some(data) => serde_json::to_string(&data).unwrap_or_default(),
+                None => String::new(),
+            },
+        };
+        Ok(result)
+    })
+}
+
+#[pyfunction]
+fn start_recording(py: Python, config_json: Option<String>) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let config = config_json
+            .and_then(|text| serde_json::from_str::<AudioConfig>(&text).ok());
+        let res = RPC::instance()
+            .call_remote("start_recording", config.map(|c| json!(c)), None)
+            .await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pyfunction]
+fn stop_recording(py: Python) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let res = RPC::instance().call_remote("stop_recording", None, None).await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pyfunction]
+fn start_play(py: Python, config_json: Option<String>) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let config = config_json
+            .and_then(|text| serde_json::from_str::<AudioConfig>(&text).ok());
+        let res = RPC::instance()
+            .call_remote("start_play", config.map(|c| json!(c)), None)
+            .await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pyfunction]
+fn stop_play(py: Python) -> PyResult<Bound<PyAny>> {
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let res = RPC::instance().call_remote("stop_play", None, None).await;
+        Ok(match res {
+            Ok(_) => true,
+            Err(_) => false,
+        })
+    })
+}
+
+#[pymodule]
+fn open_xiaoai_server(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(start_server, &m)?)?;
+    m.add_function(wrap_pyfunction!(on_output_data, &m)?)?;
+    m.add_function(wrap_pyfunction!(run_shell, &m)?)?;
+    m.add_function(wrap_pyfunction!(start_recording, &m)?)?;
+    m.add_function(wrap_pyfunction!(stop_recording, &m)?)?;
+    m.add_function(wrap_pyfunction!(start_play, &m)?)?;
+    m.add_function(wrap_pyfunction!(stop_play, &m)?)?;
+    crate::python::init_module(&m)?;
+    Ok(())
+}
diff --git a/examples/hass/src/macros.rs b/examples/hass/src/macros.rs
new file mode 100644
index 0000000..9dbdd22
--- /dev/null
+++ b/examples/hass/src/macros.rs
@@ -0,0 +1,7 @@
+#[macro_export]
+macro_rules! pylog {
+    ($($arg:tt)*) => {{
+        println!($($arg)*);
+    }};
+}
+
diff --git a/examples/hass/src/python.rs b/examples/hass/src/python.rs
new file mode 100644
index 0000000..cd72a8e
--- /dev/null
+++ b/examples/hass/src/python.rs
@@ -0,0 +1,64 @@
+use pyo3::prelude::*;
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock, RwLock};
+
+pub struct PythonManager {
+    functions: Arc<RwLock<HashMap<String, PyObject>>>,
+}
+
+static INSTANCE: LazyLock<PythonManager> = LazyLock::new(PythonManager::new);
+
+impl PythonManager {
+    pub fn new() -> Self {
+        Self {
+            functions: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    fn register_fn(&self, key: &str, function: PyObject) -> PyResult<()> {
+        let mut functions = self.functions.write().unwrap();
+        functions.insert(key.to_string(), function);
+        Ok(())
+    }
+
+    fn unregister_fn(&self, key: &str) -> PyResult<()> {
+        let mut functions = self.functions.write().unwrap();
+        functions.remove(key);
+        Ok(())
+    }
+
+    pub fn call_fn(&self, key: &str, arg: Option<PyObject>) -> PyResult<PyObject> {
+        let functions = self.functions.read().unwrap();
+        if let Some(function) = functions.get(key) {
+            Python::with_gil(|py| match arg {
+                None => function.call0(py),
+                Some(arg) => function.call1(py, (arg,)),
+            })
+        } else {
+            Err(pyo3::exceptions::PyKeyError::new_err(format!(
+                "未找到函数: {}",
+                key
+            )))
+        }
+    }
+}
+
+#[pyfunction]
+fn register_fn(key: &str, function: PyObject) -> PyResult<()> {
+    PythonManager::instance().register_fn(key, function)
+}
+
+#[pyfunction]
+fn unregister_fn(key: &str) -> PyResult<()> {
+    PythonManager::instance().unregister_fn(key)
+}
+
+pub fn init_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(register_fn, m)?)?;
+    m.add_function(wrap_pyfunction!(unregister_fn, m)?)?;
+    Ok(())
+}
diff --git a/examples/hass/src/server.rs b/examples/hass/src/server.rs
new file mode 100644
index 0000000..2b5e32d
--- /dev/null
+++ b/examples/hass/src/server.rs
@@ -0,0 +1,104 @@
+use crate::python::PythonManager;
+use open_xiaoai::base::{AppError, VERSION};
+use open_xiaoai::services::connect::data::{Event, Request, Response, Stream};
+use open_xiaoai::services::connect::handler::MessageHandler;
+use open_xiaoai::services::connect::message::{MessageManager, WsStream};
+use open_xiaoai::services::connect::rpc::RPC;
+use pyo3::types::{PyBytes, PyString};
+use pyo3::Python;
+use serde_json::json;
+use std::sync::{LazyLock, RwLock};
+use tokio::net::{TcpListener, TcpStream};
+use tokio_tungstenite::accept_async;
+
+static REMOTE_ADDR: LazyLock<RwLock<Option<String>>> = LazyLock::new(|| RwLock::new(None));
+
+pub struct AppServer;
+
+impl AppServer {
+    pub async fn connect(stream: TcpStream) -> Result<WsStream, AppError> {
+        let ws_stream = accept_async(stream).await?;
+        Ok(WsStream::Server(ws_stream))
+    }
+
+    pub async fn run() {
+        let addr = "0.0.0.0:4399";
+        let listener = TcpListener::bind(&addr)
+            .await
+            .expect(format!("❌ 绑定地址失败: {}", &addr).as_str());
+        crate::pylog!("✅ 已启动: {:?}", addr);
+        while let Ok((stream, addr)) = listener.accept().await {
+            AppServer::handle_connection(stream, addr).await;
+        }
+    }
+
+    async fn handle_connection(stream: TcpStream, addr: std::net::SocketAddr) {
+        let Ok(ws_stream) = AppServer::connect(stream).await else {
+            crate::pylog!("❌ 连接异常: {}", addr);
+            return;
+        };
+        crate::pylog!("✅ 已连接: {:?}", addr);
+        {
+            let mut remote = REMOTE_ADDR.write().unwrap();
+            *remote = Some(addr.ip().to_string());
+        }
+        AppServer::init(ws_stream).await;
+        if let Err(e) = MessageManager::instance().process_messages().await {
+            crate::pylog!("❌ 消息处理异常: {}", e);
+        }
+        AppServer::dispose().await;
+        {
+            let mut remote = REMOTE_ADDR.write().unwrap();
+            *remote = None;
+        }
+        crate::pylog!("❌ 已断开连接");
+    }
+
+    async fn init(ws_stream: WsStream) {
+        MessageManager::instance().init(ws_stream).await;
+        MessageHandler::<Event>::instance()
+            .set_handler(on_event)
+            .await;
+        MessageHandler::<Stream>::instance()
+            .set_handler(on_stream)
+            .await;
+
+        let rpc = RPC::instance();
+        rpc.add_command("get_version", get_version).await;
+    }
+
+    async fn dispose() {
+        MessageManager::instance().dispose().await;
+    }
+}
+
+async fn get_version(_: Request) -> Result<Response, AppError> {
+    let data = json!(VERSION.to_string());
+    Ok(Response::from_data(data))
+}
+
+async fn on_stream(stream: Stream) -> Result<(), AppError> {
+    let Stream { tag, bytes, .. } = stream;
+    match tag.as_str() {
+        "record" => {
+            let data = Python::with_gil(|py| PyBytes::new(py, &bytes).into());
+            PythonManager::instance().call_fn("on_input_data", Some(data))?;
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
+async fn on_event(event: Event) -> Result<(), AppError> {
+    let mut event_json_value = serde_json::to_value(&event)?;
+    if let Some(event_obj) = event_json_value.as_object_mut() {
+        let remote = REMOTE_ADDR.read().unwrap().clone();
+        if let Some(ip) = remote {
+            event_obj.insert("remote_addr".to_string(), serde_json::Value::String(ip));
+        }
+    }
+    let event_json = serde_json::to_string(&event_json_value)?;
+    let data = Python::with_gil(|py| PyString::new(py, &event_json).into());
+    PythonManager::instance().call_fn("on_event", Some(data))?;
+    Ok(())
+}
diff --git a/examples/hass/tests/__init__.py b/examples/hass/tests/__init__.py
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/examples/hass/tests/__init__.py
@@ -0,0 +1 @@
+
diff --git a/examples/hass/tests/conftest.py b/examples/hass/tests/conftest.py
new file mode 100644
index 0000000..6ee195b
--- /dev/null
+++ b/examples/hass/tests/conftest.py
@@ -0,0 +1,5 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
diff --git a/examples/hass/tests/test_assistant_session.py b/examples/hass/tests/test_assistant_session.py
new file mode 100644
index 0000000..27d22b3
--- /dev/null
+++ b/examples/hass/tests/test_assistant_session.py
@@ -0,0 +1,53 @@
+import asyncio
+
+import pytest
+
+from hass_assistant import ConversationSession
+
+
+@pytest.mark.asyncio
+async def test_session_idle_timeout_triggers_end():
+    session = ConversationSession(
+        enabled=True,
+        idle_timeout_seconds=1,
+        end_keywords=["再见", "拜拜", "滚", "退下"],
+    )
+
+    triggered = {"value": False}
+
+    async def on_timeout():
+        triggered["value"] = True
+
+    session.restart_idle_timer(on_timeout)
+    await asyncio.sleep(1.2)
+
+    assert triggered["value"] is True
+    await session.close()
+
+
+def test_session_end_keyword_contains_match():
+    session = ConversationSession(
+        enabled=True,
+        idle_timeout_seconds=15,
+        end_keywords=["再见", "拜拜", "滚", "退下"],
+    )
+
+    assert session.match_end_keyword("那就先这样吧，再见哈") == "再见"
+    assert session.match_end_keyword("你先退下吧") == "退下"
+    assert session.match_end_keyword("继续聊聊") is None
+
+
+def test_session_area_mapping_hit_and_miss():
+    session = ConversationSession(
+        enabled=True,
+        idle_timeout_seconds=15,
+        end_keywords=["再见"],
+    )
+
+    mapping = {"192.168.1.101": "客厅"}
+    assert session.resolve_area("192.168.1.101", mapping) == "客厅"
+    assert session.resolve_area("192.168.1.102", mapping) is None
+    assert session.resolve_area(None, mapping) is None
+
+    prefix = session.build_area_prefix("客厅")
+    assert "当前用户对话所在区域：客厅" in prefix
diff --git a/examples/hass/tests/test_ha_client.py b/examples/hass/tests/test_ha_client.py
new file mode 100644
index 0000000..ae8aa5b
--- /dev/null
+++ b/examples/hass/tests/test_ha_client.py
@@ -0,0 +1,123 @@
+import json
+
+import pytest
+from aioresponses import aioresponses
+
+from hass.errors import HomeAssistantAuthError
+from hass.ha_client import HomeAssistantClient
+from hass.memory import ConversationMemory
+from hass.typing import HomeAssistantConfig
+
+
+@pytest.mark.asyncio
+async def test_process_text_updates_conversation_id(tmp_path):
+    state_path = tmp_path / "state.json"
+    memory = ConversationMemory(str(state_path))
+    await memory.load()
+
+    config = HomeAssistantConfig(
+        url="http://ha.local:8123",
+        access_token="token",
+        assistant_entity_id="home_assistant",
+        language="zh-cn",
+        timeout_seconds=2.5,
+    )
+    client = HomeAssistantClient(config, memory, concurrency=2)
+    await client.start()
+
+    url = "http://ha.local:8123/api/conversation/process"
+    with aioresponses() as mocked:
+        mocked.post(
+            url,
+            payload={
+                "continue_conversation": True,
+                "conversation_id": "cid-1",
+                "response": {
+                    "response_type": "action_done",
+                    "speech": {"plain": {"speech": "好的"}},
+                },
+            },
+            status=200,
+        )
+        reply = await client.process_text("打开客厅灯")
+
+    assert reply.speech == "好的"
+    assert reply.continue_conversation is True
+    assert memory.conversation_id == "cid-1"
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_process_text_unauthorized(tmp_path):
+    memory = ConversationMemory(str(tmp_path / "state.json"))
+    await memory.load()
+
+    config = HomeAssistantConfig(url="http://ha.local:8123", access_token="bad-token")
+    client = HomeAssistantClient(config, memory)
+    await client.start()
+
+    url = "http://ha.local:8123/api/conversation/process"
+    with aioresponses() as mocked:
+        mocked.post(url, payload={"message": "unauthorized"}, status=401)
+        with pytest.raises(HomeAssistantAuthError):
+            await client.process_text("hi")
+
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_process_text_with_prefix_prompt(tmp_path):
+    memory = ConversationMemory(str(tmp_path / "state.json"))
+    await memory.load()
+
+    config = HomeAssistantConfig(url="http://ha.local:8123", access_token="token")
+    client = HomeAssistantClient(config, memory)
+    await client.start()
+
+    url = "http://ha.local:8123/api/conversation/process"
+    with aioresponses() as mocked:
+        mocked.post(
+            url,
+            payload={
+                "continue_conversation": False,
+                "conversation_id": "cid-2",
+                "response": {
+                    "response_type": "query_answer",
+                    "speech": {"plain": {"speech": "在客厅"}},
+                },
+            },
+            status=200,
+        )
+        await client.process_text(
+            "灯开了吗",
+            prefix_prompt="当前用户对话所在区域：客厅，如后续对话未明确指定区域，则默认为此区域",
+        )
+
+        request = mocked.requests[("POST", url)][0]
+        payload = json.loads(request.kwargs["data"].decode("utf-8"))
+        assert payload["text"].startswith("当前用户对话所在区域：客厅")
+        assert "用户请求：灯开了吗" in payload["text"]
+
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_reset_conversation(tmp_path):
+    state_path = tmp_path / "state.json"
+    memory = ConversationMemory(str(state_path))
+    await memory.load()
+    memory.conversation_id = "cid-old"
+    await memory.save()
+
+    config = HomeAssistantConfig(url="http://ha.local:8123", access_token="token")
+    client = HomeAssistantClient(config, memory)
+    await client.start()
+
+    await client.reset_conversation()
+
+    assert memory.conversation_id is None
+    saved = json.loads(state_path.read_text(encoding="utf-8"))
+    assert saved["conversation_id"] is None
+
+    await client.close()
+
diff --git a/examples/hass/tests/test_memory.py b/examples/hass/tests/test_memory.py
new file mode 100644
index 0000000..0e7f9d0
--- /dev/null
+++ b/examples/hass/tests/test_memory.py
@@ -0,0 +1,21 @@
+import json
+
+import pytest
+
+from hass.memory import ConversationMemory
+
+
+@pytest.mark.asyncio
+async def test_memory_persist_roundtrip(tmp_path):
+    path = tmp_path / "state.json"
+    m = ConversationMemory(str(path))
+    m.conversation_id = "abc"
+    await m.save()
+
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    assert raw["conversation_id"] == "abc"
+
+    m2 = ConversationMemory(str(path))
+    await m2.load()
+    assert m2.conversation_id == "abc"
+
diff --git a/examples/hass/tests/test_typing.py b/examples/hass/tests/test_typing.py
new file mode 100644
index 0000000..dcd2e0f
--- /dev/null
+++ b/examples/hass/tests/test_typing.py
@@ -0,0 +1,38 @@
+from hass.typing import AppConfig
+
+
+def test_xiaoai_config_defaults():
+    cfg = AppConfig.model_validate(
+        {
+            "homeassistant": {
+                "url": "http://ha.local:8123",
+                "access_token": "token",
+            }
+        }
+    )
+
+    assert cfg.xiaoai.continuous_conversation_enabled is True
+    assert cfg.xiaoai.session_idle_timeout_seconds == 15
+    assert cfg.xiaoai.session_end_keywords == ["再见", "拜拜", "滚", "退下"]
+    assert cfg.xiaoai.client_ip_area_mapping == {}
+    assert cfg.xiaoai.interrupt_on_new_input is True
+
+
+def test_xiaoai_config_area_mapping_loaded():
+    cfg = AppConfig.model_validate(
+        {
+            "homeassistant": {
+                "url": "http://ha.local:8123",
+                "access_token": "token",
+            },
+            "xiaoai": {
+                "client_ip_area_mapping": {
+                    "192.168.1.10": "客厅",
+                    "192.168.1.11": "书房",
+                }
+            },
+        }
+    )
+
+    assert cfg.xiaoai.client_ip_area_mapping["192.168.1.10"] == "客厅"
+    assert cfg.xiaoai.client_ip_area_mapping["192.168.1.11"] == "书房"
diff --git a/examples/hass/tests/test_vad.py b/examples/hass/tests/test_vad.py
new file mode 100644
index 0000000..262e298
--- /dev/null
+++ b/examples/hass/tests/test_vad.py
@@ -0,0 +1,41 @@
+import math
+import struct
+
+import pytest
+
+from hass.vad import VadConfig, VadSegmenter
+
+
+def _sine_frame(sample_rate: int, freq_hz: float, frame_ms: int = 20) -> bytes:
+    samples = int(sample_rate * frame_ms / 1000)
+    values = [
+        int(10000 * math.sin(2 * math.pi * freq_hz * i / sample_rate))
+        for i in range(samples)
+    ]
+    return struct.pack("<" + "h" * samples, *values)
+
+
+def _silence_frame(sample_rate: int, frame_ms: int = 20) -> bytes:
+    samples = int(sample_rate * frame_ms / 1000)
+    return b"\x00\x00" * samples
+
+
+@pytest.mark.parametrize("min_silence_ms", [40, 60])
+def test_vad_emits_utterance(min_silence_ms: int):
+    config = VadConfig(
+        aggressiveness=1,
+        min_silence_ms=min_silence_ms,
+        max_utterance_ms=2000,
+        sample_rate=16000,
+    )
+    seg = VadSegmenter(config)
+
+    utterances = []
+    for _ in range(6):
+        utterances.extend(seg.feed(_sine_frame(16000, 440.0)))
+    for _ in range(6):
+        utterances.extend(seg.feed(_silence_frame(16000)))
+
+    assert len(utterances) >= 1
+    assert len(utterances[0]) > 0
+
diff --git a/examples/hass/tests/test_xiaoai_bridge.py b/examples/hass/tests/test_xiaoai_bridge.py
new file mode 100644
index 0000000..1607ac7
--- /dev/null
+++ b/examples/hass/tests/test_xiaoai_bridge.py
@@ -0,0 +1,113 @@
+import asyncio
+import json
+
+import pytest
+
+from hass.typing import XiaoAiConfig
+from hass.xiaoai_bridge import XiaoAiBridge
+
+
+class FakeServer:
+    def __init__(self):
+        self._fns = {}
+        self.shell_calls = []
+
+    def register_fn(self, name, fn):
+        self._fns[name] = fn
+
+    async def run_shell(self, script, timeout):
+        self.shell_calls.append((script, timeout))
+        return '{"stdout":"ok","stderr":"","exit_code":0}'
+
+    async def start_server(self):
+        return None
+
+
+@pytest.mark.asyncio
+async def test_bridge_interrupts_before_dispatch(monkeypatch):
+    fake_server = FakeServer()
+
+    import hass.xiaoai_bridge as module
+
+    monkeypatch.setattr(module, "_LOGGER", module._LOGGER)
+
+    cfg = XiaoAiConfig(interrupt_on_new_input=True, shell_timeout_ms=123)
+    bridge = XiaoAiBridge(cfg)
+    bridge._open_xiaoai_server = fake_server
+    bridge._loop = asyncio.get_running_loop()
+
+    called = []
+
+    async def on_text(input_data):
+        called.append(input_data)
+
+    bridge.set_on_text(on_text)
+    bridge._ready.set()
+    bridge._register_callbacks()
+
+    event = {
+        "event": "instruction",
+        "data": {
+            "ClientIP": "192.168.1.101",
+            "NewLine": json.dumps(
+                {
+                    "header": {"namespace": "SpeechRecognizer", "name": "RecognizeResult"},
+                    "payload": {
+                        "is_final": True,
+                        "results": [{"text": "打开客厅灯"}],
+                    },
+                },
+                ensure_ascii=False,
+            ),
+        },
+    }
+
+    fake_server._fns["on_event"](json.dumps(event, ensure_ascii=False))
+    await asyncio.sleep(0.05)
+
+    assert len(fake_server.shell_calls) == 1
+    assert "ubus call mibrain ai_service" in fake_server.shell_calls[0][0]
+    assert len(called) == 1
+    assert called[0].text == "打开客厅灯"
+    assert called[0].client_ip == "192.168.1.101"
+
+
+@pytest.mark.asyncio
+async def test_bridge_extracts_missing_ip_as_none():
+    fake_server = FakeServer()
+    cfg = XiaoAiConfig(interrupt_on_new_input=False)
+    bridge = XiaoAiBridge(cfg)
+    bridge._open_xiaoai_server = fake_server
+    bridge._loop = asyncio.get_running_loop()
+
+    captured = []
+
+    async def on_text(input_data):
+        captured.append(input_data)
+
+    bridge.set_on_text(on_text)
+    bridge._ready.set()
+    bridge._register_callbacks()
+
+    event = {
+        "event": "instruction",
+        "data": {
+            "NewLine": json.dumps(
+                {
+                    "header": {"namespace": "SpeechRecognizer", "name": "RecognizeResult"},
+                    "payload": {
+                        "is_final": True,
+                        "results": [{"text": "查询天气"}],
+                    },
+                },
+                ensure_ascii=False,
+            ),
+        },
+    }
+
+    fake_server._fns["on_event"](json.dumps(event, ensure_ascii=False))
+    await asyncio.sleep(0.05)
+
+    assert len(captured) == 1
+    assert captured[0].client_ip is None
+    assert fake_server.shell_calls == []