AI导哈

Kafka管理接口

OCI Kafka MCP Server是一个基于模型上下文协议（MCP）的AI原生控制接口，用于安全管理Apache Kafka集群，提供结构化工具执行、安全护栏和审计日志功能。

  • by BhaumikAbhishek
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 BhaumikAbhishek/oci-kafka-mcp-server

使用说明

OCI Kafka MCP服务器

AI原生控制界面 使用Apache Kafka进行OCI流式传输，建立在 模型上下文协议（MCP） 规范。

此MCP服务器使LLM代理（Claude、GPT等）能够通过结构化工具执行安全地管理Kafka集群，并具有内置的安全防护、审计日志和企业级安全性。

特性

  • 42种结构化工具 用于集群、主题、消费者、可观察性、人工智能诊断、OCI元数据、集群生命周期、集群配置和工作请求操作
  • 默认情况下为只读 --编写工具需要明确 --allow-writes 旗帜
  • 政策保护 --每个工具都进行了风险分类（低/中/高）；破坏性操作需要确认
  • AI诊断工具 --编排多个Kafka操作以生成缩放建议和滞后根本原因分析
  • 断路器 --防止Kafka不可用时发生级联故障
  • 结构化审计日志记录 --每个工具执行都以JSON格式记录，并带有时间戳、输入哈希和持续时间
  • SASL/SCRAM-SHA-512+TLS --企业安全从第一天开始
  • 专用网络 --专为OCI专用端点设计

快速开始

先决条件

安装

git clone <repo-url>
cd oci-kafka-mcp-server
uv sync

使用本地Kafka运行（开发，Podman）

# Start a local Kafka broker
podman compose -f docker/docker-compose.yaml up -d

# Run the MCP server (read-only mode)
uv run oci-kafka-mcp

# Run with write tools enabled
uv run oci-kafka-mcp --allow-writes

# Stop local Kafka
podman compose -f docker/docker-compose.yaml down

配置OCI流

您可以通过以下任一方式配置OCI Kafka：

  1. 预先设置环境变量 （可选）
  2. 不设置变量 并让MCP服务器在运行时请求所需的值，然后调用 oci_kafka_configure_connection

如果要预先配置环境变量：

export KAFKA_BOOTSTRAP_SERVERS="bootstrap-clstr-XXXXX.kafka.us-chicago-1.oci.oraclecloud.com:9092"
export KAFKA_SECURITY_PROTOCOL="SASL_SSL"
export KAFKA_SASL_MECHANISM="SCRAM-SHA-512"
export KAFKA_SASL_USERNAME="your-username"
export KAFKA_SASL_PASSWORD="your-password"
export KAFKA_SSL_CA_LOCATION="/path/to/ca.pem"

uv run oci-kafka-mcp

或者使用OCI模板文件：

cp .env.oci.example .env.oci
# edit .env.oci with your cluster values
source .env.oci
uv run oci-kafka-mcp

注： KAFKA_* 变量是 非强制性 在服务器启动时。如果未设置，工具将指导代理/用户提供连接详细信息和使用 oci_kafka_configure_connection 在数据平面操作之前。

与MCP客户端一起使用

此服务器可与任何兼容MCP的客户端配合使用。Oracle建议 克莱恩, 光标，以及 MCP主机。请参阅 Oracle MCP客户端配置指南 了解详情。

这 env 下面的块是可选的——如果省略，服务器将提示代理调用 oci_kafka_configure_connection 在运行时显示集群详细信息。

Cline（VS代码扩展）

添加到您的Cline MCP设置中：

{
  "mcpServers": {
    "oci-kafka": {
      "type": "stdio",
      "command": "/path/to/oci-kafka-mcp-server/.venv/bin/oci-kafka-mcp",
      "args": ["--allow-writes"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-bootstrap:9092",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "SCRAM-SHA-512",
        "KAFKA_SASL_USERNAME": "your-username",
        "KAFKA_SASL_PASSWORD": "your-password"
      }
    }
  }
}

光标

增添 .cursor/mcp.json （项目层面）或 ~/.cursor/mcp.json （全球）：

{
  "mcpServers": {
    "oci-kafka": {
      "type": "stdio",
      "command": "/path/to/oci-kafka-mcp-server/.venv/bin/oci-kafka-mcp",
      "args": ["--allow-writes"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-bootstrap:9092",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "SCRAM-SHA-512",
        "KAFKA_SASL_USERNAME": "your-username",
        "KAFKA_SASL_PASSWORD": "your-password"
      }
    }
  }
}

MCP主机

添加到MCPHost配置文件中（例如。， ~/.mcphost.json):

{
  "mcpServers": {
    "oci-kafka": {
      "type": "stdio",
      "command": "/path/to/oci-kafka-mcp-server/.venv/bin/oci-kafka-mcp",
      "args": ["--allow-writes"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-bootstrap:9092",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "SCRAM-SHA-512",
        "KAFKA_SASL_USERNAME": "your-username",
        "KAFKA_SASL_PASSWORD": "your-password"
      }
    }
  }
}

然后使用以下命令启动MCPHost：

mcphost -m ollama:<model> --config ~/.mcphost.json

可用工具（42）

连接管理

工具 描述 风险
oci_kafka_configure_connection 在运行时设置或更新Kafka集群连接详细信息（无需重新启动） 低
oci_kafka_get_connection_info 使用掩码密码显示当前连接配置 LOW

集群操作

工具 描述 风险
oci_kafka_get_cluster_health 代理状态、控制器ID、主题计数 低
oci_kafka_get_cluster_config 代理级Kafka配置设置 低

主题操作

工具 描述 风险
oci_kafka_list_topics 列出所有主题 低
oci_kafka_describe_topic 分区详细信息、领导者、副本、ISR、主题配置 低
oci_kafka_create_topic 使用分区和复制因子创建主题 中等
oci_kafka_update_topic_config 更新主题配置（保留、压缩等） 中等
oci_kafka_delete_topic 删除主题（需要确认） 高

消费者操作

工具 描述 风险
oci_kafka_list_consumer_groups 列出所有消费者群体 低
oci_kafka_describe_consumer_group 组状态、成员、协调员、分区分配 低
oci_kafka_get_consumer_lag 每个分区延迟、已提交偏移量、结束偏移量 低
oci_kafka_reset_consumer_offset 将偏移重置为最早/最新/特定偏移（需要确认） 高
oci_kafka_delete_consumer_group 删除消费者组（需要确认） 高

可观测性

工具 描述 风险
oci_kafka_get_partition_skew 检测代理之间的分区领导者不平衡 低
oci_kafka_detect_under_replicated_partitions 查找ISR计数<副本计数 低的分区

AI诊断

工具 描述 风险
oci_kafka_recommend_scaling 将健康、偏斜和复制数据编排为扩展建议 低
oci_kafka_analyze_lag_root_cause 将消费者状态、滞后和拓扑与根本原因分析相关联 低

OCI控制平面元数据

工具 描述 风险
oci_kafka_list_oci_clusters 列出OCI隔间（自动发现隔间）中的所有Kafka集群 低
oci_kafka_get_oci_cluster_info 集群OCID、生命周期状态、代理形状、引导URL、标签 低

集群生命周期（OCI控制平面）

异步操作——返回工作请求OCID；使用 oci_kafka_get_work_request 投票以完成。

工具 描述 风险
oci_kafka_create_cluster 提供新的OCI Kafka集群（需要确认） 高
oci_kafka_update_cluster 更新群集显示名称、标记或应用的配置 中等
oci_kafka_scale_cluster 扩展现有集群的代理数量（需要确认） 高
oci_kafka_delete_cluster 永久删除集群及其所有数据（需要确认） 高
oci_kafka_change_cluster_compartment 将集群移动到不同的OCI隔间（需要确认） 高
oci_kafka_enable_superuser 授予群集超级用户完全管理权限 中等
oci_kafka_disable_superuser 撤销超级用户访问权限以恢复最低权限 中等

集群配置（OCI控制平面）

命名的、版本化的Kafka代理设置集，可以应用于一个或多个集群。

工具 描述 风险
oci_kafka_list_cluster_configs 列出一个隔间中的所有集群配置 低
oci_kafka_get_oci_cluster_config 获取集群配置及其最新版本 LOW
oci_kafka_create_cluster_config 创建新的命名群集配置 MEDIUM
oci_kafka_update_cluster_config 更新配置的显示名称或标签 中等
oci_kafka_delete_cluster_config 删除配置及其所有版本（需要确认） 高
oci_kafka_change_cluster_config_compartment 将配置移动到其他隔间 中等
oci_kafka_list_cluster_config_versions 列出集群配置的所有版本 低
oci_kafka_get_cluster_config_version 获取集群配置的特定版本 低
oci_kafka_delete_cluster_config_version 删除特定配置版本 MEDIUM

工作请求和节点形状（OCI控制平面）

跟踪集群生命周期和配置工具返回的异步OCI操作。

工具 描述 风险
oci_kafka_get_work_request 异步OCI操作的轮询状态和进度 LOW
oci_kafka_list_work_requests 按隔间或资源OCID LOW列出工作请求
oci_kafka_get_work_request_errors 从失败的工作请求中获取错误详细信息 低
oci_kafka_get_work_request_logs 从工作请求中获取带时间戳的日志条目 低
oci_kafka_cancel_work_request 取消正在进行的工作请求 中等
oci_kafka_list_node_shapes 列出用于集群配置的可用代理节点形状 低

安全模型

风险等级 行为 示例
低 始终允许 健康检查，列出/描述操作
中等 需要 --allow-writes 创建主题，更新配置
高 需要 --allow-writes +确认 删除主题、重置偏移量、集群生命周期

发展

# Run tests (92 tests, all unit — no Kafka broker needed)
uv run pytest

# Run tests with coverage
uv run pytest --cov=oci_kafka_mcp --cov-report=term-missing

# Lint
uv run ruff check src/ tests/

# Format
uv run ruff format src/ tests/

# Type check
uv run mypy src/

建筑

看 docs/ARCHITECTURE.md 完整的安全架构文档，包括威胁模型、依赖性审计和部署架构。

许可证

阿帕奇-2.0

相关 MCP