S3管理服务器

一个提供全面AWS S3管理功能的MCP服务器,支持存储桶操作、对象管理、访问控制、元数据标记和成本分析。

  • by aliamirk
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 aliamirk/aws-s3-mcp-server

使用说明

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

服务器将:

  1. 从凭据链加载AWS凭据
  2. 初始化boto3 S3客户端
  3. 注册所有14个工具
  4. 开始监听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"
  }
}

故障排除

服务器无法启动

问题: 服务器立即退出,出现凭据错误

解决方案:

  1. 验证是否配置了AWS凭据:
    aws configure list
    
  2. 测试证书:
    aws s3 ls
    
  3. 检查环境变量:
    echo $AWS_ACCESS_KEY_ID
    echo $AWS_SECRET_ACCESS_KEY
    

权限被拒绝错误

问题: 操作失败 AccessDenied 错误

解决方案:

  1. 验证IAM权限是否包括所需的S3操作
  2. 检查bucket策略不拒绝访问
  3. 验证凭据是否可以访问特定存储桶
  4. 使用AWS CLI进行测试:
    aws s3 ls s3://your-bucket-name
    

桶创建失败

问题: create_bucket 失败与 BucketAlreadyExists 或 InvalidBucketName

解决方案:

  1. 确保bucket名称全局唯一
  2. 验证存储桶名称是否符合命名规则:
    • 3-63个字符长
    • 仅限小写字母、数字和连字符
    • 必须以字母或数字开头和结尾
  3. 尝试使用具有唯一前缀的其他存储桶名称

对象上传失败

问题: upload_object 失败与 FILE_NOT_FOUND 错误

解决方案:

  1. 验证文件路径是否正确且绝对
  2. 检查文件是否存在:
    ls -la /path/to/file
    
  3. 验证文件的读取权限

元数据/标签JSON错误

问题: upload_object 或 add_tags 失败与 INVALID_METADATA 或 INVALID_TAGS

解决方案:

  1. 确保JSON格式正确
  2. 对键和字符串值使用双引号
  3. 测试JSON有效性:
    import json
    json.loads('{"key": "value"}')  # Should not raise error
    

成本计算显示为零

问题: calculate_bucket_storage_cost 返回零成本

解决方案:

  1. 验证铲斗是否包含对象:
    list_objects("my-bucket")
    
  2. 检查是否支持存储类
  3. 验证定价表中是否支持该地区

日志记录问题

问题: 未看到预期的日志输出

解决方案:

  1. 检查日志级别配置
  2. 确保在服务器启动之前配置了日志记录
  3. 检查日志是否写入文件而不是控制台

连接问题

问题: MCP客户端无法连接到服务器

解决方案:

  1. 验证服务器是否正在运行
  2. 检查服务器日志中的启动错误
  3. 验证MCP客户端配置
  4. 检查网络连接和防火墙规则

发展

设置开发环境

# 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

测试策略

该项目采用双重测试方法:

  1. 单元测试:验证具体示例和边缘案例
  2. 基于属性的测试:使用假设验证随机输入的通用属性

所有测试都使用 moto 库模拟AWS S3服务,允许在没有实际AWS凭据或基础设施的情况下运行测试。

日志记录

服务器以适当的级别记录所有操作:

  • 信息:正常操作(工具调用、成功操作)
  • 警告:可恢复错误(找不到对象,存储桶不为空)
  • 错误:严重错误(凭证故障、AWS API错误)
  • 调试:故障排除的详细信息

重要提示: 日志从不包含AWS凭据或文件内容等敏感信息。

贡献

欢迎投稿!请遵循以下指南:

  1. 分叉存储库
  2. 创建要素分支(git checkout -b feature/amazing-feature)
  3. 为新功能编写测试
  4. 确保所有测试通过(pytest)
  5. 格式代码(black src/ tests/)
  6. Lint代码(ruff check src/ tests/)
  7. 提交您的更改(git commit -m 'Add amazing feature')
  8. 推到分支(git push origin feature/amazing-feature)
  9. 打开拉取请求

许可证

MIT许可证-有关详细信息,请参阅许可证文件

支持

对于问题、疑问或贡献,请在GitHub存储库上打开问题。

致谢

其他资源