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，你需要准备好以下资源：

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

启动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 标头，服务器将为每个唯一令牌创建单独的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未经授权

身份验证流程：

客户端通过以下方式连接到SSE端点 `Authorization: Bearer

` 头球

服务器通过尝试创建Pulsar会话来验证令牌
如果有效，会话将被缓存并重新用于后续请求
如果无效或缺失，服务器将返回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工具

______________________________________________________________________

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软件包
`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._

为新版本生成标记（请参阅下面的版本控制）：

git tag -a v0.0.1 -m "v0.0.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

Streamnative MCP Server

快速接入

详细介绍

StreamNative MCP服务器

概述

特性

安装

Homebrew（macOS和Linux）

Docker镜像

Helm Chart（Kubernetes）

Github发布

源自

用法

先决条件

启动MCP服务器

使用stdio服务器

使用SSE（服务器发送事件）服务器

多会话脉冲星模式（仅限SSE）

命令行选项

工具配置

可用特征

组合特征集

Kafka特性

Pulsar特性

StreamNative云功能

检查MCP服务器

安装

用法

与MCP客户端集成

使用Claude Desktop

使用stdio服务器

使用SSE服务器

关于模型上下文协议（MCP）

发布

版本控制

许可证

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP