使用说明
AWS S3 MCP服务器
提供全面AWS S3管理功能的MCP(模型上下文协议)服务器。该服务器使用fastmcp框架和boto3构建,提供了14个用于bucket操作、对象管理、访问控制、元数据标记和成本分析的工具。
目录
特性
水桶管理
- list_buckets:列出您AWS帐户中的所有S3存储桶
- create_bucket:使用可选的区域规范创建新的S3存储桶
- delete_bucket:删除空的S3存储桶
- get_bucket_range:检索存储桶的AWS区域
对象操作
- list_objects:使用可选的前缀过滤和分页列出bucket中的对象
- 获取对象元数据:检索S3对象的元数据
- upload_object:使用自定义元数据和内容类型将文件上传到S3
- 删除对象:从S3中删除对象
- 复制对象:复制S3中的对象(保留元数据)
- move_object:在S3中移动对象(原子复制,然后删除)
访问控制
- generate-presigned_download_url:生成限时下载URL
元数据和标记
- 添加标签:在S3对象上添加或更新标签
- search_by_tag:按标记键和值搜索对象
成本分析
- calculate_bucket_storage_cost:计算估计的每月存储成本
需求
- Python 3.8或更高版本
- 具有适当S3权限的AWS帐户
- 已配置AWS凭据(请参阅 配置 部分)
安装
源自源头
# Clone the repository
git clone <repository-url>
cd aws-s3-mcp-server
# Install dependencies
pip install -r requirements.txt
# Install the package in development mode
pip install -e .
使用pip
pip install aws-s3-mcp-server
配置
AWS凭据
服务器使用标准的AWS凭据链。使用以下方法之一配置您的凭据:
1.环境变量
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_SESSION_TOKEN=your_session_token # Optional, for temporary credentials
export AWS_DEFAULT_REGION=us-east-1 # Optional
2.AWS凭据文件
创建或编辑 ~/.aws/credentials:
[default]
aws_access_key_id = your_access_key
aws_secret_access_key = your_secret_key
(可选)在中配置默认区域 ~/.aws/config:
[default]
region = us-east-1
3.IAM角色(AWS基础设施)
在AWS基础架构(EC2、ECS、Lambda)上运行时,服务器会自动使用附加到实例的IAM角色。不需要额外的配置。
所需IAM权限
AWS凭据必须具有以下S3权限:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:ListAllMyBuckets",
"s3:CreateBucket",
"s3:DeleteBucket",
"s3:GetBucketLocation",
"s3:ListBucket",
"s3:GetObject",
"s3:GetObjectMetadata",
"s3:PutObject",
"s3:DeleteObject",
"s3:CopyObject",
"s3:GetObjectTagging",
"s3:PutObjectTagging"
],
"Resource": "*"
}
]
}
对于生产环境,考虑限制 Resource 字段到特定桶:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": ["s3:ListAllMyBuckets"],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"s3:CreateBucket",
"s3:DeleteBucket",
"s3:GetBucketLocation",
"s3:ListBucket"
],
"Resource": "arn:aws:s3:::your-bucket-prefix-*"
},
{
"Effect": "Allow",
"Action": [
"s3:GetObject",
"s3:GetObjectMetadata",
"s3:PutObject",
"s3:DeleteObject",
"s3:CopyObject",
"s3:GetObjectTagging",
"s3:PutObjectTagging"
],
"Resource": "arn:aws:s3:::your-bucket-prefix-*/*"
}
]
}
用法
启动服务器
# Run the server
aws-s3-mcp-server
# Or run directly with Python
python -m aws_s3_mcp_server.server
服务器将:
- 从凭据链加载AWS凭据
- 初始化boto3 S3客户端
- 注册所有14个工具
- 开始监听MCP客户端连接
如果找不到凭据,服务器将记录错误并以非零状态代码退出。
与MCP客户端一起使用
服务器运行后,MCP客户端可以连接并调用可用工具。有关连接的详细信息,请参阅您的MCP客户端文档。
工具参考
所有工具都返回具有一致结构的JSON格式的响应:
成功响应:
{
"success": true,
"data": {
// Tool-specific data
}
}
错误响应:
{
"success": false,
"error": {
"message": "Descriptive error message",
"code": "AWS_ERROR_CODE",
"type": "ErrorType"
}
}
水桶管理
list_buckets
列出AWS帐户中的所有S3存储桶。
参数: 无
退货:
{
"success": true,
"data": {
"buckets": [
{
"name": "my-bucket",
"creation_date": "2024-01-01T00:00:00+00:00"
}
]
}
}
例子:
result = list_buckets()
create_bucket
创建一个新的S3存储桶。
参数:
bucket_name(string,必填):要创建的bucket的名称(必须全局唯一)region(字符串,可选):存储桶的AWS区域(默认值:us-east-1)
退货:
{
"success": true,
"data": {
"bucket_name": "my-new-bucket",
"region": "us-west-2",
"message": "Bucket 'my-new-bucket' created successfully in region us-west-2"
}
}
例子:
# Create bucket in default region (us-east-1)
result = create_bucket("my-new-bucket")
# Create bucket in specific region
result = create_bucket("my-new-bucket", region="us-west-2")
笔记:
- 存储桶名称在所有AWS帐户中必须是全局唯一的
- 存储桶名称的长度必须为3-63个字符
- 存储桶名称可以包含小写字母、数字和连字符
- 存储桶名称必须以字母或数字开头和结尾
delete_bucket
删除S3存储桶。
参数:
bucket_name(string,必填):要删除的bucket的名称
退货:
{
"success": true,
"data": {
"bucket_name": "my-old-bucket",
"message": "Bucket 'my-old-bucket' deleted successfully"
}
}
例子:
result = delete_bucket("my-old-bucket")
笔记:
- 桶必须为空才能删除
- 如果桶中包含对象,则操作将失败,并显示
BucketNotEmpty错误
get_bucket_range
获取存储桶所在的AWS区域。
参数:
bucket_name(string,必填):桶的名称
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"region": "us-west-2"
}
}
例子:
result = get_bucket_region("my-bucket")
对象操作
list_objects
列出S3存储桶中的对象。
参数:
bucket_name(string,必填):桶的名称prefix(字符串,可选):按键前缀筛选对象(默认值:“”)max_keys(整数,可选):要返回的最大对象数(默认值:1000)
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"prefix": "docs/",
"objects": [
{
"key": "docs/file1.txt",
"size": 1024,
"last_modified": "2024-01-01T00:00:00+00:00"
}
],
"count": 1
}
}
例子:
# List all objects
result = list_objects("my-bucket")
# List objects with prefix
result = list_objects("my-bucket", prefix="docs/")
# List with limit
result = list_objects("my-bucket", prefix="docs/", max_keys=10)
获取对象元数据
获取S3对象的元数据。
参数:
bucket_name(string,必填):桶的名称object_key(string,必填):对象的键
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"object_key": "docs/file.txt",
"content_type": "text/plain",
"size": 1024,
"last_modified": "2024-01-01T00:00:00+00:00",
"etag": "\"abc123...\"",
"metadata": {
"author": "John Doe",
"project": "demo"
}
}
}
例子:
result = get_object_metadata("my-bucket", "docs/file.txt")
upload_object
将文件上传到S3。
参数:
bucket_name(string,必填):桶的名称object_key(string,必填):S3中对象的密钥file_path(字符串,必填):要上传的本地文件路径content_type(string,可选):文件的MIME类型metadata(string,可选):自定义元数据为JSON字符串
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"object_key": "docs/file.txt",
"message": "Successfully uploaded object 'docs/file.txt' to bucket 'my-bucket'"
}
}
例子:
# Simple upload
result = upload_object("my-bucket", "docs/file.txt", "/tmp/file.txt")
# Upload with content type
result = upload_object("my-bucket", "docs/file.txt", "/tmp/file.txt",
content_type="text/plain")
# Upload with metadata
result = upload_object("my-bucket", "docs/file.txt", "/tmp/file.txt",
content_type="text/plain",
metadata='{"author": "John", "project": "demo"}')
笔记:
- 元数据参数必须是表示字典的有效JSON字符串
- 自定义元数据键会自动添加前缀
x-amz-meta-AWS
删除对象
从S3中删除对象。
参数:
bucket_name(string,必填):桶的名称object_key(string,必填):要删除的对象的键
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"object_key": "docs/file.txt",
"message": "Successfully deleted object 'docs/file.txt' from bucket 'my-bucket'"
}
}
例子:
result = delete_object("my-bucket", "docs/file.txt")
笔记:
- 此操作是幂等的-即使对象不存在,它也会成功
复制对象
将对象从一个S3位置复制到另一个位置。
参数:
source_bucket(字符串,必填):源存储桶名称source_key(字符串,必填):源对象键dest_bucket(字符串,必填):目标存储桶名称dest_key(字符串,必填):目标对象键
退货:
{
"success": true,
"data": {
"source_bucket": "source-bucket",
"source_key": "source/file.txt",
"dest_bucket": "dest-bucket",
"dest_key": "dest/file.txt",
"message": "Successfully copied object from 'source-bucket/source/file.txt' to 'dest-bucket/dest/file.txt'"
}
}
例子:
# Copy within same bucket
result = copy_object("my-bucket", "docs/file.txt", "my-bucket", "archive/file.txt")
# Copy to different bucket
result = copy_object("my-bucket", "docs/file.txt", "backup-bucket", "file.txt")
笔记:
- 源对象中的所有元数据都保留在副本中
- 源对象保持不变
move_object
将对象从一个S3位置移动到另一个位置。
参数:
source_bucket(字符串,必填):源存储桶名称source_key(字符串,必填):源对象键dest_bucket(字符串,必填):目标存储桶名称dest_key(字符串,必填):目标对象键
退货:
{
"success": true,
"data": {
"source_bucket": "source-bucket",
"source_key": "source/file.txt",
"dest_bucket": "dest-bucket",
"dest_key": "dest/file.txt",
"message": "Successfully moved object from 'source-bucket/source/file.txt' to 'dest-bucket/dest/file.txt'"
}
}
例子:
# Move within same bucket
result = move_object("my-bucket", "temp/file.txt", "my-bucket", "archive/file.txt")
# Move to different bucket
result = move_object("my-bucket", "temp/file.txt", "backup-bucket", "file.txt")
笔记:
- 此操作是原子性的:如果复制失败,源对象保持不变
- 该移动以复制后删除的方式实现
访问控制
generate-presigned_download_url
生成用于下载S3对象的预签名URL。
参数:
bucket_name(string,必填):桶的名称object_key(string,必填):对象的键expiration(整数,可选):URL过期时间(秒)(默认值:3600)
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"object_key": "docs/file.txt",
"presigned_url": "https://my-bucket.s3.amazonaws.com/docs/file.txt?...",
"expiration_seconds": 3600
}
}
例子:
# Generate URL with default expiration (1 hour)
result = generate_presigned_download_url("my-bucket", "docs/file.txt")
# Generate URL with custom expiration (2 hours)
result = generate_presigned_download_url("my-bucket", "docs/file.txt", expiration=7200)
笔记:
- 预签名的URL授予临时下载访问权限,无需AWS凭据
- URL在指定的持续时间后过期
- 任何拥有该URL的人都可以下载该对象,直到它过期
元数据和标记
添加标签
在S3对象上添加或更新标签。
参数:
bucket_name(string,必填):桶的名称object_key(string,必填):对象的键tags(string,必填):标记为JSON字符串(键值对)
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"object_key": "docs/file.txt",
"tags_added": {
"project": "demo",
"env": "prod"
},
"message": "Successfully added tags to object 'docs/file.txt' in bucket 'my-bucket'"
}
}
例子:
# Add tags to an object
result = add_tags("my-bucket", "docs/file.txt",
'{"project": "demo", "env": "prod", "owner": "team-a"}')
笔记:
- 如果标记键已存在,则其值将被更新
- 保留请求中未指定的现有标签
- tags参数必须是表示字典的有效JSON字符串
search_by_tag
按标签搜索bucket中的对象。
参数:
bucket_name(string,必填):桶的名称tag_key(字符串,必填):要搜索的标记键tag_value(字符串,必填):要匹配的标记值
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"tag_key": "project",
"tag_value": "demo",
"matching_objects": [
{
"key": "docs/file1.txt",
"size": 1024
},
{
"key": "docs/file2.txt",
"size": 2048
}
],
"count": 2
}
}
例子:
# Search for objects with specific tag
result = search_by_tag("my-bucket", "project", "demo")
# Search for objects by environment
result = search_by_tag("my-bucket", "env", "prod")
笔记:
- 此操作列出bucket中的所有对象并检查其标签
- 对于大型铲斗,此操作可能需要一些时间
- 标记键和值必须完全匹配(区分大小写)
成本分析
calculate_bucket_storage_cost
计算S3存储桶的存储成本。
参数:
bucket_name(string,必填):桶的名称storage_class(字符串,可选):用于定价的存储类(默认值:“STANDARD”)pricing_region(字符串,可选):用于定价的AWS区域(如果未指定,则使用bucket区域)
支持的存储类:
STANDARD-标准存储(默认)INTELLIGENT_TIERING-智能分层存储STANDARD_IA-标准不常访问ONEZONE_IA-一个区域不常访问GLACIER-冰川储存GLACIER_DEEP_ARCHIVE-冰川深处档案馆
退货:
{
"success": true,
"data": {
"bucket_name": "my-bucket",
"total_size_bytes": 1073741824,
"total_size_gb": 1.0,
"storage_class": "STANDARD",
"region": "us-east-1",
"price_per_gb": 0.023,
"estimated_monthly_cost_usd": 0.023
}
}
例子:
# Calculate cost with default storage class (STANDARD)
result = calculate_bucket_storage_cost("my-bucket")
# Calculate cost for Glacier storage
result = calculate_bucket_storage_cost("my-bucket", storage_class="GLACIER")
# Calculate cost for specific region
result = calculate_bucket_storage_cost("my-bucket",
storage_class="STANDARD_IA",
pricing_region="us-west-2")
笔记:
- 成本计算基于截至实施日期的AWS S3定价
- 实际成本可能因AWS定价变化而异
- 此计算仅包括存储成本,不包括请求成本或数据传输成本
- 如需准确计费,请参阅您的AWS计费仪表板
错误处理
服务器通过结构化错误响应实现了全面的错误处理。
常见错误类型
| 错误代码 | 描述 | 示例 |
|---|---|---|
NoSuchBucket |
指定的bucket不存在 | 尝试列出不存在bucket中的对象 |
NoSuchKey |
指定的对象不存在 | 正在尝试获取不存在对象的元数据 |
BucketAlreadyExists |
同名bucket已存在 | 创建一个同名bucket |
BucketNotEmpty |
无法删除bucket,因为它包含对象 | 正在尝试删除包含对象的bucket |
AccessDenied |
操作权限不足 | 尝试执行没有适当IAM权限的操作 |
InvalidBucketName |
存储桶名称无效 | 存储桶名称中使用了无效字符 |
FILE_NOT_FOUND |
找不到要上传的本地文件 | 正在尝试上传不存在的本地文件 |
INVALID_METADATA |
元数据JSON格式无效 | 为元数据提供格式错误的JSON |
INVALID_TAGS |
标签JSON格式无效 | 为标签提供格式错误的JSON |
INVALID_STORAGE_CLASS |
不支持的存储类 | 使用不受支持的存储类名 |
错误响应格式
所有错误都遵循以下结构:
{
"success": false,
"error": {
"message": "Human-readable error description",
"code": "ERROR_CODE",
"type": "ErrorType"
}
}
错误响应示例
找不到桶:
{
"success": false,
"error": {
"message": "Bucket 'non-existent-bucket' does not exist",
"code": "NoSuchBucket",
"type": "NoSuchBucket"
}
}
桶未空:
{
"success": false,
"error": {
"message": "Cannot delete bucket 'my-bucket': bucket must be empty",
"code": "BucketNotEmpty",
"type": "BucketNotEmpty"
}
}
拒绝访问:
{
"success": false,
"error": {
"message": "Access denied to bucket 'my-bucket'",
"code": "AccessDenied",
"type": "AccessDenied"
}
}
故障排除
服务器无法启动
问题: 服务器立即退出,出现凭据错误
解决方案:
- 验证是否配置了AWS凭据:
aws configure list - 测试证书:
aws s3 ls - 检查环境变量:
echo $AWS_ACCESS_KEY_ID echo $AWS_SECRET_ACCESS_KEY
权限被拒绝错误
问题: 操作失败 AccessDenied 错误
解决方案:
- 验证IAM权限是否包括所需的S3操作
- 检查bucket策略不拒绝访问
- 验证凭据是否可以访问特定存储桶
- 使用AWS CLI进行测试:
aws s3 ls s3://your-bucket-name
桶创建失败
问题: create_bucket 失败与 BucketAlreadyExists 或 InvalidBucketName
解决方案:
- 确保bucket名称全局唯一
- 验证存储桶名称是否符合命名规则:
- 3-63个字符长
- 仅限小写字母、数字和连字符
- 必须以字母或数字开头和结尾
- 尝试使用具有唯一前缀的其他存储桶名称
对象上传失败
问题: upload_object 失败与 FILE_NOT_FOUND 错误
解决方案:
- 验证文件路径是否正确且绝对
- 检查文件是否存在:
ls -la /path/to/file - 验证文件的读取权限
元数据/标签JSON错误
问题: upload_object 或 add_tags 失败与 INVALID_METADATA 或 INVALID_TAGS
解决方案:
- 确保JSON格式正确
- 对键和字符串值使用双引号
- 测试JSON有效性:
import json json.loads('{"key": "value"}') # Should not raise error
成本计算显示为零
问题: calculate_bucket_storage_cost 返回零成本
解决方案:
- 验证铲斗是否包含对象:
list_objects("my-bucket") - 检查是否支持存储类
- 验证定价表中是否支持该地区
日志记录问题
问题: 未看到预期的日志输出
解决方案:
- 检查日志级别配置
- 确保在服务器启动之前配置了日志记录
- 检查日志是否写入文件而不是控制台
连接问题
问题: MCP客户端无法连接到服务器
解决方案:
- 验证服务器是否正在运行
- 检查服务器日志中的启动错误
- 验证MCP客户端配置
- 检查网络连接和防火墙规则
发展
设置开发环境
# Clone the repository
git clone <repository-url>
cd aws-s3-mcp-server
# Install development dependencies
pip install -e ".[dev]"
# Run tests
pytest
# Run tests with coverage
pytest --cov=src --cov-report=html
# Format code
black src/ tests/
# Lint code
ruff check src/ tests/
# Type check
mypy src/
项目结构
aws-s3-mcp-server/
├── src/
│ └── aws_s3_mcp_server/
│ ├── __init__.py
│ ├── server.py # Main server implementation
│ ├── cost_calculator.py # Cost calculation logic
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── bucket_tools.py # Bucket management tools
│ │ └── object_tools.py # Object management tools
│ └── utils/
│ ├── __init__.py
│ ├── error_handler.py # Error handling utilities
│ └── logging_config.py # Logging configuration
├── tests/
│ ├── unit/ # Unit tests
│ ├── property/ # Property-based tests
│ └── integration/ # Integration tests
├── .kiro/
│ └── specs/ # Specification documents
├── pyproject.toml # Project configuration
├── requirements.txt # Dependencies
└── README.md # This file
运行测试
# Run all tests
pytest
# Run unit tests only
pytest tests/unit/
# Run property-based tests only
pytest tests/property/
# Run specific test file
pytest tests/unit/test_server_initialization.py
# Run with verbose output
pytest -v
# Run with coverage report
pytest --cov=src --cov-report=term-missing
# Run specific test
pytest tests/unit/test_bucket_tools.py::test_list_buckets_success
测试策略
该项目采用双重测试方法:
- 单元测试:验证具体示例和边缘案例
- 基于属性的测试:使用假设验证随机输入的通用属性
所有测试都使用 moto 库模拟AWS S3服务,允许在没有实际AWS凭据或基础设施的情况下运行测试。
日志记录
服务器以适当的级别记录所有操作:
- 信息:正常操作(工具调用、成功操作)
- 警告:可恢复错误(找不到对象,存储桶不为空)
- 错误:严重错误(凭证故障、AWS API错误)
- 调试:故障排除的详细信息
重要提示: 日志从不包含AWS凭据或文件内容等敏感信息。
贡献
欢迎投稿!请遵循以下指南:
- 分叉存储库
- 创建要素分支(
git checkout -b feature/amazing-feature) - 为新功能编写测试
- 确保所有测试通过(
pytest) - 格式代码(
black src/ tests/) - Lint代码(
ruff check src/ tests/) - 提交您的更改(
git commit -m 'Add amazing feature') - 推到分支(
git push origin feature/amazing-feature) - 打开拉取请求
许可证
MIT许可证-有关详细信息,请参阅许可证文件
支持
对于问题、疑问或贡献,请在GitHub存储库上打开问题。