AI代理消息系统集成服务

StreamNative MCP Server为AI代理提供与StreamNative Cloud、Apache Kafka和Apache Pulsar消息系统的标准化集成接口。

  • by streamnative
  • Go
  • 本地部署
  • 24 星标
  • 更新于
在 GitHub 查看 streamnative/streamnative-mcp-server

使用说明

StreamNative MCP服务器

用于将AI代理与StreamNative Cloud资源和Apache Kafka/Pulsar消息传递系统集成的模型上下文协议(MCP)服务器。

概述

StreamNative MCP Server为LLM(大型语言模型)和AI代理提供了一个标准接口,用于与StreamNative Cloud服务、Apache Kafka和Apache Pulsar进行交互。此实施遵循 模型上下文协议 规范,使AI应用程序能够通过标准化的接口访问消息服务。

服务器当前正在协商MCP协议版本 2025-11-25, 2025-06-18, 2025-03-26,以及 2024-11-05。默认首选项为 2025-11-25,而较旧的客户端仍然通过协议协商得到支持。

特性

  • StreamNative云集成:
    • 通过身份验证连接到StreamNative Cloud资源
    • 切换到组织中可用的集群
    • 描述集群资源的状态
  • Apache Kafka支持:与Apache Kafka资源交互,包括:
    • Kafka管理操作(主题、分区、消费者组)
    • 架构注册表操作
    • Kafka连接操作(*)
    • Kafka客户端操作(生产者、消费者)
  • Apache Pulsar支持:与Apache Pulsar资源交互,包括:
    • Pulsar管理操作(主题、命名空间、租户、模式等)
    • Pulsar客户端操作(生产者、消费者)
    • 功能、来源和水槽管理
    • 只读MCP资源,用于上下文、目录和有界管理摘要
  • 多种连接选项:
    • 通过服务帐户身份验证连接到StreamNative Cloud
    • 直接连接到外部Apache Kafka集群
    • 直接连接到外部Apache Pulsar集群

*Kafka Connect操作仅在StreamNative Cloud上进行测试和验证。

安装

Homebrew(macOS和Linux)

安装streamnative mcp服务器的最简单方法是使用Homebrew:

# Add the tap repository
brew tap streamnative/streamnative

# Install streamnative-mcp-server
brew install streamnative/streamnative/snmcp

Docker镜像

StreamNative MCP服务器发布Docker镜像到 streamnative/snmcp,它可以通过docker命令运行stdio服务器和sse服务器。

# Pull image from Docker Hub
docker pull streamnative/snmcp 

Helm Chart(Kubernetes)

看 charts/snmcp/README.md 通过StreamNative图表存储库安装Helm。

Github发布

访问https://github.com/streamnative/streamnative-mcp-server/releases获取StreamNative MCP服务器的最新二进制文件。

来源

# Clone the repository
git clone https://github.com/streamnative/streamnative-mcp-server.git
cd streamnative-mcp-server

go mod tidy
go mod download

# Build the binary
make

用法

先决条件

如果你想访问你的StreamNative Cloud,你需要准备好以下资源:

  1. 访问 StreamNative云.
  2. StreamNative云组织
  3. StreamNative Cloud实例和集群
  4. 具有管理员角色的服务帐户
  5. 下载服务帐户密钥文件

启动MCP服务器

使用stdio服务器

# Start MCP server with StreamNative Cloud authentication
bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json

# Start MCP server with StreamNative Cloud authentication and pre-configured context
# When --pulsar-instance and --pulsar-cluster are provided, context mutation tools are disabled
bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json --pulsar-instance my-instance --pulsar-cluster my-cluster

# Start MCP server with external Kafka
bin/snmcp stdio --use-external-kafka --kafka-bootstrap-servers localhost:9092 --kafka-auth-type SASL_SSL --kafka-auth-mechanism PLAIN --kafka-auth-user user --kafka-auth-pass pass --kafka-use-tls --kafka-schema-registry-url https://sr.local --kafka-schema-registry-auth-user user --kafka-schema-registry-auth-pass pass

# Start MCP server with external Pulsar
bin/snmcp stdio --use-external-pulsar --pulsar-web-service-url http://pulsar.example.com:8080
bin/snmcp stdio --use-external-pulsar --pulsar-web-service-url http://pulsar.example.com:8080 --pulsar-token "xxx"

# Start MCP server with stdio by docker with StreamNative Cloud authentication
docker run -i --rm -e SNMCP_ORGANIZATION=my-org -e SNMCP_KEY_FILE=/key.json -v /path/to/key-file.json:/key.json -p 9090:9090 streamnative/snmcp stdio

使用SSE(服务器发送事件)服务器

# Start MCP server with SSE and StreamNative Cloud authentication
bin/snmcp sse --http-addr :9090 --http-path /mcp --organization my-org --key-file /path/to/key-file.json

# Start MCP server with SSE and pre-configured StreamNative Cloud context
# When --pulsar-instance and --pulsar-cluster are provided, context mutation tools are disabled
bin/snmcp sse --http-addr :9090 --http-path /mcp --organization my-org --key-file /path/to/key-file.json --pulsar-instance my-instance --pulsar-cluster my-cluster

# Start MCP server with SSE and external Kafka
bin/snmcp sse --http-addr :9090 --http-path /mcp --use-external-kafka --kafka-bootstrap-servers localhost:9092

# Start MCP server with SSE and external Pulsar
bin/snmcp sse --http-addr :9090 --http-path /mcp --use-external-pulsar --pulsar-web-service-url http://pulsar.example.com:8080

# Start MCP server with SSE by docker with StreamNative Cloud authentication
docker run -i --rm -e SNMCP_ORGANIZATION=my-org -e SNMCP_KEY_FILE=/key.json -v /path/to/key-file.json:/key.json -p 9090:9090 streamnative/snmcp sse

多会话脉冲星模式(仅限SSE)

当使用外部Pulsar运行SSE服务器时,您可以启用 多会话模式 以支持每个用户的身份验证。在此模式下,每个HTTP请求都必须包含 Authorization: Bearer <token> 标头,服务器将为每个唯一令牌创建单独的Pulsar会话。

# Start SSE server with multi-session Pulsar mode
bin/snmcp sse --http-addr :9090 --http-path /mcp \
  --use-external-pulsar \
  --pulsar-web-service-url http://pulsar.example.com:8080 \
  --multi-session-pulsar \
  --session-cache-size 100 \
  --session-ttl-minutes 30

主要特点:

  • 每个用户会话:每个用户的Pulsar令牌创建一个单独的会话
  • LRU缓存:当缓存已满时,会话将通过LRU驱逐进行缓存
  • 基于TTL的清理:配置TTL后,空闲会话会自动清理
  • 严格认证:没有有效的请求 Authorization 标头接收HTTP 401未经授权

身份验证流程:

  1. 客户端通过以下方式连接到SSE端点 Authorization: Bearer <pulsar-jwt-token> 头球
  2. 服务器通过尝试创建Pulsar会话来验证令牌
  3. 如果有效,会话将被缓存并重新用于后续请求
  4. 如果无效或缺失,服务器将返回HTTP 401 Unauthorized

配置选项:

标志 默认值 描述
--multi-session-pulsar false 启用每个用户的Pulsar会话
--session-cache-size 100 缓存会话的最大数量
--session-ttl-minutes 30 驱逐前会话空闲超时

注: 多会话模式仅适用于外部Pulsar模式(--use-external-pulsar)并且仅适用于SSE服务器,不适用于stdio。

命令行选项

Usage:
  bin/snmcp [command]

Available Commands:
  stdio       Start stdio server
  sse         Start sse server
  help        Help about any command

Flags:
      --audience string                        The audience identifier for the API server (default "https://api.streamnative.cloud")
      --client-id string                       The client ID to use for authorization grants (default "AJYEdHWi9EFekEaUXkPWA2MqQ3lq1NrI")
      --config-dir string                      If present, the config directory to use
      --enable-command-logging                 When enabled, the server will log all command requests and responses to the log file
      --features strings                       Features to enable, defaults to `all`
  -h, --help                                   help for bin/snmcp
      --issuer string                          The OAuth 2.0 issuer endpoint (default "https://auth.streamnative.cloud/")
      --kafka-auth-mechanism string            The auth mechanism to use for Kafka
      --kafka-auth-pass string                 The auth password to use for Kafka
      --kafka-auth-type string                 The auth type to use for Kafka
      --kafka-auth-user string                 The auth user to use for Kafka
      --kafka-bootstrap-servers string         The bootstrap servers to use for Kafka
      --kafka-ca-file string                   The CA file to use for Kafka
      --kafka-client-cert-file string          The client certificate file to use for Kafka
      --kafka-client-key-file string           The client key file to use for Kafka
      --kafka-schema-registry-auth-pass string The auth password to use for the schema registry
      --kafka-schema-registry-auth-user string The auth user to use for the schema registry
      --kafka-schema-registry-bearer-token string The bearer token to use for the schema registry
      --kafka-schema-registry-url string       The schema registry URL to use for Kafka
      --key-file string                        The key file to use for authentication to StreamNative Cloud
      --log-file string                        Path to log file
      --organization string                    The organization to use for the API server
      --proxy-location string                  The proxy location to use for the API server (default "https://proxy.streamnative.cloud")
      --pulsar-auth-params string              The auth params to use for Pulsar
      --pulsar-auth-plugin string              The auth plugin to use for Pulsar
      --pulsar-token string                    The token to use for Pulsar
      --pulsar-cluster string                  The default cluster to use for the API server
      --pulsar-instance string                 The default instance to use for the API server
      --pulsar-tls-allow-insecure-connection   The TLS allow insecure connection to use for Pulsar
      --pulsar-tls-cert-file string            The TLS cert file to use for Pulsar
      --pulsar-tls-enable-hostname-verification The TLS enable hostname verification to use for Pulsar (default true)
      --pulsar-tls-key-file string             The TLS key file to use for Pulsar
      --pulsar-tls-trust-certs-file-path string The TLS trust certs file path to use for Pulsar
      --pulsar-web-service-url string          The web service URL to use for Pulsar
  -r, --read-only                              Read-only mode
      --server string                          The server to connect to (default "https://api.streamnative.cloud")
      --use-external-kafka                     Use external Kafka
      --use-external-pulsar                    Use external Pulsar
      --http-addr string                       HTTP server address (default ":9090")
      --http-path string                       HTTP server path for SSE endpoint (default "/mcp")
      --multi-session-pulsar                   Enable per-user Pulsar sessions based on Authorization header tokens (only for external Pulsar mode)
      --session-cache-size int                 Maximum number of cached Pulsar sessions when multi-session is enabled (default 100)
      --session-ttl-minutes int                Session TTL in minutes before eviction when multi-session is enabled (default 30)
  -v, --version                                version for bin/snmcp

工具配置

StreamNative MCP服务器支持通过以下方式启用或禁用特定功能组 --features 旗帜。这允许您控制哪些MCP工具可用于您的AI工具。仅启用所需的工具集可以帮助LLM进行工具选择并减少上下文大小。

可用特征

StreamNative MCP服务器允许您使用 --features 旗帜。这有助于您控制AI代理可用的工具,并可以减少LLM的上下文大小。

组合特征集

特性 描述
all 启用所有功能:StreamNative Cloud、Pulsar和Kafka工具

Claude连接器兼容性:以前将读写操作混合在一起的管理工具 operation 例如,参数作为单独的读/写MCP工具公开 kafka_admin_topics_read 和 kafka_admin_topics_write.阅读工具包括 annotations.readOnlyHint=true;写作或副作用工具包括 annotations.destructiveHint=true.In --read-only 模式,写入/破坏工具未注册。

Kafka特性

功能 描述 文档
all-kafka 启用所有Kafka管理和客户端工具,不使用Apache Pulsar和StreamNative Cloud工具
kafka-admin Kafka管理操作(所有管理工具)
kafka-client Kafka客户端操作(生产/消费) kafka_client_consume.md, kafka_客户端_生产.md
kafka-admin-topics 管理Kafka主题 kafka_admin_topics.md
kafka-admin-partitions 管理Kafka分区 卡夫卡_行政区划.md
kafka-admin-groups 管理Kafka消费者组 kafka_admin_groups.md
kafka-admin-schema-registry 与Kafka模式注册表交互 kafka_admin_schema_register.md
kafka-admin-connect 管理Kafka Connect连接器 kafka_admin_connect.md

Pulsar特性

功能 描述 文档
all-pulsar 启用所有Pulsar管理和客户端工具,不使用Apache Kafka和StreamNative Cloud工具 脉冲资源.md
pulsar-admin Pulsar管理操作(所有管理工具) 脉冲资源.md
pulsar-client Pulsar客户端操作(生产/消费) pulser_client_consume.md, pulser_client_produce.md
pulsar-admin-brokers 管理Pulsar经纪人 pulser_admin_brokers.md
pulsar-admin-brokers-status 检查Pulsar代理或代理状态 pulser_admin_status.md
pulsar-admin-broker-stats 访问Pulsar经纪人统计数据 pulser_admin_broker_stats.md
pulsar-admin-clusters 管理Pulsar集群 pulser_admin_clusters.md
pulsar-admin-functions-worker 管理Pulsar功能人员 pulser_admin_functions_worker.md
pulsar-admin-namespaces 管理Pulsar命名空间 pulser_admin_namespaces.md
pulsar-admin-namespace-policy 配置Pulsar命名空间策略 pulser_admin_namespace_policy.md
pulsar-admin-ns-isolation-policy 管理命名空间隔离策略 脉冲管理隔离策略.md
pulsar-admin-packages 管理Pulsar软件包 pulser_admin_packages.md
pulsar-admin-resource-quotas 配置资源配额 pulser_admin_resource_quotas.md
pulsar-admin-schemas 管理Pulsar模式 pulser_adm_schemas.md
pulsar-admin-subscriptions 管理Pulsar订阅 pulser_admin_订阅.md
pulsar-admin-tenants 管理Pulsar租户 pulser_admin_tenants.md
pulsar-admin-topics 管理Pulsar主题 pulser_adm_topics.md
pulsar-admin-sinks 管理Pulsar IO接收器 pulser_adm_links.md
pulsar-admin-functions 管理Pulsar功能 pulser_admin_functions.md
pulsar-admin-sources 管理Pulsar源 pulser_admin_sources.md
pulsar-admin-topic-policy 配置Pulsar主题策略 pulser_admin_topic_policy.md

Pulsar管理功能门还为匹配的管理界面注册只读MCP资源。这些资源使用 pulsar://... URI,返回JSON快照,并与可写工具分开;看见 脉冲资源.md 对于支持的URI模板和安全边界。


StreamNative云功能

功能 描述 文档
streamnative-cloud 管理StreamNative Cloud上下文并检查资源日志 streamnative_cloud.md
functions-as-tools 将部署的Pulsar函数动态公开为可调用的MCP工具,并具有自动输入/输出模式处理功能。 函数_as_tools.md

注: 使用时 --pulsar-instance 和 --pulsar-cluster 标记在一起,上下文变异工具(sncloud_context_use_cluster, sncloud_context_reset)由于上下文是预先配置的,因此会自动禁用。

您可以根据需要使用 --features 旗帜。例如,要仅启用Pulsar客户端功能:

# Enable only Pulsar client features
bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json --features pulsar-client

检查MCP服务器

您可以使用 @模型上下文协议/检查器 用于检查和测试MCP服务器的工具。这对于调试和验证服务器的配置特别有用。

安装

npm install -g @modelcontextprotocol/inspector

用法

# Inspect a stdio server
mcp-inspector stdio --command "bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json"

# Inspect an SSE server
mcp-inspector sse --url "http://localhost:9090/mcp"

检查器提供了一个web界面,您可以在其中:

  • 查看可用工具及其模式
  • 测试工具调用
  • 监控服务器响应
  • 调试连接问题

与MCP客户端集成

此服务器可以与任何兼容MCP的客户端一起使用,例如:

  • 克劳德桌面版
  • 其他支持MCP协议的AI助手
  • 使用MCP客户端库构建的自定义应用程序

⚠️ 提醒:请确保您与LLM提供商有一个有效的付费计划,以充分利用MCP服务器。 没有它,您可能会遇到错误: message will exceed the length limit for this chat.

使用Claude Desktop

使用stdio服务器

{
  "mcpServers": {
    "mcp-streamnative": {
      "command": "${PATH_TO_SNMCP}/bin/snmcp",
      "args": [
        "stdio",
        "--organization",
        "${STREAMNATIVE_CLOUD_ORGANIZATION_ID}",
        "--key-file",
        "${STREAMNATIVE_CLOUD_KEY_FILE}"
      ]
    }
  }
}

请记得更换 ${PATH_TO_SNMCP} 与实际路径 snmcp 二进制和 ${STREAMNATIVE_CLOUD_ORGANIZATION_ID} 和 ${STREAMNATIVE_CLOUD_KEY_FILE} 分别使用您的StreamNative Cloud组织ID和密钥文件路径。

如果您有以下选择,您可以使用docker镜像启动stdio服务器 码头工人 安装。

{
  "mcpServers": {
    "mcp-streamnative": {
      "command": "docker",
      "args": [
        "run",
        "-i",
        "--rm",
        "-e",
        "SNMCP_ORGANIZATION",
        "-e",
        "SNMCP_KEY_FILE",
        "-v",
        "${STREAMNATIVE_CLOUD_KEY_FILE}:/key.json",
        "streamnative/snmcp",
        "stdio"
      ],
      "env": {
        "SNMCP_ORGANIZATION": "${STREAMNATIVE_CLOUD_ORGANIZATION_ID}",
        "SNMCP_KEY_FILE": "/key.json"
      }
    }
  }
}

使用SSE服务器

首先,安装mcp代理工具:

pip install mcp-proxy

然后配置Claude Desktop以使用SSE服务器:

{
  "mcpServers": {
    "mcp-streamnative-proxy": {
      "command": "mcp-proxy",
      "args": [
        "http://localhost:9090/mcp/sse"
      ]
    }
  }
}

注意:如果mcp代理不在系统PATH中,则需要提供可执行文件的完整路径。例如:

  • 在macOS上: /Library/Frameworks/Python.framework/Versions/3.11/bin/mcp-proxy
  • 在Linux上: /usr/local/bin/mcp-proxy
  • 在Windows上: C:\Python311\Scripts\mcp-proxy.exe

请记得更换 http://localhost:9090/mcp/sse 使用正确的URL。

关于模型上下文协议(MCP)

模型上下文协议(MCP)是一种开放协议,它规范了应用程序如何向LLM提供上下文。MCP通过提供以下功能,帮助在LLM之上构建代理和复杂的工作流程:

  • LLM可以直接插入的预构建集成列表越来越多
  • 在LLM提供商和供应商之间切换的灵活性
  • 在基础架构中保护数据的最佳实践

有关更多信息,请访问 模型上下文协议.io.

发布

本节介绍如何发布新版本的 snmcp.

  1. 为新版本生成标记(请参阅下面的版本控制):
git tag -a v0.0.1 -m "v0.0.1"
  1. 将标签推送到git仓库:
git push origin refs/tags/v0.0.1

发布工作流程将:

  • 为支持的平台构建Go二进制文件
  • 将二进制文件存档
  • 将发布发布发布到github存储库(参考)

版本控制

此项目使用 森伯 语义。

  • 稳定: vX.Y.Z
  • 预发布: vX.Y.Z-rc.W
  • 快照: vX.Y.Z-SNAPSHOT-commit

许可证

根据Apache许可证版本2.0许可:http://www.apache.org/licenses/LICENSE-2.0