亚马逊物联网核心MCP服务器
一种模型上下文协议（MCP）服务器，能够使用多种身份验证方法将MQTT消息发布到Amazon IoT Core。
特性
- 多种身份验证方法:
- 双向TLS（X.509证书） - 自定义授权人（用户名/密码） - AWS凭据（MQTT上的WebSocket）
- 自动重试逻辑：对失败的连接重试3次
- 灵活的消息发布：支持JSON和二进制有效载荷
- QoS和保留支持：可配置的服务质量和消息保留
安装
1.克隆存储库
git clone
cd sample-MCP_server-for-Amazon_IoT_Core2.创建虚拟环境
使用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 # Windows3.安装依赖项
使用pip：
pip install -r requirements.txt使用紫外线：
uv pip install -r requirements.txt快速开始
# Clone and setup
git clone
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）-默认值：1retain：保留消息标志-默认值：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协议规范