Kinesis MCP Server

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 -更新数据流的容量模式（PROVIDED/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 Data Streams交互的官方MCP服务器提供了一套全面的工具来管理Kinesis流和处理实时数据。

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

所有工具都支持可选 region_name 参数指定要在哪个AWS区域中操作。如果没有提供，它将使用AWS_region环境变量或默认值“us-west-2”。

关键概念

流模式：

• 提供的：您指定分片的数量并为配置的容量付费
• ON需求：自动扩展模式，AWS根据吞吐量管理容量

数据编码：

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

增强扇出：

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

先决条件

1. 安装 uv 从 星体 或
2. 使用以下命令安装Python uv python install 3.10
3. 设置AWS凭据以访问Kinesis数据流

- 如果不希望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：日志记录级别（错误、信息、调试）

例子

创建流

# Create a provisioned stream with 2 shards
create_stream(
    stream_name="my-stream",
    shard_count=2,
    stream_mode_details={"StreamMode": "PROVISIONED"}
)

# Create an on-demand stream
create_stream(
    stream_name="my-on-demand-stream",
    stream_mode_details={"StreamMode": "ON_DEMAND"}
)

放置记录

# Put a single record
put_record(
    stream_name="my-stream",
    data="Hello, Kinesis!",
    partition_key="user123"
)

# Put multiple records
put_records(
    stream_name="my-stream",
    records=[
        {"Data": "Record 1", "PartitionKey": "key1"},
        {"Data": "Record 2", "PartitionKey": "key2"}
    ]
)

阅读记录

# Get shard iterator
iterator_response = get_shard_iterator(
    stream_name="my-stream",
    shard_id="shardId-000000000000",
    shard_iterator_type="LATEST"
)

# Get records
records_response = get_records(
    shard_iterator=iterator_response["ShardIterator"],
    limit=100
)

许可证

此项目在Apache License 2.0下获得许可-有关详细信息，请参阅License文件。