MSK MCP Server

MSK MCP 服务

一个功能全面的模型上下文协议服务，为 Amazon MSK（Managed Streaming for Apache Kafka）提供管理能力，包括 MirrorMaker2 连接器、自定义域名创建和灾难恢复操作。此服务使大模型能够创建、配置和管理完整的 MSK 基础设施，并支持高级 DNS 管理和灾难恢复切换功能。

注意

此服务需要具备访问 MSK Connect、S3、Route53 及相关 AWS 服务权限的 AWS 凭证。使用此 MCP 服务时请谨慎操作，确保遵循适当的 AWS 安全实践。

此服务提供两个主要模块：

1. MirrorMaker2 模块（mm2） - 提供完整的 MM2 管道编排功能，包括自定义插件创建、连接器配置和状态监控
2. 自定义域名模块（dr） - 支持创建 MSK 自定义域名、集成 Route53，并提供灾难恢复切换功能

自定义域名模块仅支持 PLAINTEXT；MSK MM2 支持多种身份验证方式（IAM、PLAINTEXT、SCRAM-SHA-512），并通过适当的超时和重试机制处理长时间运行的操作。

功能

MSK 自定义域名与灾难恢复（dr 模块）

• 自定义域名创建 - 使用 Route53 为 MSK 集群创建自定义 DNS 域名
• 灾难恢复切换 - 在主 MSK 集群与备用 MSK 集群之间自动执行灾难恢复故障转移
• Route53 集成 - 管理 DNS 记录和 VPC 关联，实现无缝切换
• 集群信息 - 获取详细的 MSK 集群和代理节点信息
• VPC 管理 - 自动发现并关联 VPC，以支持跨区域部署

MirrorMaker2 核心管理

• 完整的 MM2 管道设置 - 编排完整的 MirrorMaker2 复制设置，包含三种连接器类型（Heartbeat、Checkpoint、Source）
• 自定义插件管理 - 创建和管理 MSK Connect 自定义插件，并自动集成 S3
• 身份验证支持 - 为源集群和目标集群配置 IAM、PLAINTEXT 和 SCRAM-SHA-512 身份验证
• 状态监控 - 实时监控连接器和插件状态，并提供自动等待功能

可用工具

MSK 自定义域名与灾难恢复

• get_msk_cluster_info - 获取 MSK 集群的详细信息
• get_msk_broker_info - 获取 MSK 集群的代理节点信息和引导地址
• get_vpc_for_msk_cluster - 查找 MSK 集群网络的 VPC 信息
• get_route53_hosted_zone_info - 获取 Route53 托管区域的详细信息
• create_msk_custom_domain - 集成 Route53，为 MSK 集群创建自定义 DNS 域名
• list_route53_resource_record_sets - 列出 Route53 托管区域中的所有 DNS 记录
• perform_dr_switch - 在主集群和灾难恢复集群之间执行灾难恢复切换

插件管理

• create_empty_plugin_zip - 创建空的 MirrorMaker2 插件 ZIP 文件并上传到 S3
• create_custom_plugin - 根据存储在 S3 中的 ZIP 文件创建 MSK Connect 自定义插件
• check_plugin_status - 检查自定义插件的当前状态
• wait_for_plugin_ready - 等待插件进入 ACTIVE 状态，并处理超时情况

连接器管理

• create_mirror_heartbeat_connector - 创建用于集群监控的 MirrorHeartbeatConnector
• create_mirror_checkpoint_connector - 创建用于偏移量跟踪的 MirrorCheckpointConnector
• create_mirror_source_connector - 创建用于数据复制的 MirrorSourceConnector
• check_connector_status - 监控连接器的运行状态
• wait_for_connector_ready - 等待连接器进入 RUNNING 状态

Worker 配置

• create_worker_configuration - 创建 MSK Connect Worker 配置
• describe_worker_configuration - 获取 Worker 配置的详细信息
• delete_worker_configuration - 删除 Worker 配置

高级操作

• create_complete_mm2_setup - 端到端 MirrorMaker2 设置编排
• update_connector - 更新现有连接器配置
• delete_connector - 安全移除连接器
• list_connectors - 列出所有 MSK Connect 连接器
• describe_connector - 连接器详细信息

可用资源

• msk://clusters - 列出所有 MSK 集群及其详细信息
• msk://clusters/{cluster_name} - 特定 MSK 集群的详细信息
• msk://route53/zones - 列出所有 Route53 托管区
• msk://route53/zones/{zone_name}/records - 特定托管区的 DNS 记录
• msk://connectors - 列出所有 MSK Connect 连接器
• msk://plugins - 列出所有自定义插件
• msk://worker-configs - 列出所有工作线程配置
• msk://connector-status/{connector_name} - 连接器实时状态信息

可用提示词

• setup_msk_custom_domain - 使用 Route53 为 MSK 集群创建自定义域名的指南
• configure_msk_disaster_recovery - 在主集群和次集群之间设置 DR 的分步指南
• perform_msk_dr_failover - 执行灾难恢复故障转移操作的说明
• setup_complete_mm2_replication - 在集群之间设置完整 MirrorMaker2 复制的指南
• troubleshoot_connector - 连接器问题排查指南
• optimize_mm2_performance - 针对不同规模的性能优化建议

先决条件

1. AWS 账户设置

   • 具有适当权限的 AWS 账户
   • 已正确配置的 MSK 集群（源集群和目标集群）
   • 已为 MSK Connect 配置的 VPC 子网和安全组
   • MSK Connect 的 IAM 服务执行角色

2. 所需的 AWS 权限

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "kafkaconnect:*",
           "kafka:*",
           "s3:GetObject",
           "s3:PutObject",
           "s3:ListBucket",
           "route53:*",
           "ec2:DescribeSubnets",
           "ec2:DescribeVpcs"
         ],
         "Resource": "*"
       }
     ]
   }
   

3. S3 存储桶

   • 用于存储 MirrorMaker2 插件文件的 S3 存储桶
   • MSK Connect 访问存储桶所需的适当 IAM 权限

4. Route53 托管区 （用于自定义域名功能）

   • 用于创建自定义域名的 Route53 托管区
   • 管理 Route53 记录所需的适当 IAM 权限
   • 已为私有托管区配置 VPC 关联

5. 网络配置

   • 源 MSK 集群和目标 MSK 集群均可访问的 VPC 子网
   • 允许 MSK Connect 与集群通信的安全组

安装

使用 uv（推荐）

使用 uv 时无需特定安装。我们将使用 uvx 直接运行 msk-mcp-server。

使用 PIP

或者，也可以通过 pip 安装 msk-mcp-server：

pip install msk-mcp-server

安装后，可以将其作为脚本运行：

python -m mcp_server_msk_mm2

配置

为 Claude.app 配置

添加到 Claude 设置中：

使用最新版本的 uvx

{
  "mcpServers": {
    "msk-mcp-server": {
      "command": "uvx",
      "args": ["msk-mcp-server@latest"],
      "env": {
        "AWS_REGION": "us-east-1",
        "AWS_PROFILE": "your-aws-profile",
        "FASTMCP_LOG_LEVEL": "ERROR"
      },
      "disabled": false,
      "autoApprove": []
    }
  }
}

使用 uvx（简单方式）

{
  "mcpServers": {
    "msk-mm2": {
      "command": "uvx",
      "args": ["msk-mcp-server"],
      "env": {
        "AWS_REGION": "us-east-1",
        "AWS_PROFILE": "your-aws-profile"
      }
    }
  }
}

使用 pip 安装

{
  "mcpServers": {
    "msk-mm2": {
      "command": "python",
      "args": ["-m", "mcp_server_msk_mm2"],
      "env": {
        "AWS_REGION": "us-east-1",
        "AWS_PROFILE": "your-aws-profile"
      }
    }
  }
}

为 VS Code 配置

如需手动安装，请将以下 JSON 代码块添加到 VS Code 的用户设置（JSON）文件中。按 Ctrl + Shift + P 并输入 Preferences: Open User Settings (JSON) 即可打开该文件。

也可以将其添加到工作区中的 .vscode/mcp.json 文件中。这样便可与他人共享配置。

请注意，使用 mcp.json 文件时需要 mcp 键。

使用最新版本的 uvx

{
  "mcp": {
    "servers": {
      "msk-mcp-server": {
        "command": "uvx",
        "args": ["msk-mcp-server@latest"],
        "env": {
          "AWS_REGION": "us-east-1",
          "AWS_PROFILE": "your-aws-profile",
          "FASTMCP_LOG_LEVEL": "ERROR"
        },
        "disabled": false,
        "autoApprove": []
      }
    }
  }
}

使用 uvx（简单方式）

{
  "mcp": {
    "servers": {
      "msk-mm2": {
        "command": "uvx",
        "args": ["msk-mcp-server"],
        "env": {
          "AWS_REGION": "us-east-1",
          "AWS_PROFILE": "your-aws-profile"
        }
      }
    }
  }
}

环境变量

AWS 配置

• AWS_REGION - MSK Connect 操作使用的 AWS 区域（默认：使用 AWS SDK 默认值）
• AWS_PROFILE - 用于身份验证的 AWS 配置文件
• AWS_ACCESS_KEY_ID - AWS 访问密钥（未使用配置文件时）
• AWS_SECRET_ACCESS_KEY - AWS 秘密访问密钥（未使用配置文件时）

超时配置

• AWS_READ_TIMEOUT - AWS 操作的读取超时时间，单位为秒（默认：300）
• AWS_CONNECT_TIMEOUT - AWS 操作的连接超时时间，单位为秒（默认：60）

使用示例

MSK 自定义域名创建

为您的 MSK 集群创建自定义 DNS 域名：

{
  "tool": "create_msk_custom_domain",
  "arguments": {
    "region": "us-east-1",
    "cluster_name": "my-msk-cluster", 
    "zone_name": "kafka.example.com.",
    "vpc_id": "vpc-12345678"
  }
}

灾难恢复切换

在主集群和辅助集群之间执行灾难恢复故障转移：

{
  "tool": "perform_dr_switch",
  "arguments": {
    "primary_region": "us-east-1",
    "primary_cluster_name": "primary-cluster",
    "dr_region": "us-west-2", 
    "dr_cluster_name": "dr-cluster",
    "zone_name": "kafka.example.com.",
    "activate_dr": true
  }
}

获取 MSK 集群信息

检索集群和代理的详细信息：

{
  "tool": "get_msk_cluster_info",
  "arguments": {
    "region": "us-east-1",
    "cluster_name": "my-msk-cluster"
  }
}

完整的 MirrorMaker2 设置

该服务提供用于完整设置的高级编排工具：

{
  "tool": "create_complete_mm2_setup",
  "arguments": {
    "plugin_name": "my-mm2-plugin",
    "s3_bucket": "my-kafka-plugins",
    "s3_key": "mm2/mirrormaker2.zip",
    "source_bootstrap_servers": "source-cluster.kafka.us-east-1.amazonaws.com:9092",
    "target_bootstrap_servers": "target-cluster.kafka.us-east-1.amazonaws.com:9092",
    "service_execution_role_arn": "arn:aws:iam::123456789012:role/MSKConnectRole",
    "vpc_subnets": ["subnet-12345", "subnet-67890"],
    "vpc_security_groups": ["sg-abcdef"],
    "source_auth_type": "iam",
    "target_auth_type": "iam"
  }
}

监控连接器状态

{
  "tool": "check_connector_status",
  "arguments": {
    "connector_arn": "arn:aws:kafkaconnect:us-east-1:123456789012:connector/my-connector/abc-123"
  }
}

身份验证方法

IAM 身份验证

{
  "source_auth_type": "iam",
  "source_role_arn": "arn:aws:iam::123456789012:role/MSKSourceRole"
}

SCRAM-SHA-512 身份验证

{
  "source_auth_type": "scram-sha-512",
  "source_username": "kafka-user",
  "source_password": "kafka-password"
}

PLAINTEXT 身份验证

{
  "source_auth_type": "plaintext",
  "source_username": "kafka-user",
  "source_password": "kafka-password"
}

调试

您可以使用 MCP inspector 调试该服务：

对于 uvx 安装：

npx @modelcontextprotocol/inspector uvx msk-mcp-server

或者，如果您已将软件包安装在特定目录中，或正在进行开发：

cd path/to/msk-mcp-server
npx @modelcontextprotocol/inspector python -m mcp_server_msk_mm2

架构

该服务采用模块化架构构建：

核心框架：

• FastMCP 框架 - 采用依赖项管理的现代 MCP 服务实现
• 模块化设计 - 为不同的 MSK 管理功能提供独立模块
• 统一接口 - 使用单个服务实例挂载子模块

模块结构：

• mm2 模块 - 完整的 MirrorMaker2 连接器管理
• dr 模块 - 自定义域名创建和灾难恢复操作

技术特性：

• 同步操作 - 所有操作均为同步操作，以提高可靠性
• 全面的错误处理 - 提供包含 AWS 错误代码的详细错误消息
• 状态监控 - 内置长时间运行操作的轮询机制
• 重试逻辑 - 针对瞬时故障采用指数退避策略
• 资源验证 - 参数验证和状态检查

参与贡献

我们欢迎您参与贡献，帮助扩展和改进 msk-mcp-server。无论您想添加新工具、增强现有功能，还是改进文档，您的贡献都很有价值。

如需查看其他 MCP 服务和实现模式的示例，请参阅： https://github.com/modelcontextprotocol/servers

欢迎提交拉取请求！欢迎贡献新想法、错误修复或功能增强，让 msk-mcp-server 更加强大实用。

许可证

msk-mcp-server 基于 MIT License 许可。这意味着您可以自由使用、修改和分发该软件，但须遵守 MIT License 的条款和条件。更多详情请参阅项目仓库中的 LICENSE 文件。