流原生-mcp服务器

Name: 流原生-mcp服务器
Author: streamnative

作者：streamnative18 星标更新：2025-11-20

安装使用来源：spider==github.com

项目介绍

StreamNative MCP Server

用于将AI代理与StreamNative云资源及Apache Kafka/Pulsar消息系统集成的模型上下文协议（MCP）服务器。

概述

StreamNative MCP Server 提供了一个标准接口，使大型语言模型（LLMs）和AI代理能够与StreamNative云服务、Apache Kafka和Apache Pulsar进行交互。此实现遵循了 Model Context Protocol 规范，使得AI应用可以通过标准化接口访问消息服务。

特性

StreamNative云集成：
- 使用身份验证连接到StreamNative云资源
- 切换到组织中可用的集群
- 描述集群资源的状态
Apache Kafka支持：与Apache Kafka资源交互包括：
- Kafka管理操作（主题、分区、消费者组）
- 架构注册表操作
- Kafka连接器操作（*）
- Kafka客户端操作（生产者、消费者）
Apache Pulsar支持：与Apache Pulsar资源交互包括：
- Pulsar管理操作（主题、命名空间、租户、模式等）
- Pulsar客户端操作（生产者、消费者）
- 函数、源和接收器管理
多种连接选项：
- 使用服务账户认证连接到StreamNative云
- 直接连接到外部Apache Kafka集群
- 直接连接到外部Apache Pulsar集群

*：Kafka连接器操作仅在StreamNative云上进行了测试和验证。

安装

Homebrew（macOS和Linux）

安装streamnative-mcp-server最简单的方法是使用Homebrew：

# 添加tap仓库
brew tap streamnative/streamnative

# 安装streamnative-mcp-server
brew install streamnative/streamnative/snmcp

Docker镜像

StreamNative MCP Server发布了Docker镜像到 streamnative/snmcp，可以使用docker命令运行标准IO服务器和SSE服务器。

# 从Docker Hub拉取镜像
docker pull streamnative/snmcp

从GitHub发布

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

从源代码

# 克隆仓库
git clone https://github.com/streamnative/streamnative-mcp-server.git
cd streamnative-mcp-server

go mod tidy
go mod download

# 构建二进制文件
make

使用方法

前提条件

如果您想访问您的StreamNative云，您需要准备好以下资源：

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

启动MCP服务器

使用标准IO服务器

# 使用StreamNative云认证启动MCP服务器
bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json

# 使用预配置的上下文启动MCP服务器
# 当提供--pulsar-instance和--pulsar-cluster时，上下文管理工具被禁用
bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json --pulsar-instance my-instance --pulsar-cluster my-cluster

# 使用外部Kafka启动MCP服务器
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

# 使用外部Pulsar启动M
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"

# 使用Docker通过StreamNative云认证启动MCP服务器
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（服务器发送事件）服务器

# 使用SSE和StreamNative云认证启动MCP服务器
snmcp sse --http-addr :9090 --http-path /mcp --organization my-org --key-file /path/to/key-file.json

# 使用SSE和预配置的StreamNative云上下文启动MCP服务器
# 当提供--pulsar-instance和--pulsar-cluster时，上下文管理工具被禁用
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

# 使用SSE和外部Kafka启动MCP服务器
snmcp sse --http-addr :9090 --http-path /mcp --use-external-kafka --kafka-bootstrap-servers localhost:9092

# 使用SSE和外部Pulsar启动MCP服务器
snmcp sse --http-addr :9090 --http-path /mcp --use-external-pulsar --pulsar-web-service-url http://pulsar.example.com:8080

# 使用Docker通过StreamNative云认证启动MCP服务器
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

命令行选项

用法：
  snmcp [命令]

可用命令：
  stdio       启动标准IO服务器
  help        关于任何命令的帮助

标志：
      --audience string                        API服务器的受众标识符（默认值："https://api.streamnative.cloud"）
      --client-id string                       授权授予使用的客户端ID（默认值："AJYEdHWi9EFekEaUXkPWA2MqQ3lq1NrI"）
      --config-dir string                      如果存在，使用的配置目录
      --enable-command-logging                 当启用时，服务器将把所有命令请求和响应记录到日志文件中
      --features strings                       要启用的功能，默认为`all`
  -h, --help                                   snmcp的帮助信息
      --issuer string                          OAuth 2.0颁发者端点（默认值："https://auth.streamnative.cloud/"）
      --kafka-auth-mechanism string            用于Kafka的身份验证机制
      --kafka-auth-pass string                 用于Kafka的身份验证密码
      --kafka-auth-type string                 用于Kafka的身份验证类型
      --kafka-auth-user string                 用于Kafka的身份验证用户
      --kafka-bootstrap-servers string         用于Kafka的引导服务器
      --kafka-ca-file string                   用于Kafka的CA文件
      --kafka-client-cert-file string          用于Kafka的客户端证书文件
      --kafka-client-key-file string           用于Kafka的客户端密钥文件
      --kafka-schema-registry-auth-pass string 用于架构注册表的身份验证密码
      --kafka-schema-registry-auth-user string 用于架构注册表的身份验证用户
      --kafka-schema-registry-bearer-token string 用于架构注册表的承载令牌
      --kafka-schema-registry-url string       用于Kafka的架构注册表URL
      --key-file string                        用于身份验证到StreamNative云的密钥文件
      --log-file string                        日志文件路径
      --organization string                    API服务器使用的组织
      --proxy-location string                  API服务器使用的代理位置（默认值："https://proxy.streamnative.cloud"）
      --pulsar-auth-params string              用于Pulsar的身份验证参数
      --pulsar-auth-plugin string              用于Pulsar的身份验证插件
      --pulsar-token string                    用于Pulsar的令牌
      --pulsar-cluster string                  API服务器使用的默认集群
      --pulsar-instance string                 API服务器使用的默认实例
      --pulsar-tls-allow-insecure-connection   用于Pulsar的TLS允许不安全连接
      --pulsar-tls-cert-file string            用于Pulsar的TLS证书文件
      --pulsar-tls-enable-hostname-verification 用于Pulsar的TLS启用主机名验证（默认值：true）
      --pulsar-tls-key-file string             用于Pulsar的TLS密钥文件
      --pulsar-tls-trust-certs-file-path string 用于Pulsar的TLS信任证书文件路径
      --pulsar-web-service-url string          用于Pulsar的Web服务URL
  -r, --read-only                              只读模式
      --server string                          连接到的服务器（默认值："https://api.streamnative.cloud"）
      --use-external-kafka                     使用外部Kafka
      --use-external-pulsar                    使用外部Pulsar
      --http-addr string                       HTTP服务器地址（默认值：":9090"）
      --http-path string                       SSE端点的HTTP服务器路径（默认值："/mcp"）
  -v, --version                                snmcp的版本信息

工具配置

StreamNative MCP Server支持通过--features标志启用或禁用特定功能组。这允许您控制哪些MCP工具对您的AI工具可用。仅启用所需的工具集可以帮助LLM选择工具并减少上下文大小。

可用功能

StreamNative MCP Server允许您使用--features标志启用或禁用特定功能组。这有助于您控制哪些工具对您的AI代理可用，并可以减少LLM的上下文大小。

组合功能集

功能	描述
`all`	启用所有功能：StreamNative云、Pulsar和Kafka工具

Kafka功能

功能	描述	文档
`all-kafka`	启用所有Kafka管理工具和客户端工具，不包括Apache Pulsar和StreamNative云工具
`kafka-admin`	Kafka管理操作（所有管理工具）
`kafka-client`	Kafka客户端操作（生产/消费）	kafka_client_consume.md, kafka_client_produce.md
`kafka-admin-topics`	管理Kafka主题	kafka_admin_topics.md
`kafka-admin-partitions`	管理Kafka分区	kafka_admin_partitions.md
`kafka-admin-groups`	管理Kafka消费者组	kafka_admin_groups.md
`kafka-admin-schema-registry`	与Kafka架构注册表交互	kafka_admin_schema_registry.md
`kafka-admin-connect`	管理Kafka连接器	kafka_admin_connect.md

Pulsar功能

功能	描述	文档
`all-pulsar`	启用所有Pulsar管理工具和客户端工具，不包括Apache Kafka和StreamNative云工具
`pulsar-admin`	Pulsar管理操作（所有管理工具）
`pulsar-client`	Pulsar客户端操作（生产/消费）	pulsar_client_consume.md, pulsar_client_produce.md
`pulsar-admin-brokers`	管理Pulsar代理	pulsar_admin_brokers.md
`pulsar-admin-broker-stats`	访问Pulsar代理统计信息	pulsar_admin_broker_stats.md
`pulsar-admin-clusters`	管理Pulsar集群	pulsar_admin_clusters.md
`pulsar-admin-functions-worker`	管理Pulsar函数工作者	pulsar_admin_functions_worker.md
`pulsar-admin-namespaces`	管理Pulsar命名空间	pulsar_admin_namespaces.md
`pulsar-admin-namespace-policy`	配置Pulsar命名空间策略	pulsar_admin_namespace_policy.md
`pulsar-admin-isolation-policy`	管理命名空间隔离策略	pulsar_admin_nsisolationpolicy.md
`pulsar-admin-packages`	管理Pulsar包	pulsar_admin_packages.md
`pulsar-admin-resource-quotas`	配置资源配额	pulsar_admin_resource_quotas.md
`pulsar-admin-schemas`	管理Pulsar模式	pulsar_admin_schemas.md
`pulsar-admin-subscriptions`	管理Pulsar订阅	pulsar_admin_subscriptions.md
`pulsar-admin-tenants`	管理Pulsar租户	pulsar_admin_tenants.md
`pulsar-admin-topics`	管理Pulsar主题	pulsar_admin_topics.md
`pulsar-admin-sinks`	管理Pulsar IO接收器	pulsar_admin_sinks.md
`pulsar-admin-functions`	管理Pulsar函数	pulsar_admin_functions.md
`pulsar-admin-sources`	管理Pulsar源	pulsar_admin_sources.md
`pulsar-admin-topic-policy`	配置Pulsar主题策略	pulsar_admin_topic_policy.md

StreamNative云功能

功能	描述	文档
`streamnative-cloud`	管理StreamNative云上下文并检查资源日志	streamnative_cloud.md
`functions-as-tools`	动态暴露已部署的Pulsar函数作为可调用的MCP工具，自动处理输入/输出模式。	functions_as_tools.md

注意：当一起使用--pulsar-instance和--pulsar-cluster标志时，上下文管理工具（sncloud_context_use_cluster）会自动禁用，因为上下文已经预先配置好了。

您可以根据需要组合这些功能。例如，仅启用Pulsar客户端功能：

# 仅启用Pulsar客户端功能
bin/snmcp stdio --organization my-org --key-file /path/to/key-file.json --features pulsar-client

检查MCP服务器

您可以使用@modelcontextprotocol/inspector工具来检查和测试您的MCP服务器。这对于调试和验证服务器配置特别有用。

安装

npm install -g @modelcontextprotocol/inspector

使用

# 检查标准IO服务器
mcp-inspector stdio --command "snmcp stdio --organization my-org --key-file /path/to/key-file.json"

# 检查SSE服务器
mcp-inspector sse --url "http://localhost:9090/mcp"

检查器提供了一个Web界面，您可以在其中：

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

与MCP客户端集成

此服务器可用于任何兼容MCP的客户端，如：

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

⚠️ 提醒：请确保您拥有LLM提供商的有效付费计划以充分利用MCP服务器。没有它，您可能会遇到错误：“消息将超过此聊天的长度限制”。

与Claude桌面的使用

使用标准IO服务器

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