Kafka MCP

kafka-mcp

一个 MCP 服务，将 Kafka 调试功能作为大模型可调用的工具提供。默认通过 stdio 使用 MCP，也可选择提供可流式传输的 HTTP 服务，并通过 franz-go 与 Kafka 通信。

一个进程可以连接多个 Kafka 集群。在 stdio 模式下，会为会话选择一个端点。在 HTTP 模式下， 每个端点都有自己的路径，因此会话通过连接方式绑定到一个集群，而不是依赖调用方可能忘记发送的参数。

安装

每个版本都会以四种形式提供同一个服务：

• 二进制文件：可在发布页面 获取 kafka-mcp_<Os>_<arch> 归档文件。 将 kafka-mcp 放入 PATH。

• MCP Bundle：同一页面提供 kafka-mcp_<version>_<os>_<arch>.mcpb， 适用于 Apple Silicon Mac、Linux（amd64、arm64）和 Windows（amd64）。在支持安装 Bundle 的 客户端中打开，例如 Claude Desktop。它会要求你提供配置文件；如果文件中有多个端点，还会询问要提供哪个端点。 Intel Mac 请使用二进制文件或镜像。

• 容器镜像：ghcr.io/denizgursoy/kafka-mcp:<tag>。启动时使用 --server；对于 stdio，请挂载文件并传入 --server=false：

  docker run -i --rm \
    --mount type=bind,src=$PWD/kafka-mcp.yaml,dst=/config/kafka-mcp.yaml,readonly \
    -e CONFIG_FILE=/config/kafka-mcp.yaml \
    ghcr.io/denizgursoy/kafka-mcp:latest --server=false --endpoint local
  

• MCP Registry：在官方注册表 中列为 io.github.denizgursoy/kafka-mcp，并提供镜像和 Bundle，因此读取该注册表的客户端可以直接从中安装。

有关如何将版本发布到注册表和其他目录，请参阅 PUBLISHING.md。

配置

工作目录、~/.config/kafka-mcp/ 或 /etc 中的 kafka-mcp.{toml,yaml,yml,json} 文件用于配置服务。只有在使用 --server 时才会使用 http 块。

http:
  address: ":8090"
  base_path: /kafka-mcp # optional; prefixes every endpoint and /healthz
  # public_url: https://kafka.example.com # optional; the address download links use
clusters:
  prod:
    brokers:
      - kafka-1:9093
      - kafka-2:9093
    security:
      tls:
        enabled: true
        ca_file: /etc/kafka/ca.pem
        # cert_file: /etc/kafka/client.pem # optional mTLS; requires key_file
        # key_file: /run/secrets/client.key
      sasl:
        - scram:
            enabled: true
            algorithm: SCRAM-SHA-256 # or SCRAM-SHA-512
            user: kafka-mcp-readonly
            pass: "{env:KAFKA_PASSWORD}" # or password_file: /run/secrets/kafka
  preprod:
    brokers: kafka-preprod:9093
endpoints:
  prod-read:
    cluster: prod
    path: /mcp
    description: Production investigation and debugging
    read_only: true
  prod-write:
    cluster: prod
    path: /mcp/rw
    description: Approved production changes
    tools:
      copy_message: false
  preprod:
    cluster: preprod
    path: /mcp/preprod

clusters 管理 Kafka 连接详情：代理、TLS 和 SASL。endpoints 管理策略，以及 HTTP 模式下的 MCP 路由。stdio 按名称选择端点，不使用其路径。多个端点可以引用同一个集群，因此示例在 /kafka-mcp/mcp 以只读模式、在 /kafka-mcp/mcp/rw 以可写模式复用同一个生产连接。路径必须精确匹配：/mcp 不会匹配 /mcp/rw。description 为可选项，server_config 会报告该字段，以便调用方了解端点的用途。

http.base_path 会为端点路径和存活探测路由添加前缀。端点路径开头或结尾的斜杠均可省略。路径必须唯一，且不能是 /、/healthz，也不能包含查询或片段。省略 path 时，默认值为 /mcp/<endpoint-name>。

为兼容旧配置，未包含 endpoints 块的文件仍会为每个集群创建一个位于 /mcp/<cluster-name> 的端点。在迁移期间，现有的集群级 read_only 和 tools 值仍会作为限制下限；显式配置的端点不能放宽这些限制。新的配置应将这两个字段放在 endpoints 下。

端点可以按名称关闭单个工具：

endpoints:
  prod-read:
    cluster: prod
    path: /mcp
    read_only: true
    tools:
      create_topic: false
      commit_offset: false

跨集群写入权限按端点授予。copy_message、 produce_message 和 transform_messages 可以指定 destination_cluster，且该集群必须列在调用端点的 destinations 中：

endpoints:
  prod-read:
    cluster: prod
    read_only: true
    destinations: [preprod]   # this session may copy into preprod, nowhere else

没有 destinations 的端点不会写入其他集群。列出的集群若配置为 read_only，仍会拒绝写入；列出端点自身的集群则会报错，因为是否允许写入由 read_only 控制。名称会在启动时检查。没有 endpoints 块的文件会保留原有行为：每个集群的端点都会列出其他所有集群。没有任何端点提供服务的集群不会显示在 list_clusters、compare_clusters 和 destination_cluster 中。

映射中未提及的工具仍会公开，因此文件只声明要隐藏的内容，而不必重新列出每个工具，以免之后添加的工具悄无声息地丢失。名称必须与 server_config 中列出的名称完全一致且为小写。名称会在启动时检查：如果名称不是工具，服务器将停止运行，因为拼写错误会导致原本要隐藏的工具仍然公开。不能关闭 server_config，因为会话需要通过它获知连接到哪个集群和策略，以及该端点有哪些工具。tools 映射只能缩小端点的工具范围：无论映射如何设置，read_only: true 仍会隐藏写入工具。

最简本地文件：

clusters:
  local:
    brokers: localhost:19092
    schema_registry:
      urls: ["http://localhost:18081"]
endpoints:
  local:
    cluster: local
    path: /mcp/local

如需通过 stdio 使用自定义配置运行，请设置 CONFIG_FILE=/path/to/config.yaml go run ./cmd/server。如果配置定义了多个端点，请使用 --endpoint <name> 选择一个。添加 --server 可改为通过 HTTP 提供所有已配置端点；HTTP 模式下不使用 --endpoint。未设置 CONFIG_FILE 时，chu 会依次在工作目录、操作系统的用户配置目录（Linux 上为 ~/.config/kafka-mcp/）和 /etc 中查找 kafka-mcp.{toml,yaml,yml,json}。它使用第一个匹配的文件，而不会合并多个文件。标准加载顺序为默认值、文件、HTTP，然后是环境变量；环境变量覆盖使用 KAFKA_MCP_ 前缀（例如，KAFKA_MCP_HTTP_ADDRESS=:9000 或 KAFKA_MCP_HTTP_BASE_PATH=/kafka-mcp）。可通过 LOG_LEVEL 和 LOG_PRETTY 配置日志记录。

至少需要一个包含代理的集群。http.address 默认为 :8090，http.base_path 默认为 HTTP 根路径。

服务器从不存储文件。search_messages 和 transform_messages 配合 download 使用，以及 produce_message 配合 upload 使用时，都会处理用户自己机器上的文件。请参阅用户机器上的文件。 通过 HTTP 使用时，请在同一地址后面的每个实例上将 http.link_key（或 http.auth_token）设为相同的值。链接会基于每个请求实际访问的地址生成，并遵循 X-Forwarded-Proto 和 X-Forwarded-Host；如果代理隐藏了真实地址，请设置 http.public_url。启动时会拒绝旧的 output_dir 设置。

用户机器上的文件

用于保存或发送大量消息的工具会处理用户自己机器上的文件，而不是服务器上的文件。

通过 stdio 使用时，服务器是 Agent 在那台机器上启动的进程，因此 download 会保存文件，upload 会立即读取文件；路径相对于进程的工作目录（通常是项目目录），也可以使用绝对路径。响应会在 download.path 或 upload.path 中标明文件名；下载时会先写入隐藏的临时文件，然后再命名，并且绝不会覆盖已有文件。通过 stdio 运行此镜像需要挂载目录，例如 docker run -i --rm -v "$PWD:/work" -w /work …。

通过 HTTP 使用时，工具无法自行写入文件：其结果会返回给模型，而不是写入磁盘。因此，工具会返回链接和 curl 命令，由 Agent 在用户的机器上运行该命令：

curl -fsS -o 'orders.csv.part' 'https://kafka.example.com/mcp/prod/files/Q2x…' && mv 'orders.csv.part' 'orders.csv'
curl -fsS -T 'orders-fixed.jsonl' 'https://kafka.example.com/mcp/prod/files/Zk9…'

不会提前准备任何内容，也不会存储任何内容：命令运行时，数据会在那台机器与 Kafka 之间直接流动，因此消息正文也不会经过模型。 链接本身携带请求，并使用从 http.link_key 派生的密钥通过 AES-GCM 加密。如果未设置 http.link_key，则使用 http.auth_token。服务器在发出链接后不会保留任何状态，因此只要共享同一个密钥，负载均衡器后的任何实例都能处理由其他实例生成的链接。如果这两个设置都未配置，每个进程都会生成随机密钥，链接只能在生成它的进程上打开。加密可确保链接不会在代理日志中泄露脚本内容，也无法被修改或伪造。使用链接时，请求还会再次通过该实例的策略检查：如果集群已切换为只读，则会拒绝此前生成的上传请求。

链接会在 10 分钟后过期。在此之前，请将其视为机密信息：链接无需 Bearer 令牌，因为 Agent 的 curl 不会携带 MCP 客户端的标头；并且在过期前可以重复使用。重复运行上传链接会再次发送文件。如果在传输任何数据前发生错误（例如主题已删除），服务器会返回包含原因的 HTTP 错误，curl -f 会报告该错误；如果在传输第一个字节后发生错误，传输会中断，因此 curl -f 会失败，而不是保留不完整的文件。链接由端点下的 <endpoint path>/files/ 提供，并像端点一样检查来源。如果请求太大，无法放入链接（加密后约为 6 KiB，例如脚本过长），生成链接时会拒绝该请求。

brokers 可以是一个地址标量，也可以是包含多个地址的 YAML 列表。每个地址都会作为单独的种子代理传递给 Kafka。

使用 {env:VAR} 或 password_file 将机密信息保留在文件之外。chu 会忽略未知字段。

TLS 使用 twmb/tlscfg 配置，最低版本为 TLS 1.2，并采用推荐的密码套件。TLS 使用系统信任存储；ca_file 可添加自定义 CA。对于 mTLS，请同时提供 cert_file 和 key_file。security.sasl 是一个按优先级排序的列表：每个条目启用 scram 或 plain，也可以按下方所示使用 oauth。对于 PLAIN，请使用 plain: {enabled: true, user: alice, pass: "{env:KAFKA_PASSWORD}"}。 两者都可选用 zid（授权身份），也可使用 password_file 代替 pass；SCRAM 还接受用于委派令牌的 is_token: true。算法名称不区分大小写。禁用的条目会被忽略；重复的机制以及同时启用多个机制的条目会被拒绝。回退机制会协商代理支持的机制，但不会重试无效凭据。 所有 Kafka 连接（包括读取消息的会话）都会使用这些设置。

旧版的每集群 tls 和 sasl: {mechanism, user, password, password_file} 仍受支持，但不能与同一集群上的 security 同时使用。

对于 OAUTHBEARER 客户端凭据，请使用以下集群安全配置块：

security:
  tls:
    enabled: true
  sasl:
    - oauth:
        enabled: true
        token_url: https://identity.example.com/realms/apps/protocol/openid-connect/token
        client_id: kafka-mcp
        client_secret: "{env:KAFKA_CLIENT_SECRET}"
        scopes: [kafka]
        timeout: 10s
        # zid: optional-authorization-id
        # extensions: {tenant: example}

或者，使用 oauth: {enabled: true, token: "{env:KAFKA_TOKEN}"} 设置静态令牌。只能配置一种模式：静态 token、客户端凭据或 gcp。客户端凭据令牌会在身份验证时获取，在集群的管理连接和读取连接之间缓存，并根据 expires_in 在接近过期时续期。令牌请求使用当前身份验证上下文和超时（默认为 10 秒）。静态令牌不会续期。代理必须支持 OAUTHBEARER 并信任身份提供方。令牌端点错误响应正文不会被报告，只会报告其 HTTP 状态码，因为响应正文可能会回显机密信息。

oauth.proxy 会通过 http、https、socks5 或 socks5h 代理发送令牌请求；如果未设置，则使用 HTTPS_PROXY、HTTP_PROXY 和 NO_PROXY。

代理的 security.tls 设置仅适用于 Kafka 连接。令牌端点使用 oauth.tls，它也可以提供客户端证书（RFC 8705）。使用证书时，client_secret 为可选项：如果留空，client_id 会放入请求正文，客户端则通过证书进行身份验证。

    - oauth:
        enabled: true
        token_url: https://identity.example.com/token
        client_id: kafka-mcp
        scopes: [kafka]
        tls:
          cert_file: /etc/certs/oauth/client-cert.pem
          key_file: /etc/certs/oauth/client-key.pem
          ca_file: /etc/certs/oauth/ca.pem # optional; system roots otherwise

oauth.gcp 使用 X.509 客户端证书获取 Google Cloud 访问令牌（通过 mTLS 实现工作负载身份联合），也可选择模拟服务账号。使用 Google 的 external_account 凭据文件；其中的 credential_source 会被忽略，因为此处改用下方的证书。

    - oauth:
        enabled: true
        gcp:
          enabled: true
          credentials_file: /etc/certs/gcp/wif-credentials.json
          cert_file: /etc/certs/kafka/client-cert.pem
          key_file: /etc/certs/kafka/client-key.pem
          format: managed_kafka # Google Managed Service for Apache Kafka; default raw
          # proxy: http://proxy:3128 # default: oauth.proxy, then the environment
          # lifetime: 1h             # impersonated token lifetime
          # refresh_before: 5m       # refresh this long before expiry
          # scopes: [https://www.googleapis.com/auth/cloud-platform]

可以直接设置 audience、token_url 和 service_account_impersonation_url，并覆盖凭据文件中的设置。managed_kafka 要求模拟服务账号。刷新失败时，会继续使用缓存的令牌，直到令牌实际过期。

在 oauth.tls 和 oauth.gcp 中，可以直接提供 cert、key、ca（gcp 还包括 credentials），而不使用 _file；两种方式不能同时使用。内联密钥和凭据支持 {env:VAR}。每次刷新令牌时都会重新读取文件，因此轮换证书后无需重启即可生效。

insecure_skip_verify: true 会禁用服务器证书和主机名验证。此选项可用于 security.tls（代理）、schema_registry.tls、oauth.tls（令牌端点）和 oauth.gcp（Google STS 和模拟），默认值为 false。此设置会使连接面临中间人攻击；建议优先使用 ca_file，且不要在生产环境中使用此设置。

启动时无法访问的集群仍会提供服务，list_clusters 会将其报告为已断开。一个集群宕机不应阻碍对其他集群进行调试。

消息格式

对于会显示或搜索消息的所有工具，消息都会按以下规则中第一个适用的规则解码为 JSON：

1. 主题在 topic_formats 中指定了格式。
2. 字节包含 Schema Registry 标头（Confluent 线格式），且集群配置了 schema_registry：Avro、Protobuf（含消息索引和引用的架构）或 JSON Schema。
3. 否则依次尝试 JSON、UTF-8 文本和 base64。

clusters:
  prod:
    brokers: kafka-1:9093
    schema_registry:
      urls: ["https://schema-registry:8081"]
      user: kafka-mcp                     # or bearer_token: "{env:SR_TOKEN}"
      password: "{env:SR_PASSWORD}"       # or password_file
      # tls: {enabled: true, ca_file: /etc/sr/ca.pem}
    topic_formats:                        # topics whose messages carry no schema id
      "orders.*":                         # exact name or glob; longest match wins
        value:
          format: protobuf
          proto_files: [shop/order.proto]
          import_paths: [./protos]        # or descriptor_set: ./orders.binpb
          message_type: shop.Order
      events:
        value: {format: avro, schema_file: ./schemas/event.avsc}
      metrics:
        key: {format: text}
        value: {format: msgpack}

格式包括 avro、protobuf、json、msgpack、text 和 binary。架构文件和 .proto 源文件会在启动时加载，因此文件或类型缺失会导致服务启动失败。若消息指定了架构但无法解码，则会以 base64 形式返回，并通过 decode_error 说明原因，绝不会静默处理。JSON 输出的键按排序排列；超过 2^53 的 Avro long 会保留精确值；未设置的 Protobuf 字段会显示其默认值。

每条渲染后的消息都会为值报告 format、schema_id 和 message_type；如果键不是纯文本，还会报告 key_format / key_schema_id。

server_config 会报告端点名称、准确路径、描述、策略及其服务的集群，且绝不会报告密码。其 authentication 和 sasl_user 描述第一个配置的选项；sasl_options 按优先顺序列出所有已配置的身份，而不是单个代理连接协商出的机制。它还会报告 schema_registry URL 和 topic_formats。

身份验证和浏览器客户端

http.auth_token 会让每个 MCP 请求都携带 Authorization: Bearer <token>； 不带该字段的请求会收到 401。它接受 {env:VAR}。/healthz 和 CORS 预检请求无需令牌。若未设置该项，这些端点本身没有身份验证，只有在没有不受信任的来源能访问监听器时才安全。

通过环境变量设置令牌，可让机密完全不出现在配置文件中：

export KAFKA_MCP_HTTP_AUTH_TOKEN="$(openssl rand -hex 32)"
CONFIG_FILE=kafka-mcp.yaml kafka-mcp --server

每个配置键都可以通过这种方式覆盖：使用 KAFKA_MCP_，后接大写的键路径，并用 _ 分隔各层，因此 http.auth_token 对应 KAFKA_MCP_HTTP_AUTH_TOKEN。若要将令牌保留在文件中，但让值从环境变量读取并使用自选名称，请改为引用它：

http:
  auth_token: "{env:MY_MCP_TOKEN}"   # the server refuses to start if MY_MCP_TOKEN is unset

然后，客户端会在每个请求中发送令牌：

curl -X POST http://localhost:8090/mcp/local \
  -H "Authorization: Bearer $KAFKA_MCP_HTTP_AUTH_TOKEN" \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"1"}}}'

在连接客户端中使用的客户端配置里，为每个远程条目添加此标头。{env:...} 是客户端自身的变量替换语法，因此令牌也不会出现在该文件中：

{
  "mcp": {
    "kafka-local": {
      "type": "remote",
      "url": "http://localhost:8090/mcp/local",
      "headers": { "Authorization": "Bearer {env:KAFKA_MCP_HTTP_AUTH_TOKEN}" }
    }
  }
}

在容器中，像传递其他机密一样传递令牌：

docker run -e KAFKA_MCP_HTTP_AUTH_TOKEN="$TOKEN" \
  -e KAFKA_MCP_HTTP_CORS_ALLOW_ORIGINS="https://mcp-client.example" \
  -v "$PWD/kafka-mcp.yaml:/etc/kafka-mcp.yaml" -e CONFIG_FILE=/etc/kafka-mcp.yaml \
  -p 8090:8090 kafka-mcp

命令行客户端不会发送 Origin 标头，因此不受 CORS 影响。带有 Origin 的请求，如果来源不在 allow_origins 中，就会被拒绝并返回 403。此操作会在任何工具运行之前发生，因为 CORS 本身只能阻止页面读取响应，而请求仍会执行。默认不允许任何来源，因此用户访问的网页都无法驱动这些工具。若要使用浏览器客户端，请指定其来源：

http:
  address: ":8090"
  cors:
    allow_origins: ["https://mcp-client.example"]
    allow_methods: [GET, POST, DELETE, OPTIONS]
    allow_headers: [content-type, accept, authorization, cache-control, last-event-id, mcp-session-id, mcp-protocol-version]
    expose_headers: [Mcp-Session-Id]
    allow_private_network: true
    max_age: 600

环境变量中的同一列表以逗号分隔： KAFKA_MCP_HTTP_CORS_ALLOW_ORIGINS="https://a.example,https://b.example"。 将其设为 * 可恢复旧版的允许所有来源行为。

http.cors 是 ada 的 CORS 中间件配置，直接从文件读取，因此该中间件支持的所有选项都可在此使用。每个键都是可选的，并保留各自的默认值，因此仅设置 allow_origins 不会移除其余选项。默认值与上文相同，但不含 allow_origins。 allow_origins: ["*"] 可恢复旧版的允许所有来源行为。

其中有三个选项对 MCP 传输至关重要。allow_methods 需要包含 GET、POST 和 DELETE：请求使用 POST 发送，事件流使用 GET，而客户端使用 DELETE 结束会话。allow_headers 需要包含 mcp-session-id 和 mcp-protocol-version，客户端从第二个请求开始会发送这两个标头；若缺少其中一个标头，整个预检请求都会失败，而不是将其丢弃。Mcp-Session-Id 必须保留在 expose_headers 中：会话 ID 会随 initialize 响应返回，而无法读取该 ID 的页面无法发起第二次调用。

allow_private_network 用于响应 Chrome 的 Private Network Access 预检。公共地址上的页面在访问私有地址或环回地址上的服务之前，必须通过此预检；该选项默认开启，并且只有在预检已通过其余策略检查时才会授予访问权限。若同时设置 allow_credentials 和通配符 allow_origins，中间件会在启动时拒绝该配置，除非还设置了 unsafe_wildcard_origin_with_allow_credentials；但不应设置此选项。

获准的来源可以使用服务的 Kafka 凭据驱动所有工具，因此只列出应拥有此权限的页面，设置 auth_token（浏览器客户端随后必须发送该令牌），并在不应写入的端点上设置 read_only: true。使用 /mcp/rw 这类不明显的路径并不构成身份验证。

在 Go 应用中嵌入这些工具

已持有 Kafka 客户端的应用可以通过 github.com/denizgursoy/kafka-mcp/pkg/kafkamcp 在自己的 MCP 服务上注册这些工具。该包不会加载配置、启动监听器或建立连接。路由、身份验证和传输由宿主应用管理；工具、结果、只读拒绝和审计记录与可执行文件提供的相同。

clusters, err := kafkamcp.NewClusters(
    kafkamcp.Connection{Name: "stage", Kafka: stageClient, NewReader: newStageReader},
    kafkamcp.Connection{Name: "prod", Kafka: prodClient, NewReader: newProdReader, ReadOnly: true},
)
server := mcp.NewServer(&mcp.Implementation{Name: "my-app-stage"}, nil)
err = clusters.Register(server, "stage", kafkamcp.Options{Destinations: []string{"prod"}})
http.Handle("/mcp/stage", mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return server }, nil))

• NewReader 每次调用都必须返回一个新的客户端，直接消费主题（使用 kgo.ConsumePartitions）。这些工具会关闭这些客户端；它们绝不会关闭 Kafka、Admin、ManualProducer、BulkProducer 或 SchemaRegistry。
• 如果宿主应用想要限制或复用读取器，可用 AcquireReader 替代 NewReader：它会借出一个客户端，并提供工具恰好调用一次的 release 函数，而不是关闭客户端；它也可能一直等待空闲槽位，直到调用的上下文结束。借出的客户端应使用 kgo.KeepControlRecords()；标记会被跳过，而如果没有标记，扫描结束于事务标记时，就会等待永远不会到来的数据。
• BulkProducer 是可选的，用于写入 produce_message upload 和 transform_messages 的批次，因此 Kafka 上的单条写入不会排在这些批次后面。
• ProtoMessage 是可选的，用于指定主题的键或值在不使用 Schema Registry 封装时所持有的 protobuf 消息，适用于已自行解析 .proto 文件的宿主应用。带封装的消息仍会先发送到注册表，而且不需要 topic_formats。
• Options.Logger 接收审计记录（默认为 slog.Default），Options.User 则根据请求头标明调用者身份，使用宿主应用的身份验证代理确认的信息。
• 传递给 NewClusters 的每个集群都可供 compare_clusters 和 list_clusters 读取。只有当目标集群位于 Options.Destinations 中时，copy_message、produce_message 和 transform_messages 才会向其写入；无论请求到达哪个服务器，ReadOnly 连接都会拒绝所有写入。
• Options.ReadOnly 会隐藏并拒绝对已注册集群的写入；Options.Tools 可按名称关闭工具，未知名称会导致错误。
• server_config 会将身份验证报告为 host-managed，并报告 Connection.Brokers 和 Connection.Principal。两者都只会被报告，不会用于其他用途。

连接客户端

对于默认的 stdio 传输，请将客户端配置为启动该二进制文件。只有一个端点的配置无需参数：

{
  "mcp": {
    "kafka-local": {
      "type": "local",
      "command": ["kafka-mcp"],
      "environment": {"CONFIG_FILE": "/path/to/kafka-mcp.yaml"}
    }
  }
}

当文件中包含多个端点时，请在命令中添加端点选择项，例如 "command": ["kafka-mcp", "--endpoint", "prod-read"]。如果没有指定端点，进程会拒绝启动，以免将会话意外连接到错误的集群或权限策略。

要使用 HTTP，请启动 kafka-mcp --server，并为每个端点注册一个远程客户端条目：

{
  "mcp": {
    "kafka-local": {
      "type": "remote",
      "url": "http://localhost:8090/mcp/local"
    },
    "kafka-prod-read": {
      "type": "remote",
      "url": "http://localhost:8090/mcp",
      "enabled": false
    }
  }
}

当服务器设置了 http.auth_token 时，请在每个条目中添加 "headers": {"Authorization": "Bearer {env:KAFKA_MCP_HTTP_AUTH_TOKEN}"}；请参阅身份验证和浏览器客户端。

键名会成为工具前缀，因此这些工具会显示为 kafka-local_list_topics 和 kafka-prod-read_list_topics。请根据集群和端点策略为条目命名：前缀能最清楚地表明调用可能访问和更改什么。请使用 server_config 确认，而不要轻信客户端侧的名称。

每个启用的集群都会占用上下文：这些工具的定义约占 8k 个令牌。只启用所需的工具，并让 Agent 负责处理生产环境：

{
  "tools": { "kafka-prod*": false },
  "agent": {
    "kafka-prod": {
      "description": "Debugging against production Kafka.",
      "tools": { "kafka-prod*": true }
    }
  }
}

权限

权限分为三层，只有一层是真正的安全保障：

层级	防范对象	真正的安全保障？
写入操作上的 confirm: true	LLM 根据一个含糊的请求更改内容	否 — 这只是防护措施
endpoint read_only: true	使用服务的 Kafka 身份意外写入	否 — 任何能编辑配置的人都可以将其关闭
SASL 主体上的 Kafka ACL	未授权人员	是 — 由 broker 决定

只读 endpoint 会公开什么

read_only: true 不仅会拒绝写入：endpoint 不会列出唯一用途为写入的工具。只读 endpoint 的 tools/list 中不会出现 add_partitions、alter_topic_config、commit_offset、create_topic、delete_consumer_group、delete_records 和 delete_topic，因此客户端永远不会看到自己无法使用的工具，其预览也不会描述此 endpoint 永远不会执行的更改。

可写 endpoint 也可以通过其 tools map 隐藏单个工具。从客户端看来，这两种机制相同：工具未注册，因此不会出现在 tools/list 和 server_config 中。

copy_message、produce_message 和 transform_messages 仍会保留，因为 read_only 保护的是被写入的集群，而目标由每次调用指定。将消息从只读生产集群复制出来，或使用受保护的会话为可写的预生产集群写入初始数据，正是这些工具的用途。目标必须位于 endpoint 的 destinations 中；如果目标就是只读集群本身，每个工具都会拒绝写入。

隐藏工具决定了会公开哪些工具，而不是哪些操作获准执行：这两个工具在实际修改时仍会拒绝操作，因此注册错误不会导致写入。server_config 会报告 endpoint 实际公开的工具，供会话区分这两种情况。

为两个人设置不同的权限

Kafka 根据 SASL 主体强制实施权限，因此，两个人使用不同凭据运行同一服务时，会拥有不同的权限。添加分区需要主题上的 ALTER 权限：

# ali may read but not reshape topics
rpk acl create --allow-principal User:ali \
  --operation read,describe --topic orders

# deniz may also add partitions
rpk acl create --allow-principal User:deniz \
  --operation read,describe,alter --topic orders

两人各自将 CONFIG_FILE 指向自己的文件，文件间的区别仅在于 security.sasl[0].scram.user 和密码。ali 调用 add_partitions 时，broker 会拒绝：

not authorized to add partitions to "orders": the broker refused this request.
Adding partitions requires ALTER permission on the topic for the principal
this server connects as

ali 无法通过编辑配置或重新构建二进制文件绕过这一限制，因为决策由 Kafka 而非此服务作出。在没有 ACL 的集群上，可用的保护方式是 read_only: true。

审计日志

每次工具调用都会记录日志。每次调用在服务自身的日志流中生成一条 slog 记录，无需额外配置：

{"level":"INFO","msg":"tool call","tool":"produce_message","outcome":"ok",
 "duration_ms":12,"endpoint":"prod-write","cluster":"prod","read_only":false,
 "principal":"kafka-mcp-rw","session":"QY7MZM...","client":"claude-code",
 "client_version":"1.0.0","confirm":true,"item_count":1,"targets":"orders"}

会更改集群的工具——add_partitions、alter_topic_config、 commit_offset、create_topic、copy_message、delete_consumer_group、 delete_records、delete_topic、produce_message、transform_messages——会 以 INFO 级别记录。 其他所有工具都会以 DEBUG 级别记录，因为读取操作频繁且不会更改任何内容， 如果以相同级别记录，就会让写入记录淹没在其中。提高日志级别即可查看这些记录。

两个字段最为关键。confirm 用于区分实际写入和预览，targets 会列出每个 条目所指向的主题、组、分区和偏移量，因此记录会说明具体触及了哪个主题，而 不只是说明触及了某个主题。没有条目的工具会记录其顶层主题和目标位置， produce_message 上传操作会记录 upload。download 操作会记录 download。

绝不会记录消息内容。 produce_message、copy_message 和 transform_messages 会携带 任意载荷，而审计日志通常比其描述的数据有更多人可以读取，因此不会记录键、 值和标头。审计代码没有可供反序列化这些内容的字段，所以即使意外发生， 内容也不会进入日志。

记录的身份信息都不代表某个人，服务也无法编造此类信息。 各字段的实际含义如下：

字段	含义
principal	此服务连接时使用的 Kafka 凭据，ACL 会根据该凭据进行强制执行。所有访问同一端点的人都共用此凭据
endpoint	允许此次调用的端点策略，以及因此被访问的集群
session	一个 MCP 会话，将一系列调用归为同一次调查
client、client_version	建立连接的程序在 initialize 时自行标识的信息。由客户端自行报告，未经验证
user	仅在传入的 bearer 令牌确立用户身份时才会出现。此服务未安装令牌验证器，因此目前不会出现
request_id	HTTP 请求 ID，用于将记录与访问日志关联起来

如需将操作归属到个人，请按上文所述为每个人提供独立的端点和 SASL 凭据：这样，记录中的 principal 就能表明操作由谁执行。共用凭据无法提供此类信息。

工具

批量操作

describe_topic、sample_messages、get_message、get_schema、 consumer_lag、describe_consumer_group、open_transactions、 add_partitions、alter_topic_config、create_topic、commit_offset、 delete_consumer_group、delete_records、delete_topic、copy_message 和 produce_message 的目标只能通过 items 数组指定。没有单目标形式： 单次操作就是一个长度为 1 的 items 数组。

{"items": [{"topic": "orders"}]}
{"items": [{"topic": "orders"}, {"topic": "payments"}]}

所有用于命名或定义操作的内容都放在条目上，因此每个字段及其说明只会出现在 一个位置。控制整个调用的内容则保留在顶层，实际使用中也就是 confirm。

消息密集型工具最多接受 500 个条目；滞后、架构和管理类工具最多接受 1000 个。 如果消息数量超出此限制，transform_messages 可在一次调用中扫描主题， 而 produce_message 可通过 upload 发送任意数量的消息。

每个响应都使用相同的封装格式。结果按输入顺序排列，每个条目都包含 index， 以及 result 或 error，之后是 succeeded、failed、applied 和 atomic: false。 条目失败会作为数据放在响应中，而不会导致整个调用报错，因此成功的条目不会 被隐藏。批量写入会先预览每个条目，只有顶层 confirm 为 true 时才会应用有效 条目。它们不是事务：如果后续条目失败，Kafka 无法回滚主题、分区、偏移量或 已生成的消息。重复的写入目标会在任何更改发生之前被拒绝；但 produce_message 除外，因为两个相同的条目代表两条消息，而不是错误。

list_clusters

列出此服务的各个端点所服务的集群，并说明每个集群是否可连接、是否配置为 read_only，以及此会话是否可以向其写入（writable：除非端点为只读，否则为其自身集群；其他集群仅在 destinations 中列出时可写入）。无需参数。每个端点都提供此工具，因此会话可以发现 copy_message、produce_message、transform_messages 和 compare_clusters 可以操作哪些目标。

{"clusters": [
  {"name": "prod", "connected": true, "read_only": true, "writable": false},
  {"name": "preprod", "connected": true, "read_only": false, "writable": true}
], "count": 2}

connected 会在调用时检查，而非在启动时记录，因此会如实报告已停止运行的集群。只报告名称、可连接状态和可写状态：不会报告代理地址和凭据，因为每个端点都可以访问此工具。

compare_clusters

将其他集群的主题与此端点所服务集群的主题进行比较，并报告差异。可用于查找预生产环境中存在、生产环境中缺少的主题，或检查两个环境是否仍保持一致。

参数	类型	必填	含义
items	object[]	是	要进行比较的 1 到 1000 个集群

条目字段：

字段	类型	必填	含义
cluster	string	是	另一个集群。使用 list_clusters 查找有效名称
search	string	否	主题名称必须包含的、不区分大小写的子字符串
include_internal	bool	否	默认值为 false：排除内部主题

{"here": "preprod", "there": "prod",
 "here_cluster": {"name": "preprod", "brokers": 1, "topics": 12},
 "there_cluster": {"name": "prod", "brokers": 3, "topics": 11},
 "only_here": [{"topic": "orders-v2", "partitions": 6, "replication_factor": 1,
                "configs": {"retention.ms": "604800000"}}],
 "only_there": [],
 "differing": [{"topic": "orders", "differences": ["partitions"],
                "here": {"partitions": 1}, "there": {"partitions": 12}}],
 "in_both": 11}

only_here 和 only_there 指明差异的方向，由你调用的端点决定：“此处”始终指此端点所服务的集群。条目包含拥有该主题的集群中的分区数、副本因子和显式设置的配置，因此可以直接传给 create_topic。

此工具不会创建或更改任何内容。 若要创建缺失的主题，请将选定的条目传给 create_topic；它会先根据代理进行预览，并警告分区数无法减少。

differing 通常更有价值：两边都存在、但分区数或 retention.ms 不同的主题，通常是某个 bug 只在一个环境中复现的常见原因。比较时只检查主题自行设置的配置，因为两个集群可能有不同的代理默认值；若比较继承的值，就会报告每个主题都有差异。

存在差异不一定意味着有问题。生产环境中缺少某个主题往往是有意为之，因此报告只说明有哪些差异，而不会判断哪种情况才正确。

主题列表来自 Kafka 客户端的元数据缓存，缓存数据可能有几秒延迟，因此刚刚创建的主题仍可能出现在 only_there 中。请重复比较，而不要再次创建该主题。

list_topics

按名称对集群中的主题排序并列出，每个主题包含其分区数、副本因子、为自身设置的配置以及磁盘占用空间。

Parameter	Type	Required	Meaning
script	string	no	用于判断是否列出某个主题的 JavaScript 谓词
timeout_seconds	int	no	执行脚本的时间上限。默认值为 30，最多为 3600

谓词可读取以下变量：

Variable	Type	Meaning
topic	string	主题名称
partitions	number	分区数
replication_factor	number	第一个分区的副本数
internal	boolean	Kafka 自有主题，例如 __consumer_offsets
configs	object	此主题为自身设置的值，例如 configs['retention.ms']
size_bytes	number	此主题日志段在磁盘上的单份副本大小；未报告时为 -1

return topic.indexOf('orders') >= 0
return size_bytes > 10e9                       // the topics filling the disk
return partitions > 6
return configs['cleanup.policy'] === 'compact'
return replication_factor === 1 && !internal

{"name": "list_topics", "arguments": {"script": "return partitions > 6"}}

{"topics": [{"topic": "orders", "partitions": 12, "replication_factor": 3,
             "configs": {"retention.ms": "604800000"}, "size_bytes": 48318382080}],
 "count": 1}

过滤仅支持 JavaScript，与 search_messages 一样：要匹配名称，可使用 return topic.indexOf('orders') >= 0。谓词还能回答子字符串匹配无法回答的问题——哪些主题的分区数超过 6、只有一个副本，或采用压缩清理策略。

仅报告主题为自身设置的配置。不会包含继承的集群默认值，因为包含这些值会让每个主题看起来都配置过。

如果主题的谓词抛出异常，或因超时而被中断，该主题会计入 script_errors，而不会列出，因此损坏的过滤器绝不会被误认为是空集群。

大小来自 DescribeLogDirs，需要对集群具有 describe 权限。如果 Broker 未响应，则所有 size_bytes 都为 -1，且 warnings 会说明原因。 大小表示压缩后的磁盘占用字节数。

describe_topic

报告主题的分区、偏移量范围、消息数、磁盘占用空间、时间跨度以及完整配置。可在搜索前使用，以了解搜索会读取多少数据，以及该主题最多能保留多久的数据。

Parameter	Type	Required	Meaning
items	object[]	yes	要描述的主题，数量为 1 到 500 个

条目字段：

Field	Type	Required	Meaning
topic	string	yes	要描述的主题

{"results": [{"index": 0, "result": {
   "topic": "orders", "partition_count": 1, "message_count": 3,
   "size_bytes": 2048, "replicated_size_bytes": 6144,
   "partitions": [{"partition": 0, "start_offset": 0, "end_offset": 3, "message_count": 3, "size_bytes": 2048}],
   "configs": [{"key": "cleanup.policy", "value": "delete", "source": "DYNAMIC_TOPIC_CONFIG", "is_default": false},
               {"key": "retention.ms", "value": "604800000", "source": "DEFAULT_CONFIG", "is_default": true}]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

configs 会以 Kafka 报告的字符串形式列出每项主题配置，其中 -1 表示无限制。若 is_default 为 true，表示该值是继承而来，而非在主题上设置。两项配置决定消息是否仍可能存在：retention.ms（消息保留时长）和 cleanup.policy（compact 仅保留每个键对应的最新消息）。

size_bytes 表示主题的单份副本大小，按分区及总量列出； replicated_size_bytes 则统计所有副本的大小，也就是 Broker 磁盘上的实际占用。若 Broker 未报告日志目录，则这两项会连同警告一起省略。

sample_messages

读取少量最新消息样本，并报告其外观： 值的格式、带类型的字段路径、关键统计信息、哪些值字段包含消息键，以及写入值时使用的架构。Avro、 Protobuf 和 JSON Schema 值会先解码，因此其字段路径会以 JSON 的形式报告。可在搜索前使用此工具来决定搜索方式，并在 生成消息前使用此工具来查找要依据的架构。

参数	类型	必填	含义
items	object[]	是	要采样的主题，数量为 1 到 500 个

项目字段：

字段	类型	必填	含义
topic	string	是	要采样的主题
sample_size	int	否	要读取的消息总数。默认值为 20，最多 1000 条
partitions	int[]	否	限定为这些分区
max_value_bytes	int	否	每条消息的值字节数。默认值为 512，最多 1 MiB

{"results": [{"index": 0, "result": {
   "value_formats": {"json": 20, "text": 0, "binary": 0},
   "json_fields": [{"path": "payload.amount", "types": ["number"], "present": 20, "example": "500"}],
   "key_stats": {"present": 20, "absent": 0, "unique": 20, "all_unique": true},
   "key_in_value": ["payload.orderId"],
   "schemas": [{"format": "avro", "schema_id": 7, "message_type": "shop.Order", "count": 20}],
   "value_bytes": {"min": 180, "p50": 412, "max": 9020},
   "sampled_ranges": [{"partition": 0, "start": 980, "end": 1000}]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

key_in_value 指定的字段表示键就是该标识符，因此仅搜索键即可进行精确且低成本的查找。value_formats 还会统计 avro、protobuf、json_schema、msgpack、null 和 undecodable——这些值指定了架构但无法解码，这是配置问题，而不是二进制数据。

search_messages

扫描主题中范围受限的消息，并使用 JavaScript 表达式筛选消息。Kafka 不支持服务端搜索，因此此工具会读取消息并在客户端筛选；结果会报告扫描范围。

参数	类型	必填	含义
topic	string	是	要搜索的主题
script	string	否	JavaScript 筛选表达式。省略则匹配每条消息
parallelism	int	否	并发读取器数量，范围为 1–16。默认值为 1
partitions	int[]	否	限定为这些分区
from_offset / to_offset	int	否	Offset 窗口，结束位置不包含在内
from_timestamp / to_timestamp	string	否	RFC3339 时间窗口
direction	string	否	newest_first（默认）或 oldest_first
max_matches	int	否	达到此匹配数量后停止。默认值为 10，最多 1000 条
max_messages_scanned	int	否	最多读取的消息数。默认值为 10000，最多 100000000 条
max_value_bytes	int	否	每条匹配消息的值字节数。默认值为 512，最多 1 MiB
timeout_seconds	int	否	墙上时钟时间限制。默认值为 30，最多 3600 秒
count_only	bool	否	仅返回计数，不返回消息正文
download	object	否	{format, filename}：通过 HTTP 链接，将每条匹配结果保存到用户的计算机上，格式为 jsonl（默认）或 csv
group_by	string	否	返回每条匹配结果分桶的表达式；统计每个桶中的数量
max_groups	int	否	使用 group_by 时返回的最大桶数。默认值为 20，最多 1000 个

脚本

返回 true 以保留消息。作用域内：

名称	值
value	解析后的 JSON 文档；对于 Avro、Protobuf、JSON Schema 或已配置的格式，是解码后的记录；其他情况则是原始文本
key	字符串；如果键具有架构，则为解码后的文档；如果不存在，则为 null
headers	标头名称到字符串的对象
partition、offset	数字
timestamp	一个 Date
value_bytes、key_bytes	解码前的原始字节大小
format	json、avro、protobuf、json_schema、msgpack、text、binary 或 null
schema_id	值所指向的注册表 ID；如果没有，则为 null
decode_error	值指向架构但无法解码的原因；如果没有，则为 null

return key === 'order-123'
return value.eventType === 'NEW' && value.payload.amount >= 500
return value.payload.cancelledAt === null      // present and null
return value.payload.cancelledAt === undefined // field absent
return headers['correlation-id'] === 'corr-999'
return /ORD-\d{4}/.test(value.payload.orderId)
return value.indexOf('ERROR') >= 0             // non-JSON topic: value is a string
return schema_id === 57                        // written with one schema
return value_bytes > 900000                    // close to max.message.bytes

按键精确搜索比搜索消息正文准确得多：123 也会出现在 "amount": 1123 中，这些误报可能占满 max_matches，导致找不到目标消息。

如果脚本处理消息时抛出异常，该消息会计入 script_errors，扫描会继续，因此可以区分脚本错误和确实没有匹配项。脚本在沙箱中运行，无法访问文件系统、网络或主机；如果超出搜索超时，脚本会被终止。脚本不受内存限制：例如 'x'.repeat(1e12) 可能耗尽服务进程的内存。脚本属于可信输入；影响范围是此服务，而非集群。

扫描顺序

每个分区会同时读取，每次读取一个数据块：先读取每个分区中最新的数据块，再读取其前一个数据块，以此类推。因此，限制结果数量的最新优先搜索返回的是主题中最新的匹配项，而不是碰巧先读取的某个分区中的最新匹配项。

Kafka 会在分区内对记录排序，但不会跨分区排序，因此匹配项会按时间戳合并并报告；时间戳相同时，则按分区和偏移量排序。除非主题使用 LogAppendTime，否则时间戳由生产者设置，因此可能存在偏差；但时间戳仍是唯一可用于比较不同分区的依据。

一次扫描通过单个连接读取所有分区，因此宽主题所需的连接数不会比窄主题更多。但它确实会读取更多内容：对于包含 12 个分区的主题，限制搜索会检查全部 12 个分区的最新数据块，而不是在第一个分区内停止。max_messages_scanned 仍会限制扫描范围；在宽主题上，它可能比在窄主题上更早成为 stopped_reason。

并行度

parallelism 会使用指定数量的并发读取器读取单分区主题，这能加快对单个大型分区的完整扫描。读取器会按从新到旧（或从旧到新）的顺序读取数据块；只有按此顺序完成的数据块才会触发 max_matches 停止条件，因此返回的匹配项与顺序搜索相同。由于读取器可能按不同顺序完成，scanned_ranges 可能会为该分区列出多个区间。多分区主题已经会跨分区并行读取，因此此设置不适用于这类主题；无法再划分的范围则由一个读取器读取。

此设置适用于 count_only、group_by、download 以及单分区的完整扫描。

结果

{"topic": "orders", "match_count": 1,
 "matches": [{"partition": 0, "offset": 17, "key": "order-42", "value": "...", "encoding": "utf8"}],
 "scanned_messages": 120, "scanned_ranges": [{"partition": 0, "start": 0, "end": 120}],
 "stopped_reason": "range_exhausted", "complete": true}

只有读取了整个范围时，complete 才为 true。只有当 complete 为 true 时，空的匹配列表才表示“没有匹配项”；否则，请检查 stopped_reason 并缩小搜索范围。

对于大型结果集，可先使用 count_only 确认匹配项数量，然后使用 download 保存匹配项，而不是直接返回它们。

下载

download 目前不会执行任何扫描。它会检查主题、窗口和脚本，然后返回一个链接，以及一条用于在用户机器上运行的命令（参见 用户机器上的文件）。搜索会在获取链接的同时运行，并将每条匹配结果流式写入文件；如果搜索因 max_messages_scanned 或 timeout_seconds 而提前结束，传输也会中断，而不会留下一个看起来完整的文件。

{"name": "search_messages", "arguments": {"topic": "orders-dlq",
  "script": "return value.amount >= 500", "max_messages_scanned": 10000000,
  "download": {"format": "csv", "filename": "orders.csv"}}}

{"topic": "orders-dlq", "matches": [], "stopped_reason": "not_started",
 "download": {"url": "https://kafka.example.com/mcp/prod/files/Q2x…", "method": "GET",
   "command": "curl -fsS -o 'orders.csv.part' 'https://…/files/Q2x…' && mv 'orders.csv.part' 'orders.csv'",
   "expires_at": "2026-10-23T10:10:00Z", "note": "…"}}

文件采用 JSON 行格式，或在设置 format: "csv" 时采用 CSV 格式。两种格式都使用 kafka-console 的导出布局，因此文件可直接导入该工具。该文件是数据副本，而非数据视图：

• 值不会被截断；max_value_bytes 不适用；
• 空键、空值或空标头仍为空（CSV 中对应 key_null 和 value_null 列），与空字符串不同；
• 负载要么是原始字节，且 key_encoding 和 value_encoding 设为 utf8 或 base64；要么是在其带有 Schema Registry 架构 ID 时，保存为解码后的文档，并附有 value_format、 value_schema_id 和 value_schema_message，以便导入时再次编码为相同的 架构。

{"partition":0,"offset":3,"timestamp":"2026-10-23T09:00:00Z","key":"o-1","key_encoding":"utf8","key_format":"text",
 "value":"{\"amount\":900,\"id\":\"o-1\"}","value_encoding":"utf8","value_format":"avro","value_schema_id":1,
 "value_schema_message":"shop.Order","headers":[],"key_bytes":3,"value_bytes":11,"truncated":false}

分组

group_by 会进行汇总，而非列出结果。它是一个表达式，使用与 script 相同的变量，并返回一个分组名称。范围内的每条匹配结果都会按分组计数，与 count_only 的行为相同，且不会返回消息正文。null 和 undefined 会归入 null 分组；对象则按其 JSON 内容分组。

{"name": "search_messages", "arguments": {"topic": "orders-dlq",
  "group_by": "return headers['error-reason']", "from_timestamp": "2026-10-10T06:00:00Z"}}

{"topic": "orders-dlq", "match_count": 1840, "matches": [],
 "groups": [{"key": "timeout", "count": 1702, "example": {"partition": 2, "offset": 90311}},
            {"key": "bad_schema", "count": 138, "example": {"partition": 0, "offset": 4410}}],
 "complete": true}

分组按计数排序。存在的分组数量超过 max_groups 时，groups_truncated 为 true。表达式执行时发生异常的消息会计入 script_errors，且不会归入任何分组。group_by 不能与 download 同时使用。

get_message

读取指定偏移量的消息，以及可选的邻近消息。

参数	类型	必填	含义
items	object[]	是	要读取的 1 至 500 个地址

条目字段：

字段	类型	必填	含义
topic	string	是	要读取的主题
partition	int	是	要读取的分区
offset	int	是	要读取的指定偏移量
context	int	否	同时返回两侧的消息数量，最多 100 条
max_value_bytes	int	否	要返回的值字节数。默认值为 4096，最多 1 MiB

{"name": "get_message", "arguments": {"items": [
  {"topic": "orders", "partition": 0, "offset": 17, "context": 1}]}}

采用架构编码的值会被解码为 JSON，并通过 format、schema_id 和 message_type 说明其格式。无法解码且不是有效 UTF-8 的值会进行 base64 编码，并将 encoding 设为 base64；decode_error 会说明一个声明了架构但无法解码的值。

{"partition": 0, "offset": 17, "key": "o-1",
 "value": "{\"amount\":42,\"id\":\"o-1\"}", "format": "avro",
 "schema_id": 7, "message_type": "shop.Order", "encoding": "utf8", "value_bytes": 12}

get_schema

按 subject 或消息携带的 schema_id 从集群的 Schema Registry 读取 schema。在向采用 schema 编码的主题写入消息前先读取 schema：它会列出值所需的每个字段和枚举，而采样消息可能无法展示这些内容。

参数	类型	必填	含义
items	object[]	是	要查找的 1 至 1000 个 schema

条目字段，提供 subject 或 id：

字段	类型	必填	含义
subject	string	二选一	Subject，通常为 <topic>-value 或 <topic>-key
version	int	否	Subject 版本。默认为最新版本
id	int	二选一	Schema ID，例如来自 get_message
check_schema	object	否	{"schema": "...", "type": "avro"}：检查候选 schema 是否符合 subject，而不进行注册。需要提供 subject

{"results": [{"index": 0, "result": {
   "schema_id": 7, "subject": "orders-value", "version": 3, "versions": [1, 2, 3],
   "type": "avro", "schema": "{...}", "references": [],
   "message_types": null, "used_by": null, "compatibility": "BACKWARD",
   "check": {"compatible": false, "messages": ["{errorType:'TYPE_MISMATCH', …}"]}}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

message_types 会列出 Protobuf 消息，produce_message 会将其作为 message_type 使用。通过 id 查找时，used_by 会列出使用该 ID 的 subject 版本。按 subject 查找会返回 compatibility，即当前生效的兼容性级别；如果 subject 未设置该级别，则继承 Registry 的全局级别。check_schema 会询问 Registry，在该级别下候选 schema 是否可以注册为该 subject 的下一个版本。若无法注册，check.messages 会给出 Registry 提供的原因。不会注册任何内容。如果集群没有 schema_registry，整个调用会失败。

list_consumer_groups

列出消费者组及其状态、成员数量和所消费的主题。

参数	类型	必填	含义
topic	string	否	仅列出正在消费此主题或已对此主题提交偏移量的组
states	string[]	否	按状态筛选，例如 Stable、Empty

{"groups": [{"group": "payments", "state": "Stable", "members": 2, "topics": ["orders"]}], "count": 1}

状态为 Empty 的组仍可能报告滞后：已提交的偏移量会在创建它们的消费者退出后继续保留。Kafka 没有主题到组的索引，因此按 topic 筛选会描述集群上的每个组。

describe_consumer_group

详细描述消费者组：成员有哪些、每个成员拥有哪些分区，以及组在每个分区上的情况。用它将“某个分区卡住了”定位为“这台主机上的这个 Pod 卡住了”。

参数	类型	必填	含义
items	object[]	是	1 至 1000 个组，每个组为 {"group": "...", "sample_seconds": 30}

sample_seconds（可选，最多 60 秒）会在描述组之前持续观察指定时长，每秒读取一次，并添加 observation：

"observation": {"seconds": 30, "samples": 31,
  "states": ["Stable", "PreparingRebalance", "CompletingRebalance", "Stable"],
  "joined": [{"member_id": "payments-7-b…", "client_id": "payments-7", "host": "/10.0.4.17"}],
  "left":   [{"member_id": "payments-7-a…", "client_id": "payments-7", "host": "/10.0.4.17"}],
  "unstable": true}

重启的成员会以新的 member_id 重新加入，因此它会以相同的主机名同时出现在 left 和 joined 中。这正是发现重平衡风暴中反复崩溃重启的消费者的方式。调用会在观察时段内一直阻塞。

{"results": [{"index": 0, "result": {
   "group": "payments", "state": "Stable", "protocol_type": "consumer", "assignor": "cooperative-sticky",
   "coordinator": 1, "total_lag": 4200,
   "members": [{"member_id": "payments-7-…", "client_id": "payments-7", "host": "/10.0.4.17",
                "assignments": [{"topic": "orders", "partitions": [0, 1]}]}],
   "partitions": [{"topic": "orders", "partition": 0, "has_commit": true, "committed_offset": 812,
                   "end_offset": 5012, "lag": 4200, "member_id": "payments-7-…",
                   "client_id": "payments-7", "host": "/10.0.4.17"}]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

分区是成员拥有的分区与组已提交偏移量的分区之并集，因此状态为 Empty 的组仍会显示其位置。has_commit: false 表示该组拥有一个从未提交过偏移量的分区，因此它从何处开始由消费者的 auto.offset.reset 决定，而不是由偏移量决定。

open_transactions

查找阻塞 read_committed 消费者的未完成事务。事务型生产者在事务中途挂起或退出，会导致分区的最后稳定偏移量停滞不前，所有 read_committed 消费者都会停在那里。在 consumer_lag 中，这看起来与毒消息完全一样。

参数	类型	必填	含义
items	object[]	是	1 到 1000 个主题，每项为 {"topic": "..."}

{"results": [{"index": 0, "result": {
   "topic": "orders", "blocked": true,
   "partitions": [{"partition": 0, "last_stable_offset": 812, "high_watermark": 5012,
                   "unreadable_messages": 4200,
                   "producers": [{"producer_id": 2004, "producer_epoch": 0, "transaction_start_offset": 812,
                                  "transactional_id": "payments-writer-1", "state": "Ongoing",
                                  "started_at": "2026-10-02T08:14:03Z", "open_for": "41m12s", "timeout_ms": 900000}]}]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

解决方法在生产者端：重启或隔离由 transactional_id 指定的应用，或者等待 timeout_ms，此后代理会中止事务。移动消费者的偏移量没有帮助。

consumer_lag

衡量主题的消费者落后多少、消息生产和消费的速度，以及积压何时会清空。

参数	类型	必填	含义
items	object[]	是	要执行的 1 到 1000 次测量

条目字段：

字段	类型	必填	含义
topic	string	是	要测量的主题
group	string	否	默认为所有消费该主题的组
sample_seconds	int	否	消费速率采样窗口。默认值为 5，最多 300。调用会阻塞
skip_consume_rate	bool	否	立即返回，不提供速率或估算值
measure_backlog_age	bool	否	同时报告每项积压的持续时间，并与保留期限比较。每个滞后分区执行一次获取

每次最多并行运行 4 个采样窗口，因此每处理 4 个条目，调用大约耗时 sample_seconds。

{"results": [{"index": 0, "result": {
   "topic": "orders", "total_lag": 4200,
   "produce_rate": {"last_minute": {"messages": 3000, "per_second": 50, "per_minute": 3000, "per_hour": 180000}},
   "groups": [{"group": "payments", "state": "Stable", "members": 2, "lag": 4200,
               "consume_rate": {"per_second": 120, "sampled_seconds": 5},
               "drain_per_second": 70, "eta_seconds": 60, "eta_human": "1m 0s",
               "status": "draining"}]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

两种速率的测量方式不同，输出中会说明：

• produce_rate — 根据消息时间戳得出的历史数据，涵盖过去 1 秒、1 分钟和 1 小时。window_truncated 表示主题的存续时间短于该窗口。
• consume_rate — 采样值：先读取已提交偏移量，再于 sample_seconds 后重新读取。sample_inconclusive 表示偏移量没有变化。

积压会以消费速率减去生产速率的速度减少。status 说明这些数字的含义：caught_up、draining（附带预计时间）、growing（永远无法清空，并提供 growing_by_per_minute）、stalled、no_active_consumers 或 not_measured。只有滞后量确实在减少时才会给出预计时间。

每个分区还会报告 start_offset；当已提交偏移量小于该值时，会显示 offset_expired: true。此时，保留策略已删除该组的位置：消费者会根据其 auto.offset.reset 重置，而不是从原位置恢复，并且该组会带有警告。produce_rate.partitions 提供每个分区过去 1 分钟和 1 小时的计数，可用于区分键倾斜和消费者速度缓慢。

启用 measure_backlog_age 后，每个滞后分区都会报告 committed_timestamp（下一条待消费消息的时间戳）和 lag_seconds。每个组都会报告 oldest_unconsumed_at，主题则会报告 retention_ms。当最早的未消费消息已消耗一半保留期限时，会设置 retention_risk 并给出警告：如果该组未能赶上，保留策略会在消息被消费前将其删除。

cluster_health

通过一次调用检查此端点所服务的集群：代理、控制器以及所有未完全健康的分区。

参数	类型	必填	含义
search	string	否	仅检查名称包含此内容的主题。不区分大小写
include_internal	bool	否	也检查 __consumer_offsets 和其他内部主题

{"cluster_id": "…", "controller": 1, "healthy": false,
 "brokers": [{"id": 1, "host": "kafka-1", "port": 9092, "rack": "eu-1a", "controller": true, "leaders": 61}],
 "summary": {"topics": 40, "partitions": 182, "offline": 0, "under_replicated": 3, "under_min_isr": 1, "errored": 0},
 "problems": [{"topic": "orders", "partition": 4, "issues": ["under_replicated", "under_min_isr"],
               "leader": 1, "replicas": [1, 2, 3], "isr": [1], "min_insync_replicas": 2}],
 "warnings": []}

offline 表示分区没有主节点，因此无法读取或写入。under_replicated 表示某个副本不同步。under_min_isr 是 NOT_ENOUGH_REPLICAS 背后的状况：使用 acks=all 的生产者会失败，直到同步副本恢复。某个代理不再领导任何分区，而其他代理领导许多分区时，通常说明该代理重启后从未重新获得领导权。

问题分区在代理之间迁移时，会包含 reassigning: true 以及对应的 adding_replicas 和 removing_replicas，summary.reassigning 会统计这些分区。新增副本在复制完日志前都不同步，因此迁移期间出现 under_replicated 属于预期情况，并非服务中断。

一些兼容 Kafka 的代理（包括 Redpanda）不会报告 min.insync.replicas。受影响的主题会列在 min_isr_unknown 中；对于这些主题，不会判断 under_min_isr，而不是进行猜测。

list_acls

列出访问控制条目，适用于客户端遇到 TOPIC_AUTHORIZATION_FAILED 或 GROUP_AUTHORIZATION_FAILED 的情况。

参数	类型	必填	含义
principal	string	否	仅显示此主体，例如 User:payments
resource_type	string	否	topic、group、cluster、transactional_id、delegation_token
resource_name	string	否	此名称上应用的所有 ACL，包括带前缀的 ACL 和 *。需要指定 resource_type

{"acls": [{"principal": "User:payments", "host": "*", "resource_type": "topic", "resource_name": "orders",
           "pattern_type": "literal", "operation": "read", "permission": "allow"}], "count": 1}

deny 会覆盖所有 allow。没有授权器的代理会报错 SECURITY_DISABLED，这表示完全不会强制执行 ACL。

server_config

报告生效配置：端点名称、精确路径、描述、集群、代理、身份验证机制和主体、TLS、只读状态、导出目录以及此端点公开的工具。不接受任何参数。绝不会报告密码。

tools 列出的是此端点的工具，而非整个部署的工具：只读端点不会注册 add_partitions、alter_topic_config、commit_offset、create_topic、delete_consumer_group、delete_records 和 delete_topic；任何端点都会省略其 tools 配置禁用的工具。

当结果出乎意料时，可以使用此工具：在本地代理上主题列表为空，与在生产环境中主题列表为空，含义截然不同。

add_partitions

向主题添加分区。不可逆——Kafka 无法减少分区数。只在非只读端点上提供。

参数	类型	必填	含义
items	object[]	是	要更改的 1 至 1000 个主题
confirm	bool	否	默认值为 false：仅预览，不做任何更改

项目字段：

字段	类型	必填	含义
topic	string	是	要更改的主题
partitions	int	是	最终分区总数，而非要添加的数量。重复调用是安全的
acknowledge_key_ordering	bool	否	消息带有键时必填
sample_size	int	否	检查键的消息数。默认值为 20

不设置 confirm 时，工具会报告将要发生的变化：当前和目标分区数、消息是否带有键、哪些消费者组将重新平衡，以及相关警告。

添加分区会改变键所哈希到的分区，因此现有键将失去顺序保证。因此，带键的主题还必须设置 acknowledge_key_ordering。如果请求的分区数少于主题现有分区数，操作会被拒绝并说明原因，而不会尝试执行。

alter_topic_config

更改主题级配置：保留时间、清理策略、最大消息大小，以及 Kafka 允许按主题设置的其他配置。更改是增量式的，因此未指定的每个键都会保留其值。只在非只读端点上提供。

参数	类型	必填	含义
items	object[]	是	要更改的 1 至 1000 个主题
confirm	bool	否	默认值为 false：仅预览，不做任何更改

项目字段：

字段	类型	必填	含义
topic	string	是	要更改的主题
set	object	否	要设置的键，例如 {"retention.ms": "86400000"}
delete	string[]	否	要移除的覆盖项，以便再次应用集群默认值

{"results": [{"index": 0, "result": {
   "topic": "orders", "applied": false, "messages_past_retention": 18000,
   "changes": [{"key": "retention.ms", "action": "set", "current": "604800000",
                "current_source": "DYNAMIC_TOPIC_CONFIG", "requested": "86400000"}],
   "warnings": ["18000 message(s) are already older than the new retention of 24h0m0s and become eligible for deletion as soon as it applies; Kafka cannot bring them back"]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

预览会请求代理验证更改，因此未知键或无效值会在设置 confirm 前被拒绝。缩短 retention.ms 时，会报告已有多少条消息超出新的限制；更改 cleanup.policy 或设置 retention.bytes 时会发出警告。应用更改后，会从代理重新读取每个键，这是获知已删除覆盖项的继承值的唯一方式。

create_topic

创建主题。如果主题已存在，则拒绝创建，而不是调整该主题。 只读端点不会公开此工具。

Parameter	Type	Required	Meaning
items	object[]	yes	要创建的 1 至 1000 个主题
confirm	bool	no	默认为 false：代理会验证请求，但不会创建任何内容

项目字段：

Field	Type	Required	Meaning
topic	string	yes	要创建的主题名称
partitions	int	no	在 Kafka 2.4+ 上省略时使用代理默认值。之后可以增加，不能减少
replication_factor	int	no	在 Kafka 2.4+ 上省略时使用代理默认值。不能超过代理数量
configs	map	no	主题级配置，例如 retention.ms 或 cleanup.policy

如果不指定 confirm，请求会以 ValidateOnly 发送给代理，因此预览会报告集群给出的实际结果——例如无效名称、未知配置键或大于集群规模的复制因子——而不是猜测。指定 confirm 后，主题会被创建，并从集群读取创建后的分区数和复制因子；这样，省略的数量就会报告为代理实际选用的值。

使用代理默认值需要 Kafka 2.4 或更高版本，其 CreateTopics v4 API 引入了 -1，表示“使用代理默认值”。对于较旧的代理，请明确传入两个数量。

如果复制因子大于代理数量，此处会拒绝请求，并在消息中注明代理数量，因为不同代理对仅验证请求是否会捕获此问题的处理方式并不一致。

使用 add_partitions 更改现有主题的分区数；此工具绝不会修改它未创建的主题。

delete_topic

删除主题。这是此处破坏性最强的工具：删除主题会销毁其中的每条消息，Kafka 无法撤销此操作，而且读取该主题的任何消费者组都会中断。只读端点不提供此工具。

参数	类型	必填	含义
items	object[]	是	要删除的 1 到 1000 个主题
confirm	bool	否	默认为 false：仅预览，不会删除任何内容

条目字段：

字段	类型	必填	含义
topic	string	是	要删除的主题。该主题必须存在
acknowledge_data_loss	bool	否	当主题仍包含消息时必填

如果未设置 confirm，工具会逐个主题报告将销毁的消息数量、该主题的分区数量，以及已为该主题提交偏移量的消费者组：

{"results": [{"index": 0, "result": {
   "topic": "orders-old", "partitions": 6, "message_count": 41207,
   "consumer_groups": ["payments"], "deleted": false, "would_delete": true,
   "warnings": ["41207 message(s) would be destroyed, and Kafka cannot restore them: the only recovery is a backup taken beforehand"]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

需要分别确认两项内容，因为它们回答的是不同的问题。confirm 表示调用方确实有意删除；acknowledge_data_loss 表示调用方了解其中的内容。若主题包含消息，但未同时提供这两项确认，删除请求将被拒绝。删除时会重新读取消息数量，因此即使主题在预览后新增了消息，也会被发现。

无论确认级别如何，都会直接拒绝删除 __consumer_offsets 等内部主题：它们保存的是集群状态，而非调用方的数据；删除其中任何一个都会导致所有消费者同时中断。

同一批次中的重复主题会在删除任何内容之前被拒绝。删除操作并非原子操作——如果后续条目失败，之前已删除的主题不会恢复。

delete_records

删除分区中最早的消息，但不会删除主题。主题、其配置和消费者组都会保留。不可撤销。 只读端点不会提供此功能。

Parameter	Type	Required	Meaning
items	object[]	yes	要截断的 1 到 1000 个分区
confirm	bool	no	默认值为 false：仅预览，不删除任何内容

项目字段：

Field	Type	Required	Meaning
topic	string	yes	要从中删除消息的主题
partition	int	yes*	要从中删除消息的分区。与 all_partitions 一起使用时不适用
before_offset	int	one of	删除此偏移量之前的所有内容；此偏移量将成为起始偏移量
before_timestamp	string	one of	RFC3339：删除此时间点之前写入的所有内容，按分区解析
all_partitions	bool	no	与 before_timestamp 配合使用，截断主题的每个分区
acknowledge_data_loss	bool	no	执行操作时必需

{"results": [{"index": 0, "result": {
   "topic": "orders", "partition": 0, "start_offset": 0, "end_offset": 5012, "before_offset": 812,
   "messages_deleted": 812, "would_delete": true, "deleted": false,
   "affected_groups": [{"group": "replay-job", "committed_offset": 100, "unprocessed_lost": 712}]}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

affected_groups 会列出所有提交位置低于截断点的组，以及它们在未处理消息的情况下会丢失的消息数；此类组将从新的起始位置继续。要清空分区，请将该分区的结束偏移量用作 before_offset。

before_timestamp 会解析为此时间点或之后写入的第一个偏移量，预览中的 before_offset 会显示解析结果。使用 all_partitions 时，一个项目涵盖整个主题：partitions 会列出每个分区各自的截断点、起始位置和结束位置，以及受影响的组；该项目包含 all_partitions: true 和 partition: -1，而 messages_deleted 是这些分区的总数。没有早于指定时间点的数据的分区不会更改；如果没有任何分区符合条件，该项目会被拒绝。偏移量不能跨分区使用，因此 all_partitions 只能接受时间戳。截断会逐个分区执行，不具备原子性。

delete_consumer_group

删除消费者组及其已提交的偏移量。可用于清理消费者已停用的组：这些组的提交记录会持续报告积压，但再也没有消费者会处理这些消息。只读端点不会提供此功能。

Parameter	Type	Required	Meaning
items	object[]	yes	1 到 1000 个组，每个组的格式为 {"group": "..."}
confirm	bool	no	默认值为 false：仅预览，不删除任何内容

预览会列出每个组已提交的偏移量和积压量。存在活跃成员的组会被拒绝，并在删除时再次检查。之后若有消费者使用相同的组 ID 启动，它会从其 auto.offset.reset 指定的位置开始，而不是从该组上次停止的位置开始。

commit_offset

移动消费者组已提交的偏移量。向前移动可跳过消息，向后移动可重放消息，或移动到某个时间点以重新处理此后所有消息。不可逆，因为被跳过的消息永远不会被处理。只读端点不提供此操作。

参数	类型	必填	含义
items	object[]	是	要执行的移动，数量为 1 到 1000
confirm	bool	否	默认值为 false：仅预览，不做任何更改

条目字段——必须且只能提供 offset、timestamp 或 position 中的一项：

字段	类型	必填	含义
topic	string	是	要移动偏移量的主题
group	string	是	要移动的消费者组
partition	int	否	与 offset 搭配时必填。与 timestamp 或 position 搭配时省略，即应用于所有分区
offset	int	—	组接下来读取的偏移量。要跳过偏移量 42，请提交 43
timestamp	string	—	RFC3339。每个分区移动到此时间或之后的第一条消息
position	string	—	earliest 或 latest
allow_active_members	bool	否	即使有正在运行的消费者，也继续执行

{"items": [{"topic": "orders", "group": "payments", "timestamp": "2026-10-01T09:00:00Z"}]}

{"results": [{"index": 0, "result": {
   "topic": "orders", "group": "payments", "state": "Empty", "members": 0,
   "partitions": [{"partition": 0, "current_offset": 5012, "target_offset": 4100, "replayed_messages": 912},
                  {"partition": 1, "current_offset": 4990, "target_offset": 4021, "replayed_messages": 969}],
   "replayed_messages": 1881, "applied": false}}],
 "succeeded": 1, "failed": 0, "applied": 0, "atomic": false}

如果某个分区在 timestamp 指定的时间或之后没有消息，该分区将移动到末尾，其条目会附带说明这一情况的 note。同一批次中，如果同一消费者组和主题同时包含整个主题的条目和分区条目，操作将被拒绝，因为具体生效哪一项会取决于顺序。

消费者组不能有活跃成员。正在运行的消费者会在内存中保留其位置，并且只会在加入时读取已提交的偏移量，因此，在它运行期间提交的偏移量会被它下一次提交覆盖，消费者组也不会移动。请先停止消费者。

copy_message

将消息复制到另一个主题，同时保留键、值和标头。只需提供每条消息的地址，而非其内容，因此只能复制集群中已有的消息。

参数	类型	必填	含义
items	object[]	是	要复制的 1 到 500 条消息
confirm	bool	否	默认值为 false：仅预览，不会写入任何内容

项目字段：

字段	类型	必填	含义
source_topic、source_partition、source_offset		是	要复制的消息
destination_topic	string	是	写入位置。必须已存在
destination_cluster	string	否	要写入的另一个集群。默认为此端点所属集群
translate_schema	bool	否	在目标集群的注册表中重新注册架构 ID
max_value_bytes	int	否	预览时的值大小上限。始终会复制整个值

将 destination_cluster 设为此服务所服务的另一个集群，即可复制到该集群。这样可以将生产环境中的消息复制到预生产主题，以便安全地调试。使用 list_clusters 查看有效名称。

每次复制都会附带来源标头 — kafka-mcp-copied-from-cluster、-from-topic、-from-partition、-from-offset、-copied-at、-copied-by-tool、-copied-by-principal — 因此可以追溯死信主题或预生产主题中的消息原始来源。如果消息已包含其中某个标头，则保留原始标头并报告冲突。

键和值的字节会原样复制。架构 ID 仅在颁发它的注册表中有效，因此当目标集群使用不同的注册表或没有注册表时，响应会发出警告。使用 translate_schema 时，架构会连同其引用一起注册到目标注册表中的 <destination_topic>-value（或 -key），并改写副本中的 ID；负载不会更改。注册架构会写入目标注册表，且仅在 confirm 为 true 时执行。

对于复制到此端点所属集群的操作，其 read_only 策略会保护目标位置。只读端点仍可作为跨集群复制的来源，因为从该端点复制不会更改任何内容。其他目标集群必须列在调用端点的 destinations 中，且不能配置为 read_only；list_clusters 会通过 writable 报告这一状态。如果目标为只读，则会完全拒绝该工具，包括预览操作，因为该工具只执行写入。

produce_message

向现有主题写入新消息。与 copy_message 不同，调用方需要提供内容，因此可以将消息写入从未收到过生产者消息的主题。

参数	类型	必填	含义
items	object[]	二选一	要写入的 1 到 500 条消息
upload	object	二选一	{filename}：通过 HTTP 链接发送用户设备上的文件
confirm	bool	否	默认值为 false：仅预览，不会写入任何内容

项目字段：

字段	类型	必填	含义
topic	string	是	要写入的现有主题
value	string	是	消息正文
key	string	否	未指定 partition 时，用于决定分区
headers	object	否	标头名称和值
partition	int	否	指定分区。省略时由 key 决定
encoding	string	否	utf8（默认）或 base64，用于已有的精确字节
value_schema	object	否	将 value（JSON）编码为注册表架构
key_schema	object	否	将 key（JSON）编码为注册表架构
destination_cluster	string	否	要写入的另一个集群。默认为此端点自身的集群
max_value_bytes	int	否	预览值的上限。始终会写入完整值

每条消息都会附带 kafka-mcp-produced-at、kafka-mcp-produced-by-tool、 kafka-mcp-produced-by-principal，并且在客户端标识自身时，还会附带 kafka-mcp-produced-by-client，因此可以将伪造消息与真实消息区分开来。 如果调用方提供了同名标头，则会按原样保留，并报告名称冲突。

value_schema 和 key_schema 接受 {subject, version, id, message_type}，这些字段均为可选：{} 表示使用目标集群注册表中 <topic>-value（或 -key）的最新版本，id 固定使用指定架构，message_type 则选择 Protobuf 消息。值以 JSON 格式提供，并在预览时根据架构进行验证——缺失或拼写错误的字段会按名称拒绝——随后会以包含架构 ID 的帧格式写入，这是支持注册表的使用者所需的格式。topic_formats 中指定了格式的主题会自动按该格式编码。两者均不能与 encoding: base64 搭配使用。响应会在 value_encoding / key_encoding 中报告所用架构。

主题必须已存在：如果主题不存在，系统会拒绝操作，而不会任由其自动创建。除非必须指定确切分区，否则请省略 partition——分区位置由 key 决定，而指定分区会导致带 key 的消息写入与其 key 哈希结果不符的分区，从而破坏该 key 的消息顺序。使用显式分区时，响应会发出警告。

upload 会从用户机器上的文件发送任意数量的消息，例如由 transform_messages 下载保存的文件。使用 stdio 时会立即读取文件；使用 HTTP 时，调用会返回一个链接和一条要在该处运行的 curl -T 命令（参见 用户机器上的文件）。文件中的每一行都是一个 JSON 条目，包含上述条目字段；拼写错误的字段会被拒绝，而不会忽略。文件会在上传时读取，且不会存储。命令会打印摘要，而不是逐行返回结果：lines、valid、invalid、written、前三条消息，以及按行号列出的最多 50 行失败内容。 如果没有 confirm，上传只会检查文件，不会写入任何内容；请使用 confirm 获取新链接以发送文件。有效行会按文件顺序写入，无效行则会跳过。

{"name": "produce_message", "arguments": {"upload": {"filename": "orders-fixed.jsonl"}, "confirm": true}}

read_only 会保护正在写入的集群，因此只读端点仍可向其 destinations 中列出的其他集群写入，但如果要写入端点自身的集群，则会直接拒绝——包括预览在内。已写入的消息无法删除；它会一直保留，直到保留策略将其移除。

transform_messages

使用 JavaScript 转换消息，并将结果写成新消息，写入同一主题、另一个主题或另一个集群。适用于修复这类问题，例如文档被封装在不应包含它的字段中：

{"name": "transform_messages", "arguments": {
  "topic": "orders",
  "script": "return value.payload !== undefined",
  "transform": "return { value: value.payload }",
  "destination_topic": "orders-fixed"
}}

参数	类型	必填	含义
topic	string	是	要读取的主题
transform	string	是	返回 {value, key, headers} 的 JavaScript，可返回其中任意子集，也可返回 null 以跳过
script	string	否	选择要转换哪些消息的谓词，用法与 search_messages 相同
partitions	int[]	否	要读取的分区。默认全部
from_offset	int	否	每个读取分区的起始偏移量（包含）
to_offset	int	否	结束偏移量（不包含）
from_timestamp	string	否	RFC3339 起始时间（包含）
to_timestamp	string	否	RFC3339 结束时间（不包含）
destination_topic	string	否	要写入的现有主题。默认使用源主题
destination_cluster	string	否	要写入的另一个集群
keep_partition	bool	否	写入源分区编号，而不让 key 决定分区
value_schema	object	否	按此注册表 schema 编码新值，用法与 produce_message 相同
key_schema	object	否	同上，用于 key
max_messages	int	否	要处理的选中消息数。默认 100000，最多 10000000
timeout_seconds	int	否	默认 300，最多 3600
max_value_bytes	int	否	示例值的限制。始终写入完整值
download	object	否	{filename}：通过 HTTP 链接，将新消息以 produce_message 项的形式保存到用户的机器上
confirm	bool	否	默认 false：不会向 Kafka 写入任何内容

转换函数是 JavaScript（ES2020：箭头函数、展开语法、解构、正则表达式、JSON、Math、Date；不支持 console、require、计时器或网络），每条消息运行一次，使用与 search_messages 相同的变量。 它会返回包含 value、key、headers 中任意字段的对象；省略的字段会保留原值。未知字段会被拒绝，以防拼写错误导致原内容未经修改便被写入。返回 headers 会替换所有 headers：若要保留原有内容，请展开原对象；若要移除某个 header，将其设为 null，例如 { headers: { ...headers, 'error-reason': null } }。

超过 2^53 的整数（例如 64 位 id）会以 BigInt 的形式传入转换函数，因此直接传递会原样写回。请用 BigInt 字面量（value.id === 1234567890123456789n）或 String() 对其进行比较。将其转为 Number 会造成舍入；如果消息输出包含这种舍入后的整数，该消息会被拒绝，而不会写入。script 看到的值与 search_messages 中的完全相同，因此在那里统计的数量与选中的数量一致。字符串值会以文本形式写入，其他值则以 JSON 形式写入。使用 schema id 读取的值会用同一个 schema 重新编码；value_schema 可选择其他 schema，如果目标使用不同的注册表，则必须指定。topic_formats 中的主题会按其格式编码。 如果不指定 confirm，响应会统计 scanned、matched、transformed、skipped 和 failed，显示三个转换前后的 examples，并在 errors 中按原因归类失败项，每项附带示例地址。如果指定 confirm，则会分批生成新消息，written 会说明写入了多少条。每条新消息都带有 kafka-mcp-transformed-from-cluster、-from-topic、-from-partition、-from-offset、-at、-by-tool、-by-principal，并且在已知时还会带有 -by-client。

原始消息绝不会更改：Kafka 无法编辑消息，因此修复后的版本会追加写入。写回源主题时，读取会在调用开始时确定的末尾偏移处停止，因此不会读取自己写出的内容。分区按升序读取。如果运行因 max_messages 或超时而提前结束，响应会报告 complete: false、stopped_reason 和 scanned_ranges；使用 partitions 和 from_offset 即可继续运行。写入不是原子操作，停止前已写入的内容会保留。

download 会将新消息保存到用户的计算机上：使用 stdio 时会一次性传输；使用 HTTP 时则通过链接传输。转换会在获取链接时运行，并将新消息作为 produce_message 项流式传输到用户的计算机上，以便检查后再使用 produce_message upload 发送。链接发出前会先转换第一条消息，因此转换失败会立即报告；如果运行失败或提前结束，传输也会中断。下载和预览都不会写入 Kafka，因此在只读端点上也能使用；download 不能与 confirm 同时使用，因为 confirm 需要可写的目标端点。与 copy_message 一样，该工具在只读端点上仍会注册，因为目标端点是在每次调用时选择的。

技能

Agent 会加载的唯一技能是 skills/kafka-debugging/SKILL.md。它会引导用户查看 skills/kafka-debugging/references/ 下的场景指南，而不是自身包含所有工作流，因此一次会话只会读取所需的指南：

• find-message.md — 根据用户已知的信息定位消息。
• check-lag.md — 衡量延迟和吞吐量，并判断积压何时会清空。
• scale-partitions.md — 判断增加分区是否有帮助，并安全地添加分区。
• skip-poison-message.md — 在保留消息的前提下，解除消费者被无法处理的消息卡住的状态。
• create-topic.md — 创建主题，并有意选择分区数和保留策略；也可将其用作 copy_message 的目标。
• produce-message.md — 写入消息：修复并重新注入消息、在另一个集群中重现故障，或为主题填充初始消息。
• compare-clusters.md — 找出两个环境之间的差异，并创建其中一个环境缺少的主题。
• delete-topic.md — 在用户了解删除后果后，移除主题及其中的所有内容。
• replay-messages.md — Bug 修复上线后，重新处理某个时间点以来的所有内容。
• tune-topic-config.md — 更改主题的保留策略、清理策略或消息大小，并了解更改对现有数据的影响。
• cluster-health.md — 查找导致 NOT_ENOUGH_REPLICAS 和主题不可读的离线分区及副本不足分区。
• purge-messages.md — 保留主题，同时删除某个分区中的旧消息；或移除已弃用的消费者组。
• authorization-error.md — 查明哪个 ACL 拒绝了客户端的请求。
• unstable-consumer-group.md — 查明消费者组持续重新平衡的原因，以及哪个消费者不断离开。
• partition-skew.md — 当一个分区出现延迟时，区分键倾斜和消费者缓慢，并找出热点键。
• schema-error.md — 在架构更改上线前进行测试，并追查是哪个架构导致生产者或消费者出现故障。

总指南还会处理这些指南之间的重叠情况：“消费者落后”会打开其中三个指南，而 consumer_lag 的 status 会决定哪一个指南适用。

开发

命令	用途
make env-up / make env-down	启动 / 停止 Redpanda 和 Console
make build	使用 goreleaser 构建到 dist/
make run	从源代码运行 HTTP 服务
go test ./...	所有测试，包括容器测试
go test -short ./...	无需容器的测试
go vet ./...	对所有包运行 vet

文档站点位于 _docs（Vite、pnpm）。在该目录中运行 pnpm install && pnpm dev 即可在本地提供服务；将 _docs/ 下的更改推送到 main 后，会通过 .github/workflows/docs.yml 发布到 GitHub Pages。

测试会使用由 internal/domain/testenv 启动的真实容器运行（一个 Redpanda 代理和 Console），因此运行完整测试套件必须能够使用 Docker。

测试和发布工作流使用最新的稳定版 Go（go-version: 'stable'、 check-latest: true），不受 go.mod 中版本的影响。模块的最低 Go 版本仍然适用。

启动 HTTP 服务

需要 Go 1.27 或更高版本。使用 chu 加载配置；设置 CONFIG_FILE 以选择 YAML 或 JSON 文件。into 管理进程生命周期，ada 使用 由上下文驱动的关闭机制提供 HTTP 服务，logi 则初始化结构化日志。

make env-up                                  # local Redpanda + Console
make run                                     # runs with --server and serves kafka-mcp.local.yaml

kafka-mcp.local.yaml 已提交到仓库，并指向 compose 代理，因此克隆仓库后 无需编写任何配置即可使用。make env-up 会将代理发布到 localhost:19092，将 Schema Registry 发布到 localhost:18081，并将 Redpanda Console 发布到 http://localhost:8080。

检查服务是否已启动：

curl http://localhost:8090/healthz