最近在智能家居开发领域，Alexa Plus 的更新引起了广泛关注。这次更新不仅增强了设备连接能力，还引入了备受期待的 MCP 开放标准支持。对于从事智能家居开发的工程师来说，这意味着更统一的开发体验和更强大的功能扩展能力。本文将深入解析 Alexa Plus 的最新特性，重点介绍 MCP 标准在智能家居领域的应用实践。

无论你是刚开始接触 Alexa 技能开发，还是已经有一定经验的开发者，本文都将为你提供从基础概念到实战开发的完整指导。通过阅读本文，你将掌握 Alexa Plus 的新特性、MCP 标准的核心原理，以及如何利用这些技术构建更智能的家居应用。

1. Alexa Plus 与智能家居开发现状

1.1 Alexa Plus 平台概述

Alexa Plus 是亚马逊推出的增强型语音助手平台，相比标准版 Alexa，它在设备连接、响应速度和功能扩展方面都有显著提升。最新版本的 Alexa Plus 重点优化了智能家居设备的兼容性，支持更多类型的物联网设备接入。

从技术架构来看，Alexa Plus 采用微服务架构，通过 Alexa Voice Service (AVS) 提供统一的语音交互接口。开发者可以通过 Alexa Skills Kit (ASK) 创建自定义技能，实现设备控制、场景联动等复杂功能。平台支持多种通信协议，包括 Wi-Fi、蓝牙、Zigbee 和 Z-Wave，为不同类型的智能设备提供灵活的连接方案。

1.2 智能家居设备连接的技术挑战

在实际开发中，智能家居设备连接面临诸多技术挑战。首先是协议碎片化问题，不同厂商的设备可能使用不同的通信协议和数据结构，导致集成复杂度高。其次是设备发现和配网难题，特别是对于大规模设备部署场景，如何实现快速、稳定的设备连接是关键挑战。

此外，设备状态同步也是常见问题。当多个客户端同时控制同一设备时，需要确保状态的一致性。Alexa Plus 通过改进的设备状态管理机制，提供了更可靠的状态同步方案，减少了状态冲突的概率。

2. MCP 开放标准深度解析

2.1 MCP 协议的基本概念

MCP（Model Context Protocol）是一种新兴的开放标准协议，旨在为 AI 应用和外部工具之间提供统一的交互接口。在智能家居场景中，MCP 充当了 Alexa 语音助手与智能设备之间的桥梁，实现了更规范的通信机制。

MCP 协议的核心优势在于其标准化程度高。它定义了清晰的请求-响应模式，支持多种数据格式的传输。协议采用 JSON-RPC 2.0 规范，确保了跨平台兼容性。对于开发者而言，这意味着可以专注于业务逻辑实现，而不必担心底层通信细节。

2.2 MCP 在智能家居中的应用价值

MCP 标准为智能家居开发带来了多重价值。首先，它简化了设备集成流程。传统上，每接入一种新类型的设备，都需要编写特定的适配器代码。而采用 MCP 标准后，只需要实现标准的 MCP 服务器接口，就能快速完成设备对接。

其次，MCP 提升了系统的可扩展性。当需要新增功能时，可以通过扩展 MCP 工具集来实现，而不需要修改核心架构。这种设计使得智能家居系统能够灵活适应未来的需求变化。

3. 开发环境准备与工具配置

3.1 基础环境要求

在开始 Alexa Plus 开发之前，需要准备相应的开发环境。推荐使用以下配置：

• 操作系统 ：Windows 10/11、macOS 12+ 或 Ubuntu 20.04+
• Node.js ：版本 18.x 或更高（Alexa Skills Kit 依赖）
• Python ：版本 3.8+（用于 MCP 服务器开发）
• 开发工具 ：VS Code 或 WebStorm，安装相应的 Alexa 开发插件

3.2 Alexa 开发者账户配置

首先需要注册 Alexa 开发者账户并完成基础配置：

1. 访问 Alexa 开发者控制台（developer.amazon.com/alexa）
2. 使用亚马逊账户登录或注册新账户
3. 完成开发者信息验证，包括邮箱确认和手机验证
4. 在账户设置中启用技能测试权限

3.3 本地开发环境搭建

安装必要的开发依赖和工具链：

# 安装 Alexa Skills Kit CLI
npm install -g ask-cli

# 配置 ASK CLI
ask configure

# 安装 MCP 相关工具
pip install mcp-client mcp-server

创建项目目录结构：

alexa-plus-project/
├── skill-package/
│   ├── interactionModels/
│   ├── skill.json
│   └── assets/
├── lambda/
│   ├── package.json
│   ├── index.js
│   └── node_modules/
└── mcp-server/
    ├── requirements.txt
    ├── server.py
    └── devices/

4. Alexa Plus 技能开发实战

4.1 创建基础语音技能

首先创建一个简单的灯光控制技能，演示 Alexa Plus 的基本开发流程：

// lambda/index.js
const Alexa = require('ask-sdk-core');

const LaunchRequestHandler = {
    canHandle(handlerInput) {
        return Alexa.getRequestType(handlerInput.requestEnvelope) === 'LaunchRequest';
    },
    handle(handlerInput) {
        const speakOutput = '欢迎使用智能家居控制，您可以说打开客厅灯光';
        return handlerInput.responseBuilder
            .speak(speakOutput)
            .reprompt(speakOutput)
            .getResponse();
    }
};

const LightControlIntentHandler = {
    canHandle(handlerInput) {
        return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
            && Alexa.getIntentName(handlerInput.requestEnvelope) === 'LightControlIntent';
    },
    async handle(handlerInput) {
        const slotValue = Alexa.getSlotValue(handlerInput.requestEnvelope, 'action');
        const deviceName = Alexa.getSlotValue(handlerInput.requestEnvelope, 'device');
        
        // 调用设备控制逻辑
        const result = await controlDevice(deviceName, slotValue);
        
        const speakOutput = result.success ? 
            `已${slotValue}${deviceName}的灯光` : 
            `操作失败，请检查设备状态`;
            
        return handlerInput.responseBuilder
            .speak(speakOutput)
            .getResponse();
    }
};

async function controlDevice(deviceName, action) {
    // 设备控制逻辑实现
    // 这里后续会集成 MCP 调用
    return { success: true, message: '操作成功' };
}

exports.handler = Alexa.SkillBuilders.custom()
    .addRequestHandlers(
        LaunchRequestHandler,
        LightControlIntentHandler
    )
    .lambda();

4.2 配置技能交互模型

创建语音交互模型定义文件：

{
  "interactionModel": {
    "languageModel": {
      "invocationName": "智能家居",
      "intents": [
        {
          "name": "LightControlIntent",
          "slots": [
            {
              "name": "action",
              "type": "ACTION_TYPE"
            },
            {
              "name": "device",
              "type": "DEVICE_NAME"
            }
          ],
          "samples": [
            "打开{device}灯光",
            "关闭{device}灯光",
            "{action}{device}的灯"
          ]
        }
      ],
      "types": [
        {
          "name": "ACTION_TYPE",
          "values": [
            { "name": { "value": "打开" } },
            { "name": { "value": "关闭" } },
            { "name": { "value": "调节" } }
          ]
        },
        {
          "name": "DEVICE_NAME",
          "values": [
            { "name": { "value": "客厅" } },
            { "name": { "value": "卧室" } },
            { "name": { "value": "厨房" } }
          ]
        }
      ]
    }
  }
}

5. MCP 服务器集成实战

5.1 创建基础 MCP 服务器

接下来实现一个支持设备控制的 MCP 服务器：

# mcp-server/server.py
import asyncio
from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
from mcp.types import Tool, TextContent, EmbeddedResource

class SmartHomeMCPServer:
    def __init__(self):
        self.server = Server("smart-home-mcp")
        self.devices = {
            "living_room_light": {"name": "客厅灯光", "state": "off"},
            "bedroom_light": {"name": "卧室灯光", "state": "off"}
        }
        
        # 注册可用工具
        self.server.list_tools()(self.list_tools)
        self.server.call_tool()(self.call_tool)

    async def list_tools(self) -> list[Tool]:
        """返回可用的设备控制工具"""
        return [
            Tool(
                name="control_light",
                description="控制智能灯光设备",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "device": {
                            "type": "string",
                            "enum": list(self.devices.keys()),
                            "description": "设备标识符"
                        },
                        "action": {
                            "type": "string", 
                            "enum": ["on", "off", "toggle"],
                            "description": "执行的操作"
                        }
                    },
                    "required": ["device", "action"]
                }
            ),
            Tool(
                name="get_device_status",
                description="获取设备状态",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "device": {
                            "type": "string",
                            "enum": list(self.devices.keys()),
                            "description": "设备标识符"
                        }
                    },
                    "required": ["device"]
                }
            )
        ]

    async def call_tool(self, name: str, arguments: dict) -> list[TextContent]:
        """执行工具调用"""
        if name == "control_light":
            device = arguments["device"]
            action = arguments["action"]
            return await self.control_light(device, action)
        elif name == "get_device_status":
            device = arguments["device"]
            return await self.get_device_status(device)
        else:
            raise ValueError(f"未知工具: {name}")

    async def control_light(self, device: str, action: str) -> list[TextContent]:
        """控制灯光设备"""
        if device not in self.devices:
            return [TextContent(type="text", text=f"设备 {device} 不存在")]
        
        if action == "on":
            self.devices[device]["state"] = "on"
            message = f"已打开 {self.devices[device]['name']}"
        elif action == "off":
            self.devices[device]["state"] = "off" 
            message = f"已关闭 {self.devices[device]['name']}"
        elif action == "toggle":
            current_state = self.devices[device]["state"]
            new_state = "off" if current_state == "on" else "on"
            self.devices[device]["state"] = new_state
            message = f"已切换 {self.devices[device]['name']} 状态"
        
        return [TextContent(type="text", text=message)]

    async def get_device_status(self, device: str) -> list[TextContent]:
        """获取设备状态"""
        if device not in self.devices:
            return [TextContent(type="text", text=f"设备 {device} 不存在")]
        
        status = self.devices[device]["state"]
        return [TextContent(type="text", text=f"{self.devices[device]['name']} 状态: {status}")]

async def main():
    server = SmartHomeMCPServer()
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.server.run(
            read_stream, 
            write_stream,
            InitializationOptions(
                server_name="smart-home-mcp",
                server_version="1.0.0",
                capabilities=server.server.get_capabilities(
                    notification_options=None,
                    experimental_capabilities=None
                )
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

5.2 集成 MCP 客户端到 Alexa 技能

修改之前的 Alexa 技能代码，集成 MCP 客户端：

// lambda/mcp-client.js
const { Client } = require('mcp-client');

class SmartHomeMCPClient {
    constructor() {
        this.client = null;
        this.isConnected = false;
    }

    async connect() {
        try {
            this.client = new Client({
                server: process.env.MCP_SERVER_URL || 'ws://localhost:8000'
            });
            
            await this.client.connect();
            this.isConnected = true;
            console.log('MCP 服务器连接成功');
        } catch (error) {
            console.error('MCP 连接失败:', error);
            this.isConnected = false;
        }
    }

    async controlDevice(deviceName, action) {
        if (!this.isConnected) {
            await this.connect();
        }

        try {
            // 映射设备名称到 MCP 设备标识符
            const deviceMap = {
                '客厅': 'living_room_light',
                '卧室': 'bedroom_light',
                '厨房': 'kitchen_light'
            };

            const deviceId = deviceMap[deviceName];
            if (!deviceId) {
                return { success: false, message: '不支持的设备' };
            }

            // 调用 MCP 工具
            const result = await this.client.callTool('control_light', {
                device: deviceId,
                action: action === '打开' ? 'on' : 'off'
            });

            return { success: true, message: result.content[0].text };
        } catch (error) {
            console.error('设备控制失败:', error);
            return { success: false, message: '设备控制异常' };
        }
    }

    async getDeviceStatus(deviceName) {
        if (!this.isConnected) {
            await this.connect();
        }

        try {
            const deviceMap = {
                '客厅': 'living_room_light',
                '卧室': 'bedroom_light'
            };

            const deviceId = deviceMap[deviceName];
            const result = await this.client.callTool('get_device_status', {
                device: deviceId
            });

            return { success: true, status: result.content[0].text };
        } catch (error) {
            return { success: false, status: '未知' };
        }
    }
}

module.exports = SmartHomeMCPClient;

更新主要的技能处理器：

// lambda/index.js (更新版本)
const Alexa = require('ask-sdk-core');
const SmartHomeMCPClient = require('./mcp-client');

const mcpClient = new SmartHomeMCPClient();

const LightControlIntentHandler = {
    canHandle(handlerInput) {
        return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
            && Alexa.getIntentName(handlerInput.requestEnvelope) === 'LightControlIntent';
    },
    async handle(handlerInput) {
        const slotValue = Alexa.getSlotValue(handlerInput.requestEnvelope, 'action');
        const deviceName = Alexa.getSlotValue(handlerInput.requestEnvelope, 'device');
        
        // 使用 MCP 客户端控制设备
        const result = await mcpClient.controlDevice(deviceName, slotValue);
        
        const speakOutput = result.success ? 
            result.message : 
            `操作失败，请检查设备状态`;
            
        return handlerInput.responseBuilder
            .speak(speakOutput)
            .getResponse();
    }
};

6. 设备发现与连接管理

6.1 实现设备自动发现机制

Alexa Plus 增强了设备发现能力，我们可以利用这一特性实现智能设备的自动注册：

# mcp-server/device_discovery.py
import asyncio
import json
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class DeviceInfo:
    id: str
    name: str
    type: str
    capabilities: List[str]
    manufacturer: str
    model: str

class DeviceDiscoveryService:
    def __init__(self):
        self.discovered_devices: Dict[str, DeviceInfo] = {}
        self.discovery_handlers = []
    
    async def start_discovery(self):
        """启动设备发现流程"""
        # 模拟设备发现过程
        simulated_devices = [
            DeviceInfo(
                id="light_001",
                name="客厅智能灯",
                type="light",
                capabilities=["on_off", "brightness"],
                manufacturer="Philips",
                model="Hue White"
            ),
            DeviceInfo(
                id="thermostat_001", 
                name="卧室温控器",
                type="thermostat",
                capabilities=["temperature_control", "mode_setting"],
                manufacturer="Nest",
                model="Learning Thermostat"
            )
        ]
        
        for device in simulated_devices:
            await self.register_device(device)
    
    async def register_device(self, device: DeviceInfo):
        """注册新发现的设备"""
        self.discovered_devices[device.id] = device
        print(f"发现新设备: {device.name} ({device.type})")
        
        # 通知所有注册的处理器
        for handler in self.discovery_handlers:
            await handler(device)
    
    def add_discovery_handler(self, handler):
        """添加设备发现处理器"""
        self.discovery_handlers.append(handler)
    
    async def get_available_devices(self) -> List[DeviceInfo]:
        """获取所有可用设备"""
        return list(self.discovered_devices.values())

6.2 设备状态同步与冲突解决

在多人同时控制设备的场景下，状态同步至关重要：

# mcp-server/state_manager.py
import asyncio
from datetime import datetime
from typing import Dict, Optional
import threading

class DeviceStateManager:
    def __init__(self):
        self.device_states: Dict[str, Dict] = {}
        self.state_lock = threading.Lock()
        self.state_history: Dict[str, List] = {}
    
    async def update_device_state(self, device_id: str, new_state: Dict, source: str = "unknown"):
        """更新设备状态（带冲突检测）"""
        with self.state_lock:
            current_state = self.device_states.get(device_id, {})
            
            # 检查状态冲突
            if self._has_state_conflict(device_id, current_state, new_state):
                print(f"状态冲突检测: 设备 {device_id} 存在并发更新")
                # 解决冲突：采用时间戳最新的状态
                if new_state.get('timestamp', 0) > current_state.get('timestamp', 0):
                    await self._apply_state_update(device_id, new_state, source)
                else:
                    print("忽略旧的状态更新")
            else:
                await self._apply_state_update(device_id, new_state, source)
    
    def _has_state_conflict(self, device_id: str, current: Dict, new: Dict) -> bool:
        """检测状态冲突"""
        if device_id not in self.device_states:
            return False
        
        # 检查关键状态字段是否冲突
        conflict_fields = ['power', 'mode', 'temperature']
        for field in conflict_fields:
            if field in current and field in new and current[field] != new[field]:
                return True
        return False
    
    async def _apply_state_update(self, device_id: str, new_state: Dict, source: str):
        """应用状态更新"""
        new_state['last_updated'] = datetime.now().isoformat()
        new_state['last_source'] = source
        
        self.device_states[device_id] = new_state
        
        # 记录状态历史
        if device_id not in self.state_history:
            self.state_history[device_id] = []
        self.state_history[device_id].append({
            'timestamp': datetime.now(),
            'state': new_state.copy(),
            'source': source
        })
        
        # 保持历史记录数量
        if len(self.state_history[device_id]) > 100:
            self.state_history[device_id] = self.state_history[device_id][-50:]
    
    async def get_device_state(self, device_id: str) -> Optional[Dict]:
        """获取设备当前状态"""
        return self.device_states.get(device_id)
    
    async def get_state_history(self, device_id: str, limit: int = 10) -> List[Dict]:
        """获取设备状态历史"""
        history = self.state_history.get(device_id, [])
        return history[-limit:]

7. 高级功能与场景联动

7.1 实现智能场景配置

利用 Alexa Plus 的场景管理功能，实现复杂的设备联动：

// lambda/scene-manager.js
class SceneManager {
    constructor() {
        this.scenes = new Map();
        this.loadDefaultScenes();
    }

    loadDefaultScenes() {
        // 早安场景
        this.scenes.set('morning', {
            name: '早安模式',
            description: '清晨起床场景',
            actions: [
                { device: 'living_room_light', action: 'on', brightness: 50 },
                { device: 'bedroom_light', action: 'on', brightness: 30 },
                { device: 'coffee_maker', action: 'on' }
            ],
            triggers: ['语音触发', '定时触发']
        });

        // 晚安场景
        this.scenes.set('goodnight', {
            name: '晚安模式', 
            description: '睡前准备场景',
            actions: [
                { device: 'living_room_light', action: 'off' },
                { device: 'bedroom_light', action: 'on', brightness: 10 },
                { device: 'thermostat', action: 'set_temperature', value: 22 }
            ],
            triggers: ['语音触发', '运动传感器']
        });
    }

    async executeScene(sceneId) {
        const scene = this.scenes.get(sceneId);
        if (!scene) {
            throw new Error(`场景 ${sceneId} 不存在`);
        }

        const results = [];
        for (const action of scene.actions) {
            try {
                const result = await this.executeDeviceAction(action);
                results.push({
                    device: action.device,
                    success: result.success,
                    message: result.message
                });
            } catch (error) {
                results.push({
                    device: action.device,
                    success: false,
                    message: error.message
                });
            }
        }

        return {
            scene: scene.name,
            results: results,
            timestamp: new Date().toISOString()
        };
    }

    async executeDeviceAction(action) {
        // 调用 MCP 客户端执行具体设备操作
        // 这里简化实现，实际需要根据设备类型调用不同的 MCP 工具
        return { success: true, message: '操作完成' };
    }

    createCustomScene(name, description, actions) {
        const sceneId = this.generateSceneId(name);
        this.scenes.set(sceneId, {
            name,
            description,
            actions,
            triggers: ['语音触发']
        });
        return sceneId;
    }

    generateSceneId(name) {
        return name.toLowerCase().replace(/\s+/g, '_');
    }
}

module.exports = SceneManager;

7.2 语音技能的场景集成

在 Alexa 技能中集成场景控制功能：

// lambda/scene-handlers.js
const SceneManager = require('./scene-manager');

const sceneManager = new SceneManager();

const SceneControlIntentHandler = {
    canHandle(handlerInput) {
        return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
            && Alexa.getIntentName(handlerInput.requestEnvelope) === 'SceneControlIntent';
    },
    async handle(handlerInput) {
        const sceneName = Alexa.getSlotValue(handlerInput.requestEnvelope, 'scene');
        
        try {
            const sceneId = sceneManager.generateSceneId(sceneName);
            const result = await sceneManager.executeScene(sceneId);
            
            const speakOutput = `已执行${sceneName}场景，共完成${result.results.filter(r => r.success).length}个设备操作`;
            
            return handlerInput.responseBuilder
                .speak(speakOutput)
                .withSimpleCard(sceneName, `场景执行完成`)
                .getResponse();
        } catch (error) {
            return handlerInput.responseBuilder
                .speak(`执行场景时出错：${error.message}`)
                .getResponse();
        }
    }
};

const CreateSceneIntentHandler = {
    canHandle(handlerInput) {
        return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
            && Alexa.getIntentName(handlerInput.requestEnvelope) === 'CreateSceneIntent';
    },
    async handle(handlerInput) {
        // 解析用户输入的场景配置
        const sceneName = Alexa.getSlotValue(handlerInput.requestEnvelope, 'sceneName');
        const actions = await this.parseSceneActions(handlerInput);
        
        try {
            const sceneId = sceneManager.createCustomScene(sceneName, '用户自定义场景', actions);
            
            return handlerInput.responseBuilder
                .speak(`成功创建场景${sceneName}，您可以说"执行${sceneName}场景"来使用`)
                .getResponse();
        } catch (error) {
            return handlerInput.responseBuilder
                .speak(`创建场景失败：${error.message}`)
                .getResponse();
        }
    }
};

8. 测试与调试最佳实践

8.1 本地测试环境搭建

建立完整的本地测试流程：

# tests/test_mcp_server.py
import pytest
import asyncio
from mcp_server.server import SmartHomeMCPServer
from mcp_server.device_discovery import DeviceDiscoveryService

class TestSmartHomeMCP:
    @pytest.fixture
    async def mcp_server(self):
        server = SmartHomeMCPServer()
        await server.server.initialize()
        return server

    @pytest.fixture
    async def discovery_service(self):
        service = DeviceDiscoveryService()
        await service.start_discovery()
        return service

    @pytest.mark.asyncio
    async def test_light_control(self, mcp_server):
        """测试灯光控制功能"""
        # 测试打开灯光
        result = await mcp_server.call_tool('control_light', {
            'device': 'living_room_light',
            'action': 'on'
        })
        
        assert len(result) == 1
        assert '已打开' in result[0].text
        
        # 验证状态更新
        status_result = await mcp_server.call_tool('get_device_status', {
            'device': 'living_room_light'
        })
        assert '状态: on' in status_result[0].text

    @pytest.mark.asyncio 
    async def test_device_discovery(self, discovery_service):
        """测试设备发现功能"""
        devices = await discovery_service.get_available_devices()
        assert len(devices) >= 2
        
        device_types = [device.type for device in devices]
        assert 'light' in device_types
        assert 'thermostat' in device_types

8.2 Alexa 技能测试策略

创建完整的技能测试套件：

// tests/alexa-skill-test.js
const { handler } = require('../lambda/index');
const { test } = require('ava');

// 模拟 Alexa 请求环境
function createHandlerInput(request) {
    return {
        requestEnvelope: {
            version: '1.0',
            session: {
                new: true,
                sessionId: 'amzn1.echo-api.session.123456',
                application: { applicationId: 'amzn1.ask.skill.123456' },
                user: { userId: 'amzn1.ask.account.123456' }
            },
            context: {
                System: {
                    application: { applicationId: 'amzn1.ask.skill.123456' },
                    user: { userId: 'amzn1.ask.account.123456' },
                    device: { deviceId: 'amzn1.ask.device.123456' }
                }
            },
            request: request
        },
        responseBuilder: {
            speak: function(text) { this.speakOutput = text; return this; },
            reprompt: function(text) { this.repromptOutput = text; return this; },
            withSimpleCard: function(title, content) { this.card = {title, content}; return this; },
            getResponse: function() { 
                return {
                    version: '1.0',
                    sessionAttributes: {},
                    response: {
                        outputSpeech: { type: 'PlainText', text: this.speakOutput },
                        reprompt: this.repromptOutput ? {
                            outputSpeech: { type: 'PlainText', text: this.repromptOutput }
                        } : undefined,
                        card: this.card,
                        shouldEndSession: !this.repromptOutput
                    }
                };
            }
        }
    };
}

test('LaunchRequest should return welcome message', async t => {
    const handlerInput = createHandlerInput({
        type: 'LaunchRequest',
        requestId: 'amzn1.echo-api.request.123456',
        timestamp: '2024-01-01T00:00:00Z',
        locale: 'zh-CN'
    });

    const response = await handler(handlerInput);
    t.true(response.response.outputSpeech.text.includes('欢迎使用'));
    t.false(response.response.shouldEndSession);
});

test('LightControlIntent should handle valid request', async t => {
    const handlerInput = createHandlerInput({
        type: 'IntentRequest',
        requestId: 'amzn1.echo-api.request.123456',
        timestamp: '2024-01-01T00:00:00Z',
        locale: 'zh-CN',
        intent: {
            name: 'LightControlIntent',
            slots: {
                action: { name: 'action', value: '打开' },
                device: { name: 'device', value: '客厅' }
            }
        }
    });

    const response = await handler(handlerInput);
    t.true(response.response.outputSpeech.text.includes('灯光'));
});

9. 性能优化与生产部署

9.1 MCP 服务器性能优化

针对高并发场景优化 MCP 服务器性能：

# mcp-server/optimization.py
import asyncio
from concurrent.futures import ThreadPoolExecutor
import time
from functools import wraps
import logging

def async_timed(name):
    """性能监控装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start_time = time.time()
            try:
                result = await func(*args, **kwargs)
                elapsed = time.time() - start_time
                if elapsed > 1.0:  # 超过1秒记录警告
                    logging.warning(f"{name} 执行时间较长: {elapsed:.2f}s")
                else:
                    logging.debug(f"{name} 执行时间: {elapsed:.2f}s")
                return result
            except Exception as e:
                elapsed = time.time() - start_time
                logging.error(f"{name} 执行失败 (耗时{elapsed:.2f}s): {e}")
                raise
        return wrapper
    return decorator

class OptimizedMCPServer:
    def __init__(self, max_workers=10):
        self.thread_pool = ThreadPoolExecutor(max_workers=max_workers)
        self.request_cache = {}  # 请求缓存
        self.cache_ttl = 30     # 缓存有效期（秒）
    
    @async_timed("工具调用")
    async def call_tool_optimized(self, name: str, arguments: dict):
        """优化后的工具调用方法"""
        # 检查缓存
        cache_key = self._generate_cache_key(name, arguments)
        if cache_key in self.request_cache:
            cached_result, timestamp = self.request_cache[cache_key]
            if time.time() - timestamp < self.cache_ttl:
                return cached_result
        
        # 执行实际调用
        if name in self.cpu_intensive_tools:
            # CPU密集型任务使用线程池
            result = await asyncio.get_event_loop().run_in_executor(
                self.thread_pool, 
                self._execute_cpu_intensive_tool, 
                name, arguments
            )
        else:
            # IO密集型任务直接执行
            result = await self._execute_io_tool(name, arguments)
        
        # 更新缓存
        self.request_cache[cache_key] = (result, time.time())
        return result
    
    def _generate_cache_key(self, name: str, arguments: dict) -> str:
        """生成缓存键"""
        return f"{name}:{str(sorted(arguments.items()))}"
    
    @property
    def cpu_intensive_tools(self):
        """返回CPU密集型工具列表"""
        return ['complex_calculation', 'image_processing']

9.2 生产环境部署配置

配置生产环境部署参数：

# docker-compose.prod.yml
version: '3.8'
services:
  mcp-server:
    build: ./mcp-server
    ports:
      - "8000:8000"
    environment:
      - MCP_SERVER_PORT=8000
      - LOG_LEVEL=INFO
      - MAX_WORKERS=20
      - REDIS_URL=redis://redis:6379
    deploy:
      resources:
        limits:
          memory: 512M
          cpus: '1.0'
        reservations:
          memory: 256M
          cpus: '0.5'
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  alexa-lambda:
    build: ./lambda
    environment:
      - NODE_ENV=production
      - MCP_SERVER_URL=ws://mcp-server:8000
      - AWS_REGION=us-east-1
    deploy:
      resources:
        limits:
          memory: 256M
          cpus: '0.5'

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes

volumes:
  redis_data:

10. 常见问题与解决方案

10.1 连接与通信问题

问题现象	可能原因	解决方案
MCP 服务器连接失败	网络配置错误	检查防火墙设置和端口映射
设备控制无响应	设备离线或协议不匹配	验证设备状态和通信协议
语音指令识别错误	交互模型配置问题	优化语音样本和槽位定义

10.2 性能与稳定性问题

高并发下的性能优化

• 使用连接池管理设备连接
• 实现请求队列和限流机制
• 添加缓存层减少重复计算

设备状态同步问题

• 实现乐观锁机制防止状态冲突
• 添加状态变更通知机制
• 定期同步设备实际状态

10.3 安全最佳实践

设备认证与授权

# mcp-server/security.py
import hashlib
import hmac
from datetime import datetime, timedelta

class SecurityManager:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key.encode()
    
    def generate_device_token(self, device_id: str, expires_hours: int = 24) -> str:
        """生成设备访问令牌"""
        expires = datetime.now() + timedelta(hours=expires_hours)
        payload = f"{device_id}:{expires.timestamp()}"
        signature = hmac.new(
            self.secret_key, 
            payload.encode(), 
            hashlib.sha256
        ).hexdigest()
        return f"{payload}:{signature}"
    
    def validate_token(self, token: str, device_id: str) -> bool:
        """验证设备令牌"""
        try:
            payload, signature = token