AWS Kinesis 数据流 MCP 服务器

用于与 AWS Kinesis 数据流交互的 MCP 服务器。

该综合服务器提供了操作 Kinesis 数据流管理功能，包含超过 30 种工具，用于管理数据流、分片、记录、消费者等。

可用的 MCP 工具

数据流管理

• create_stream - 在 PROVISIONED 或 ON_DEMAND 模式下创建一个 Kinesis 数据流
• delete_stream - 删除一个 Kinesis 数据流及其所有分片和数据
• describe_stream - 返回关于数据流的详细信息，包括分片、状态和配置
• describe_stream_summary - 返回数据流的概要信息，不包括分片详情
• list_streams - 返回账户中的 Kinesis 数据流列表
• update_shard_count - 更新 PROVISIONED 模式下数据流的分片数量
• update_stream_mode - 更新数据流的容量模式（PROVISION-ED/ON_DEMAND）

分片管理

• list_shards - 返回 Kinesis 数据流中的分片列表
• merge_shards - 合并两个相邻的分片（仅适用于 PROVISIONED 流）
• split_shard - 将一个分片拆分为两个新的分片（仅适用于 PROVISIONED 流）

记录操作

• put_record - 将单个数据记录写入 Kinesis 数据流
• put_records - 在一次调用中将多个数据记录写入流
• get_shard_iterator - 获取用于从特定分片读取记录的分片迭代器
• get_records - 使用分片迭代器从分片中检索记录

消费者管理（增强型扇出）

• register_stream_consumer - 注册一个消费者以使用增强型扇出功能
• deregister_stream_consumer - 注销一个消费者
• describe_stream_consumer - 返回已注册消费者的详细信息
• list_stream_consumers - 列出已注册到流的消费者

监控和指标

• enable_enhanced_monitoring - 启用增强监控以获取详细的 CloudWatch 指标
• disable_enhanced_monitoring - 禁用增强监控

加密

• start_stream_encryption - 使用 AWS KMS 启用服务器端加密
• stop_stream_encryption - 禁用服务器端加密

标签

• add_tags_to_stream - 添加或更新数据流的标签
• remove_tags_from_stream - 从数据流中移除标签
• list_tags_for_stream - 列出数据流的标签

存储期

• increase_stream_retention_period - 增加存储期（24-8760 小时）
• decrease_stream_retention_period - 减少存储期（24-8760 小时）

操作指南

与 AWS Kinesis 数据流交互的官方 MCP 服务器提供了一整套工具，用于管理和处理实时数据。

要使用这些工具，请确保您已配置了具有适当权限的 AWS 凭据，以便进行 Kinesis 操作。服务器将自动使用来自环境变量（AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN）或其他标准 AWS 凭据来源的凭据。

所有工具都支持可选的 region_name 参数，用于指定操作所在的 AWS 区域。如果没有提供，则会使用 AWS_REGION 环境变量或默认为 'us-west-2'。

关键概念

数据流模式：

• PROVISIONED：您指定分片的数量，并按预置容量付费
• ON_DEMAND：自动扩展模式，AWS 根据吞吐量管理容量

数据编码：

• 记录可以包含字符串或二进制数据
• 服务器会自动处理 UTF-8 编码和 base64 编码/解码
• 分区键决定记录写入哪个分片

增强型扇出：

• 注册消费者以获得专用吞吐量（每个消费者每分片 2 MB/s）
• 适用于需要低延迟和高吞吐量的应用程序

先决条件

1. 从 Astral 或 GitHub README 安装 uv
2. 使用 uv python install 3.10 安装 Python
3. 设置具有访问 Kinesis 数据流权限的 AWS 凭据
   • 如果不想让 LLM 修改任何资源，考虑设置只读权限

安装

将 MCP 添加到您喜欢的代理工具中：

{
  "mcpServers": {
    "awslabs.kinesis-mcp-server": {
      "command": "uvx",
      "args": ["awslabs.kinesis-mcp-server@latest"],
      "env": {
        "KINESIS-MCP-READONLY": "true",
        "AWS_PROFILE": "default",
        "AWS_REGION": "us-west-2",
        "FASTMCP_LOG_LEVEL": "ERROR"
      },
      "disabled": false,
      "autoApprove": []
    }
  }
}

环境变量

• KINESIS-MCP-READONLY：设置为 "true" 以防止任何修改操作
• AWS_PROFILE：用于身份验证的 AWS 配置文件
• AWS_REGION：使用的 AWS 区域（默认为 us-west-2）
• FASTMCP_LOG_LEVEL：日志级别（ERROR, INFO, DEBUG）

示例

创建数据流

# 创建一个具有 2 个分片的预置数据流
create_stream(
    stream_name="my-stream",
    shard_count=2,
    stream_mode_details={"StreamMode": "PROVISIONED"}
)

# 创建一个按需数据流
create_stream(
    stream_name="my-on-demand-stream",
    stream_mode_details={"StreamMode": "ON_DEMAND"}
)

写入记录

# 写入单个记录
put_record(
    stream_name="my-stream",
    data="Hello, Kinesis!",
    partition_key="user123"
)

# 写入多个记录
put_records(
    stream_name="my-stream",
    records=[
        {"Data": "Record 1", "PartitionKey": "key1"},
        {"Data": "Record 2", "PartitionKey": "key2"}
    ]
)

读取记录

# 获取分片迭代器
iterator_response = get_shard_iterator(
    stream_name="my-stream",
    shard_id="shardId-000000000000",
    shard_iterator_type="LATEST"
)

# 获取记录
records_response = get_records(
    shard_iterator=iterator_response["ShardIterator"],
    limit=100
)

许可证

本项目根据 Apache License 2.0 授权 - 详情见 LICENSE 文件。