Iot Query MCP

IoT查询MCP服务器

  

一种模型上下文协议（MCP）服务器，用于对Navixy IoT Query数据库进行安全、只读的自然语言查询，并设计用于平滑的数据分析。

了解更多详情 Navixy物联网查询!

*灵感源自 西延SQL 文本转SQL研究。*

特性

• 🔒 按设计只读 -只允许SELECT查询，在多个层强制执行
• 🛡️ SQL防火墙 -使用sqlglot进行基于AST的查询分析，以实现全面的安全性
• 🧠 NL→SQL -使用LLM进行自然语言到SQL的转换
• 🔄 自动SQL修复 -自动重试并更正错误
• 📊 结果解释 -查询结果的人类可读分析
• 📝 MSchema格式 -用于精确生成SQL的结构化模式表示
• ⚡ 连接池 -使用asyncpg实现高效的异步数据库访问
• 🎯 护栏 -复杂性限制、模式分配、自动LIMIT注入

快速开始

安装

# Clone the repository
git clone https://github.com/your-org/iot-mcp.git
cd iot-mcp

# Create virtual environment
python -m venv venv
source venv/bin/activate  # or `venv\Scripts\activate` on Windows

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

配置

1. 复制示例配置：

cp config/config.example.yaml config/config.yaml

1. 编辑 config/config.yaml 使用您的数据库凭据：

database:
  host: your-postgres-host.com
  port: 5432
  database: iot_db
  user: readonly_user
  password: ${PG_PASSWORD}  # Set via environment variable
  sslmode: require
  schemas_allowlist:
    - iot
    - analytics

1. 设置环境变量：

export PG_PASSWORD="your-password"
export OPENAI_API_KEY="your-api-key"

运行服务器

# Start the MCP server (STDIO mode)
iot-query serve

# With custom config
iot-query serve --config /path/to/config.yaml

# Test connection
iot-query test-connection

# Validate config
iot-query validate-config

网络聊天界面（测试用户界面）

对于测试和开发，您可以运行基于web的聊天界面：

# Start the web UI
iot-query web

# With custom host/port
iot-query web --host 0.0.0.0 --port 8080

然后打开http://localhost:8080在您的浏览器中。

特征：

• 🔗 在界面中直接输入PostgreSQL DSN和OpenAI API密钥
• 💬 用于自然语言查询的聊天界面
• 🔀 两种模式： 仅限SQL （生成SQL）或 全周期 (NL→SQL→执行→翻译）
• 📊 带解释的实时结果
• 🎨 现代深色主题UI
• 📦 可嵌入小部件位于 /embed

Web UI Screenshot

Claude桌面集成

添加到您的Claude桌面配置(~/Library/Application Support/Claude/claude_desktop_config.json).

看 claude_desktop_configure.example.json 对于模板：

{
  "mcpServers": {
    "iot-query": {
      "command": "/path/to/venv/bin/python",
      "args": ["-m", "iot_query.cli", "serve", "--config", "/path/to/config.yaml"],
      "env": {
        "PG_PASSWORD": "your-password",
        "OPENAI_API_KEY": "your-api-key"
      }
    }
  }
}

建筑

┌─────────────────────────────────────────────────────────────┐
│                     MCP Layer                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │   Tools     │  │  Resources  │  │   Typed Responses   │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                   Guardrails Layer                           │
│  ┌─────────────────┐  ┌───────────────┐  ┌───────────────┐ │
│  │  SQL Firewall   │  │    Policy     │  │    Limits     │ │
│  │  (AST parsing)  │  │  Enforcement  │  │ (rows/joins)  │ │
│  └─────────────────┘  └───────────────┘  └───────────────┘ │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                     LLM Layer                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │  NL→SQL     │  │  SQL Fixer  │  │    Interpreter      │ │
│  │  Generator  │  │  (3 retries)│  │   (Summarization)   │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────┐
│                    Database Layer                            │
│  ┌─────────────────┐  ┌───────────────────────────────────┐ │
│  │ Connection Pool │  │      Schema Introspection         │ │
│  │    (asyncpg)    │  │  (tables/columns/FK/samples)      │ │
│  └─────────────────┘  └───────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

MCP工具

系统工具

分析工具

MCP资源

安全特性

SQL防火墙

防火墙使用基于AST的分析（sqlglot）来验证查询：

• ✅ 语句类型：只允许SELECT（包括WITH…SELECT）
• ✅ 关键字屏蔽：插入、更新、删除、删除等。
• ✅ 架构访问：基于允许列表的访问控制
• ✅ 复杂性限制：最大联接数、CTE、子查询数
• ✅ 自动限制：如果缺少LIMIT，则自动添加LIMIT
• ✅ 多语句：阻止多个语句（防止SQL注入）

配置示例

guardrails:
  deny_keywords:
    - insert
    - update
    - delete
    - drop
    - pg_sleep  # Prevent timing attacks
    - dblink    # Prevent lateral movement
  
  max_joins: 6
  max_cte: 5
  max_subqueries: 3
  
  require_limit: true
  block_select_star: false
  block_system_catalogs: true

关键模式

1. MSchema格式：LLM提示的结构化模式表示
2. SQL修复循环：自动重试并更正错误（最多3次尝试）
3. 架构上下文：丰富的上下文，包括列类型、示例和关系
4. 结果解释：查询结果的人类可读分析
5. 物联网领域上下文：内置远程信息处理数据结构知识

发展

运行测试

# Run all tests
pytest

# With coverage
pytest --cov=iot_query

# Run specific test file
pytest tests/test_firewall.py -v

对照防火墙检查SQL

iot-query check-sql "SELECT * FROM iot.devices LIMIT 10"

项目结构

iot-mcp/
├── src/iot_query/
│   ├── __init__.py
│   ├── cli.py              # CLI entry point
│   ├── config.py           # Configuration management
│   ├── models.py           # Pydantic data models
│   ├── db/
│   │   ├── connection.py   # Database connection pool
│   │   ├── executor.py     # Query execution
│   │   └── introspection.py # Schema introspection
│   ├── guardrails/
│   │   ├── firewall.py     # SQL firewall (AST-based)
│   │   └── policy.py       # Security policies
│   ├── llm/
│   │   ├── client.py       # LLM API clients
│   │   ├── nl2sql.py       # NL→SQL generator
│   │   ├── sql_fixer.py    # SQL error fixer
│   │   ├── interpreter.py  # Result interpreter
│   │   └── prompts.py      # Prompt templates
│   └── mcp/
│       └── server.py       # MCP server implementation
├── config/
│   ├── config.yaml         # Main configuration
│   └── config.example.yaml # Example configuration
├── tests/
│   ├── test_firewall.py
│   ├── test_models.py
│   └── test_config.py
├── pyproject.toml
├── requirements.txt
└── README.md

配置参考

看 config/config.example.yaml 查看完整配置选项。

关键配置部分

许可证

Apache许可证2.0-请参阅 许可证 了解详情。

参考文献

• 模型上下文协议 -MCP规范
• 西延SQL -文本到SQL模式的启示