使用说明

亚马逊物联网核心MCP服务器

一种模型上下文协议（MCP）服务器，能够使用多种身份验证方法将MQTT消息发布到Amazon IoT Core。

特性

多种身份验证方法:
- 双向TLS（X.509证书）
- 自定义授权人（用户名/密码）
- AWS凭据（MQTT上的WebSocket）
自动重试逻辑：对失败的连接重试3次
灵活的消息发布：支持JSON和二进制有效载荷
QoS和保留支持：可配置的服务质量和消息保留

安装

1.克隆存储库

git clone <repository-url>
cd sample-MCP_server-for-Amazon_IoT_Core

2.创建虚拟环境

使用Python venv：

macOS/Linux：

python3 -m venv venv
source venv/bin/activate

窗户：

python -m venv venv
venv\Scripts\activate

使用紫外线（更快的替代品）：

macOS/Linux/Windows：

# Install uv if not already installed
pip install uv

# Create and activate virtual environment
uv venv
source .venv/bin/activate  # macOS/Linux
# or
.venv\Scripts\activate     # Windows

3.安装依赖项

使用pip：

pip install -r requirements.txt

使用紫外线：

uv pip install -r requirements.txt

快速开始

# Clone and setup
git clone <repository-url>
cd sample-MCP_server-for-Amazon_IoT_Core

# Create virtual environment (choose one method)
python3 -m venv venv && source venv/bin/activate  # Python venv
# OR
uv venv && source .venv/bin/activate              # uv (faster)

# Install dependencies
pip install -r requirements.txt  # or: uv pip install -r requirements.txt

# Configure AWS credentials and IoT endpoint
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export IOT_ENDPOINT=your-endpoint.iot.region.amazonaws.com

# Run MCP server
python server.py

配置

AWS凭据

对于WebSocket身份验证，请确保您的AWS凭据已配置：

aws configure

或设置环境变量：

export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_SESSION_TOKEN=your_session_token  # Optional

物联网核心端点

将IoT Core端点设置为环境变量（可选）：

export IOT_ENDPOINT=your-endpoint.iot.region.amazonaws.com

获取您的物联网核心端点：

aws iot describe-endpoint --endpoint-type iot:Data-ATS --region ap-northeast-1

用法

运行MCP服务器

请确保您的虚拟环境已先激活：

macOS/Linux：

source venv/bin/activate  # or source .venv/bin/activate if using uv

窗户：

venv\Scripts\activate     # or .venv\Scripts\activate if using uv

然后运行服务器：

python server.py

代理的MCP配置

亚马逊Q CLI

创建MCP配置文件：

cd ~/.aws/amazonq/
touch mcp.json

编辑 ~/.aws/amazonq/mcp.json:

{
  "mcpServers": {
    "amazon-iotcore": {
      "command": "path_of_your_python_venv/bin/python",
      "args": ["/absolute/path/to/sample-MCP_server-for-Amazon_IoT_Core/server.py"],
      "env": {
        "AWS_ACCESS_KEY_ID": "your_access_key",
        "AWS_SECRET_ACCESS_KEY": "your_secret_key",
        "IOT_ENDPOINT": "your-endpoint.iot.region.amazonaws.com"
      }
    }
  }
}

在Q CLI中使用：

q chat
# Check MCP servers: /mcp
# Then ask: "Publish a message to IoT Core topic 'test' with payload 'Hello'"

克劳德桌面版

配置文件： ~/Library/Application Support/Claude/claude_desktop_config.json （macOS）

{
  "mcpServers": {
    "amazon-iotcore": {
      "command": "python", 
      "args": ["/absolute/path/to/sample-MCP_server-for-Amazon_IoT_Core/server.py"],
      "env": {
        "AWS_ACCESS_KEY_ID": "your_key",
        "AWS_SECRET_ACCESS_KEY": "your_secret",
        "IOT_ENDPOINT": "your-endpoint.iot.region.amazonaws.com"
      }
    }
  }
}

📖 有关详细的配置指南，请参阅 docs/mcp_gonfiguration.md

测试

简单测试脚本

使用AWS凭据：

export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export IOT_ENDPOINT=your-endpoint.iot.region.amazonaws.com

cd examples
python simple_publish.py test/topic "Hello IoT Core"

使用X.509证书：

cd examples
python simple_publish.py device/data "sensor data" \
  --endpoint your-endpoint.iot.region.amazonaws.com \
  --cert /path/to/device-cert.pem \
  --key /path/to/private-key.pem

使用自定义授权器：

cd examples
python simple_publish.py device/status "online" \
  --endpoint your-endpoint.iot.region.amazonaws.com \
  --username device001 \
  --password device_password \
  --custom-authorizer MyCustomAuthorizer

通过QoS和保留：

cd examples
python simple_publish.py alerts/fire "Emergency!" \
  --endpoint your-endpoint.iot.region.amazonaws.com \
  --qos 2 --retain

详细示例

这 examples/example_usage.py 显示了不同的身份验证方法：

编辑文件，将占位符值替换为实际的IoT Core端点和凭据
运行： python examples/example_usage.py

工具：publish_mqtt

向Amazon IoT Core发布MQTT消息。

所需参数：

topic：要发布到的MQTT主题
payload：消息有效负载（JSON字符串或纯文本）

可选参数：

endpoint：IoT核心端点URL（如果未提供，则使用IoT_endpoint环境变量）

可选参数：

cert_path：设备证书文件的路径
key_path：私钥文件的路径
username：自定义授权器的MQTT用户名
password：自定义授权人的MQTT密码
custom_authorizer：自定义授权人名称
qos：服务质量级别（0、1或2）-默认值：1
retain：保留消息标志-默认值：false

身份验证优先级：

双向TLS：如果两者都有 cert_path 和 key_path 已提供且文件存在
自定义授权人：如果 username 和 password 提供
AWS凭据：默认回退使用WebSocket over MQTT

示例用法：

使用X.509证书：

{
  "endpoint": "your-endpoint.iot.us-east-1.amazonaws.com",
  "topic": "device/data",
  "payload": "{\"temperature\": 25.5, \"humidity\": 60}",
  "cert_path": "/path/to/device-cert.pem",
  "key_path": "/path/to/private-key.pem",
  "qos": 1
}

使用自定义授权器：

{
  "endpoint": "your-endpoint.iot.us-east-1.amazonaws.com", 
  "topic": "device/data",
  "payload": "{\"sensor_id\": \"temp001\", \"value\": 23.1}",
  "username": "device001",
  "password": "device_password",
  "custom_authorizer": "MyCustomAuthorizer"
}

使用AWS凭据：

{
  "endpoint": "your-endpoint.iot.us-east-1.amazonaws.com",
  "topic": "device/status", 
  "payload": "Device online",
  "qos": 0,
  "retain": true
}

日志记录配置

日志等级控制

通过设置环境变量 IOT_MCP_LOG_LEVEL 来控制日志等级：

可用的日志等级：

调试 - 最详细的日志，包含所有调试信息
```
export IOT_MCP_LOG_LEVEL=DEBUG
```
信息 - 一般信息日志（默认等级）
```
export IOT_MCP_LOG_LEVEL=INFO
```
警告 - 警告信息
```
export IOT_MCP_LOG_LEVEL=WARNING
```
错误 - 错误信息
```
export IOT_MCP_LOG_LEVEL=ERROR
```
关键的 - 严重错误
```
export IOT_MCP_LOG_LEVEL=CRITICAL
```

日志文件位置

日志文件保存在：./logs/iotcore_mcp.log

文件大小限制：10MB
保留历史文件：5个
编码：UTF-8
自动轮转：当文件达到10MB时自动创建新文件

日志格式

时间戳 - 模块名 - 日志等级 - 函数名:行号 - 消息内容

示例：

2025-09-07 17:30:15,123 - __main__ - INFO - publish_message:45 - Starting publish_message - topic: device/data, qos: 1, retain: False

使用示例

启动时设置日志等级：

# 设置为DEBUG等级（最详细）
export IOT_MCP_LOG_LEVEL=DEBUG
python server.py

# 设置为ERROR等级（只记录错误）
export IOT_MCP_LOG_LEVEL=ERROR
python server.py

在不同场景下的推荐设置：

开发调试：IOT_MCP_LOG_LEVEL=DEBUG
生产环境：IOT_MCP_LOG_LEVEL=INFO
故障排查：IOT_MCP_LOG_LEVEL=DEBUG
性能优化：IOT_MCP_LOG_LEVEL=WARNING

故障排查

当外部调用MCP server失败时，建议：

设置DEBUG日志等级：
```
export IOT_MCP_LOG_LEVEL=DEBUG
```
检查日志文件：
```
tail -f logs/iotcore_mcp.log
```
常见问题检查点：
- IoT Core endpoint配置
- 证书文件路径和权限
- AWS凭证配置
- 网络连接状态
- MQTT主题权限

错误处理

连接失败：自动重试3次
身份验证错误：清除凭据问题的错误消息
网络问题：超时处理和适当的错误报告
无效参数：验证和描述性错误消息

需求

Python 3.8+
AWS IoT设备SDK for Python v2
MCP（模型上下文协议）

许可证

MIT许可证

贡献

复刻仓库
创建要素分支
进行更改
如果适用，添加测试
提交拉取请求

支持

对于问题和疑问：

在GitHub存储库中创建问题
查看AWS IoT Core文档
审查MCP协议规范

使用说明

亚马逊物联网核心MCP服务器

特性

安装

1.克隆存储库

2.创建虚拟环境

3.安装依赖项

快速开始

配置

AWS凭据

物联网核心端点

用法

运行MCP服务器

代理的MCP配置

亚马逊Q CLI

克劳德桌面版

测试

简单测试脚本

详细示例

工具：publish_mqtt

所需参数：

可选参数：

可选参数：

身份验证优先级：

示例用法：

日志记录配置

日志等级控制

可用的日志等级：

日志文件位置

日志格式

使用示例

故障排查

错误处理

需求

许可证

贡献

支持

相关 MCP

串口AI控制工具

智能家居设备控制服务器

NodeMCU设备管理服务

NodeMCU设备管理服务

TCP套接字MCP服务器

MCP协议服务器