语义操作服务器

openHAB语义MCP服务器是一个轻量级的模型上下文协议服务器,用于基于语义过滤器的命令发送、物品查询和实时状态更新。

  • by DrRSatzteil
  • Python
  • 本地部署
  • 4 星标
  • 更新于
在 GitHub 查看 DrRSatzteil/openhab-semantic-mcp

使用说明

openHAB语义MCP服务器

用于openHAB语义操作的轻量级MCP(模型上下文协议)服务器。

特性

  • 基于语义过滤器向openHAB项发送命令
  • 按位置、设备、点和属性查询项目
  • 命令验证:使用openHAB命令描述元数据来防止无效命令
  • 状态验证:使用openHAB状态描述元数据来防止无效的状态更新
  • 递归设备关系:用于设备分组的父子设备链
  • 双层次结构支持:基于类型和基于父级的语义层次结构
  • 从语义库存中获取详细的物料信息
  • 通过服务器发送事件(SSE)实时更新状态
  • 监控任务:使用webhook触发器创建基于时间的监控任务
  • 动态时区支持:使用LLM感知工具描述自动处理时区
  • Docker部署支持
  • 大型作业的安全确认

安装

Docker(推荐)

  1. 设置环境变量

    创建一个 .env 使用您的openHAB凭据在项目根目录中创建文件:

    cat > .env << EOF
    # openHAB Configuration (Required)
    OPENHAB_BASE_URL=https://your-openhab-instance.org
    OPENHAB_API_TOKEN=your_api_token_here
    
    # MCP Server Configuration
    MCP_HOST=0.0.0.0
    MCP_PORT=8000
    MCP_TRANSPORT=streamable-http
    LOG_LEVEL=INFO
    
    # Inventory Configuration
    INVENTORY_REFRESH_MINUTES=60
    
    # Monitoring Configuration (Required)
    MONITORING_WEBHOOK_URL=https://your-webhook-endpoint.org/webhook
    MONITORING_WEBHOOK_AUTH_HEADER=Authorization: Bearer your_webhook_token
    # MONITORING_TIMEZONE=Europe/Berlin  # Optional: defaults to UTC
    EOF
    
  2. 使用Docker Compose构建和运行

    docker-compose up -d
    

    服务器将在端口8000上启动。

  3. 检查日志

    docker-compose logs -f openhab-semantic-mcp
    
  4. 停止服务

    docker-compose down
    

本地开发

  1. 克隆并设置环境

    git clone <repository-url>
    cd openhab-semantic-mcp
    python -m venv venv
    source venv/bin/activate  # On Windows: venv\Scripts\activate
    pip install -e .
    
  2. 配置环境

    cp src/openhab_semantic_mcp/.env.example .env
    # Edit the .env file with your openHAB configuration
    
  3. 运行服务器

    python -m openhab_semantic_mcp
    

配置

使用环境变量配置服务器 .env 文件:

OPENHAB_BASE_URL=https://your-openhab-instance.org
OPENHAB_API_TOKEN=your_api_token_here
MCP_HOST=0.0.0.0
MCP_PORT=8000
MCP_TRANSPORT=streamable-http
LOG_LEVEL=INFO
INVENTORY_REFRESH_MINUTES=60

# Monitoring Configuration (Required)
MONITORING_WEBHOOK_URL=https://your-webhook-endpoint.org/webhook
MONITORING_WEBHOOK_AUTH_HEADER=Authorization: Bearer your_webhook_token
# MONITORING_TIMEZONE=Europe/Berlin  # Optional: defaults to UTC

# Optional Monitoring Settings
MONITORING_STORAGE_TYPE=memory
MONITORING_CLEANUP_INTERVAL_MINUTES=60
MONITORING_RETAIN_COMPLETED_DAYS=7
MONITORING_RETAIN_CANCELLED_DAYS=3
MONITORING_RETAIN_ERROR_DAYS=7
MONITORING_ENABLE_AUTO_CLEANUP=true

必修的:

  • OPENHAB_BASE_URL:openHAB实例的URL
  • OPENHAB_API_TOKEN:用于身份验证的API令牌
  • MONITORING_WEBHOOK_URL:用于监视任务通知的Webhook端点
  • MONITORING_TIMEZONE:监测任务的时区(例如,欧洲/柏林、美国/纽约)。可选:如果未设置,则默认为UTC。

可选:

  • MCP_HOST:绑定MCP服务器的主机(默认值:0.0.0.0)
  • MCP_PORT:MCP服务器的端口(默认值:8000)
  • MCP_TRANSPORT:MCP通信的传输模式(默认:可流式传输http)
    • streamable-http:基于HTTP的传输(推荐用于Docker/容器)
    • stdio:标准投入/产出运输(仅用于当地发展- 与Docker不兼容)
    • sse:服务器发送事件传输
  • LOG_LEVEL:日志记录级别(默认值:INFO)
  • INVENTORY_REFRESH_MINUTES:刷新语义清单的间隔(默认值:60)
  • MONITORING_WEBHOOK_AUTH_HEADER:webhook请求的授权标头(格式: Key: Value)
  • MONITORING_STORAGE_TYPE:存储后端类型: memory, file,或 caldav (默认值:内存)
  • MONITORING_STORAGE_CONFIG:JSON字符串形式的后端特定配置(请参见 存储后端)
  • MONITORING_CLEANUP_INTERVAL_MINUTES:清理间隔(分钟)(默认值:60)
  • MONITORING_RETAIN_COMPLETED_DAYS:保留已完成任务的天数(默认值:7)
  • MONITORING_RETAIN_CANCELLED_DAYS:保留已取消任务的天数(默认值:3)
  • MONITORING_RETAIN_ERROR_DAYS:保留错误任务的天数(默认值:7)
  • MONITORING_ENABLE_AUTO_CLEANUP:启用自动清理(默认值:true)

可用工具

MCP服务器提供以下语义工具:

核心语义工具

  • get_available_semantic_实体:发现所有语义实体(位置、设备、点、属性)
  • 获取项目:使用语义过滤器查询项目
  • send_command_to-entities:根据语义过滤器向项目发送命令
  • update_entities_state:根据语义过滤器更新项目的状态

监视工具

  • 创建监控任务:使用webhook触发器创建基于时间的监控任务
  • get_monitoring_task_status:获取监控任务的状态和详细信息
  • cancel_monitor_task:取消活动的监视任务

命令和状态验证

服务器使用openHAB的元数据自动验证命令和状态更新:

命令验证

  • 命令说明:用途 commandDescription.commandOptions 来自openHAB
  • 预防:在发送到openHAB之前阻止无效命令
  • 反馈:显示命令元数据中的有效命令

状态验证

  • 状态描述:用途 stateDescription.options 来自openHAB
  • 预防:阻止无效的状态更新
  • 反馈:显示状态元数据中的有效状态

错误响应示例

{
  "success": false,
  "error": "Command 'BLINK' not allowed. Allowed commands: ['ON', 'OFF', 'AUTO']",
  "allowed_commands": ["ON", "OFF", "AUTO"]
}

语义层次结构

服务器支持 双重等级制度 对于强大的语义查询:

基于类型的层次结构

使用带下划线分隔符的语义命名约定:

  • Lighting_CeilingLight_Downlight → 索引在 Lighting, Lighting_CeilingLight,以及 Lighting_CeilingLight_Downlight
  • Indoor_Room_DiningRoom → 索引在 Indoor, Indoor_Room,以及 Indoor_Room_DiningRoom

基于父级的层次结构

使用openHAB isPartOf 语义关系:

  • 设备可以有父设备关系
  • 位置继承自父位置
  • 没有直接位置的项目从父设备继承位置

查询示例

# Type-based queries
get_items(location="Indoor")           # All indoor items
get_items(equipment="Lighting")       # All lighting equipment
get_items(equipment="Lighting_CeilingLight")  # All ceiling lights

# Parent-based queries (with recursive location inheritance)
get_items(location="Indoor_Room_DiningRoom")  # Items in dining room (including nested equipment)
get_items(equipment="LightSource_AccentLight") # All accent lights (inherited from parent equipment)

# Combined queries
get_items(location="Indoor", equipment="LightSource")  # All indoor lighting
get_items(location="Indoor_Floor_GroundFloor", equipment="LightSource", point="Control_Switch", property="Light")  # All ground floor light switches
get_items(point="Measurement", property="Humidity")  # All humidity measurements
get_items(equipment="HVAC", point="Control")  # All controls related to HVAC

监控任务

服务器支持基于时间的任务调度的高级监控功能:

特性

  • 基于时间的调度:创建在特定时间窗口内监视项目的任务
  • Webhook通知:满足监控条件时自动触发webhook
  • 动态时区支持:具有LLM感知描述的自动时区处理
  • 多个存储后端:用于任务持久化的内存、文件或CalDAV存储
  • 自动清理:已完成任务的可配置保留策略

创建监控任务

# Monitor a light switch for 10 minutes
create_monitoring_task(
    mode="time_window",
    start_time="2026-02-10T14:48:00",  # Interpreted in configured timezone
    end_time="2026-02-10T14:58:00",
    filters={
        "location": "Indoor_Room_LivingRoom",
        "equipment": "LightSource_FloorLamp", 
        "point": "Control_Switch"
    }
)

# One-shot task - triggers once when condition is met
create_monitoring_task(
    mode="one_shot",
    end_time="2026-02-10T23:59:00",
    filters={"point": "Status_OpenState", "state": {"kind": "exact", "states": ["OPEN"]}}
)

时区处理

所有时间都会在配置的时区中自动解释:

# Configure timezone (optional - defaults to UTC)
MONITORING_TIMEZONE=Europe/Berlin    # European time
MONITORING_TIMEZONE=America/New_York  # US Eastern time  
MONITORING_TIMEZONE=Asia/Tokyo        # Japan time

LLM自动接收工具描述中的时区信息,确保正确的时间解释。

存储后端

选择后端 MONITORING_STORAGE_TYPE 并通过以下方式进行配置 MONITORING_STORAGE_CONFIG (由后端名称键入的JSON字符串):

记忆 (默认)-在内存存储中,重新启动时数据丢失:

MONITORING_STORAGE_TYPE=memory
MONITORING_STORAGE_CONFIG='{"memory": {}}'

文件 -JSON文件持久性:

MONITORING_STORAGE_TYPE=file
MONITORING_STORAGE_CONFIG='{"file": {"file_path": "monitoring_tasks.json"}}'

CalDAV -带背景同步的基于日历的存储:

MONITORING_STORAGE_TYPE=caldav
MONITORING_STORAGE_CONFIG='{"caldav": {"url": "https://caldav.example.org/remote.php/dav/principals/users/user/", "username": "user", "password": "pass", "calendar_name": "monitoring", "sync_interval": 300}}'

Webhook有效载荷

当监控任务触发时,它会发送一个带有详细事件信息的webhook:

{
  "task_id": "monitor_abc123",
  "mode": "time_window",
  "triggered_at": "2026-02-10T14:52:30+01:00",
  "trigger_count": 1,
  "item": {
    "name": "floorlamp_livingroom_toggle",
    "state": "ON",
    "display_state": "An",
    "unit": null
  },
  "task_config": {
    "filters": {
      "location": "Indoor_Room_LivingRoom",
      "equipment": "LightSource_FloorLamp"
    },
    "refinement": null,
    "last_state_transition": "2026-02-10T14:48:00+01:00"
  },
  "time_window": {
    "start_time": "2026-02-10T14:48:00+01:00",
    "end_time": "2026-02-10T14:58:00+01:00"
  }
}

测试

# Install test dependencies
pip install -e ".[test]"

# Run tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=openhab_semantic_mcp --cov-report=html

测试覆盖范围包括:

  • DTO模型和关系
  • 具有双重层次结构的库存索引
  • openHAB客户端语义解析
  • 监控系统(服务层、触发器评估、webhook管理)
  • CalDAV后端(连接、事件映射、日历同步)
  • 存储后端(内存、文件、CalDAV)