技术文档 | 使用 Spring AI 实现一个简单的 Pulsar MCP Server

本文来自 Pulsar 社区贡献者 Sinan Liu 👏

✨如果你也有关于 Pulsar 的技术解析、实践应用、花式玩法等内容想与社区分享，欢迎向我们投稿或推荐内容~

✨欢迎添加 Pulsarbot（微信号：pulsarbot），加入 Pulsar+AI 交流群

大家好，我是 sinan。Github ID: Denovo1998。

作为一名开发者兼系统维护者，你是否也曾被 pulsar-admin 的命令行参数劝退过？创建租户、设置权限、再创建分区 Topic……一套操作下来很容易出错。

想象一下这个场景：

你只需要对着一个聊天框，用大白话输入：

“帮我在 cluster-a 上创建一个叫 my-tenant 的租户，管理员是 admin-role”

或

“往 my-tenant/my-ns/my-topic 这个 Topic 里发一条内容是 hello world 的消息”

然后，Pulsar 集群里的一切就都自动帮你搞定了(๑•̀ㅂ•́)و✧

是不是感觉非常 nice？

通过本文，我将给大家展示一个基于 Spring AI 和 Pulsar 的 MCP（Model Control Protocol）服务器。它能将你的自然语言指令精准地转化为对 Pulsar 集群的实际操作。

🚀 一起看看如何实现吧！

1 宏观架构：AI 如何指挥 Pulsar？

在深入代码细节之前，我们必须先建立一个全局视野。这个项目的核心思想其实非常简单：用 AI 作为翻译官，将自然语言翻译成机器（Pulsar）能听懂的 API 调用。

下面这张架构图清晰地展示了系统的四大组成部分和一次典型的交互流程：

1. MCP 客户端 (Client)：这是用户入口，可以是一个聊天界面或者任何兼容 MCP 协议的程序。用户在这里输入自然语言指令，例如“帮我创建一个 Topic”。

2. 我们的应用 (Pulsar MCP Server)：这是整个架构的核心，它本身又分为三层：

- Spring AI MCP Framework：这是由 Spring AI 提供的标准框架。它负责对外暴露 SSE 端点，处理 MCP 协议，与 LLM 进行通信，并根据 LLM 的决策来编排和调用我们定义的工具。我们无需关心这部分的复杂实现。

- Pulsar Tools (@Tool Beans)：这是我们需要编写的核心代码。这里包含了 PulsarTenantTools、PulsarTopicTools 等，它们将具体业务操作（如创建租户）封装成一个个带有 @Tool 注解的方法，是 AI 可调用的“能力清单”。

- Pulsar Admin/Client SDKs：这是 Apache Pulsar 官方提供的 Java 客户端库，是我们的工具最终与 Pulsar 集群交互的桥梁。

3. 大语言模型 (LLM)：这是一个外部的 AI 服务。它的职责是“理解”用户的意图和我们工具的描述，然后做出决策——应该调用哪个工具，以及传入什么参数。

4. Pulsar 集群 (Pulsar Cluster)：这是我们最终要操作的目标系统。

步骤（1-4）清晰地展示了一次完整的请求生命周期：用户的指令通过我们的应用，交由 LLM 进行“翻译”，翻译结果（工具调用）再由我们的应用执行，最终作用于 Pulsar 集群。

接下来，让我们进入技术核心区，看看代码层面的实现过程。

2 技术内核：三步构建 Pulsar 的 AI 大脑

第 1 步：项目基石与依赖

万丈高楼平地起，我们首先需要一个标准的 Spring Boot 项目，并引入几个关键依赖。

<!-- pom.xml --><dependencies>    <!-- 核心：Spring AI MCP Server WebFlux Starter -->    <dependency>        <groupId>org.springframework.ai</groupId>        <artifactId>spring-ai-starter-mcp-server-webflux</artifactId>    </dependency>
    <!-- Pulsar 操作库：Admin Client & 普通 Client -->    <dependency>        <groupId>org.apache.pulsar</groupId>        <artifactId>pulsar-client-admin</artifactId>        <version>${pulsar.version}</version>    </dependency>    <dependency>        <groupId>org.apache.pulsar</groupId>        <artifactId>pulsar-client</artifactId>        <version>${pulsar.version}</version>    </dependency>      <!-- 其他 Spring Boot 相关依赖 --></dependencies>

剖析 (Why):

* spring-ai-starter-mcp-server-webflux：这是 Spring AI 提供的 MCP 服务器启动器，它处理了与 AI 模型通信的所有底层协议和复杂性，我们只需要专注于实现“工具”即可。

* pulsar-client-admin 和 pulsar-client：这是我们实际操作 Pulsar 的“手和脚”，分别用于执行管理命令和收发消息。

第 2 步：核心魔法 - 定义 AI 工具 (@Tool)

这是整个项目最有趣、也是最核心的部分。我们通过编写普通的 Java 方法，并附加上 @Tool 注解，来赋予 AI 新的能力。

让我们以“创建租户”这个功能为例，看看 PulsarTenantTools.java 是如何实现的：

// src/main/java/org/apache/pulsar/mcp/server/tools/PulsarTenantTools.javapackage org.apache.pulsar.mcp.server.tools;
// ... importsimport org.springframework.ai.tool.annotation.Tool;import org.springframework.ai.tool.annotation.ToolParam;import org.springframework.stereotype.Service;
@Servicepublic class PulsarTenantTools {    private final PulsarAdmin pulsarAdmin;    // ... constructor
    @Tool(description = "Create a new tenant with specified configuration")    public String createTenant(            @ToolParam(description = "The tenant name") String tenant,            @ToolParam(description = "List of admin roles") List<String> adminRoles,            @ToolParam(description = "List of allowed clusters") Set<String> allowedClusters    ) {        try {            TenantInfo tenantInfo = TenantInfo.builder()                    .adminRoles(Set.copyOf(adminRoles))                    .allowedClusters(allowedClusters)                    .build();            pulsarAdmin.tenants().createTenant(tenant, tenantInfo);            return "Successfully created tenant: " + tenant;        } catch (PulsarAdminException e) {            return "Error creating tenant: " + e.getMessage();        }    }      // ... 其他方法，如 listTenants, deleteTenant 等}

剖析 (Why):

* @Tool(description = "...")：这是关键中的关键！ 这里的描述文字不是给程序员看的，而是直接提供给 LLM 的。LLM 会根据这个描述来理解这个方法能做什么。描述得越清晰、越准确，AI 调用它的成功率就越高。

* @ToolParam(description = "...")：同理，这是对方法参数的描述。AI 会根据这个描述来决定如何从用户的自然语言指令中提取参数值。

• 方法体：方法体内部就是我们再熟悉不过的 pulsarAdmin API 调用。Spring AI 优雅地将 AI 的“决策”和我们的“执行”解耦了。

同样地，我们也可以为 PulsarTopicTools（管理 Topic）和 PulsarProduceTools（发送消息）定义类似的工具。

第 3 步：组装与配置

定义好工具后，我们需要一个地方把它们“注册”到 Spring AI 的体系中。

// src/main/java/org/apache/pulsar/mcp/server/config/PulsarMcpToolConfiguration.javapackage org.apache.pulsar.mcp.server.config;
// ... importsimport org.springframework.ai.tool.ToolCallbackProvider;import org.springframework.ai.tool.method.MethodToolCallbackProvider;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;
@Configurationpublic class PulsarMcpToolConfiguration {    @Bean    public ToolCallbackProvider pulsarAdminTools(            PulsarTenantTools tenantTools,            PulsarTopicTools topicTools    ) {        // 将所有 Admin 相关的工具类注册        return MethodToolCallbackProvider.builder()                .toolObjects(					tenantTools,					topicTools                )                .build();    }      @Bean    public ToolCallbackProvider pulsarClientTools(            PulsarProduceTools produceTools    ) {        // 将 Client 相关的工具类注册        return MethodToolCallbackProvider.builder()                .toolObjects(produceTools)                .build();    }}

剖析 (Why):

* MethodToolCallbackProvider：你可以把它理解为一个“工具扫描器”。它会自动扫描你提供的 toolObjects（比如 tenantTools 实例）中所有被 @Tool 注解标记的公共方法，并将它们转换为 AI 可以调用的标准格式。

最后，在 application.yml 中配置 Pulsar 的连接信息：

# src/main/resources/application.ymlserver:  port: 8888
pulsar:  admin:    url: http://localhost:8080  service:    url: pulsar://localhost:6650
# ... mcp server 相关配置```
现在，万事俱备！只需一行命令即可启动：
```bash# 确保你的 Pulsar 集群已经运行mvn spring-boot:run

现在，万事俱备！只需一行命令即可启动：

# 确保你的 Pulsar 集群已经运行mvn spring-boot:run

服务启动后，你就可以通过兼容 MCP 协议的客户端（如 Cherry-Studio，Cline 等）连接到 http://localhost:8888，开始用自然语言操控你的 Pulsar 集群了！

3 来吧，展示！

客户端（Chat UI）配置 MCP Server

上一章我们启动了 MCP Server，这一章我们就使用 cherry-studio 来使用一下这个简单的 MCP Server。

如下进行配置并保存。

你还能看到可使用的 Tools：

配置 Pulsar Cluster

我们直接在 Pulsar 官网(https://pulsar.apache.org/docs/4.0.x/getting-started-docker/) 找到启动 Pulsar 的 最简单方式。

docker run -it \  -p 6650:6650 \  -p 8080:8080 \  --mount source=pulsardata,target=/pulsar/data \  --mount source=pulsarconf,target=/pulsar/conf \  apachepulsar/pulsar:(这里填写你的Pulsar集群版本号) \  bin/pulsar standalone

启动成功后开始测试~

简单使用

在 cherry-studio 中创建一个新的助手并选中我们创建的 pulsar-mcp-server。

我们先用 createTopic 工具创建一个新的 Topic

然后我们用 listTopics 工具测试一下 Topic 有没有创建成功

最后测试一下 Topic 是否真的创建成功

大功告成！

4 开源驱动，欢迎共建

👍 Talk is cheap, show me the code!

这个项目的完整代码已经开源在 GitHub 上。它目前虽然只是一个简单的原型，但展示了 AI 与基础软件结合的巨大潜力。

GitHub:

https://github.com/Denovo1998/pulsar-mcp-server

我设计了灵活的 mcp.features 开关，你可以很方便地在 application.yml 中启用或禁用某一组功能，这为后续的功能扩展打下了良好的基础。欢迎大家 Star、Fork，更欢迎提交 PR，一起把它打造成一个真正强大的 Pulsar AI 助手！

5

延伸：为什么是 Pulsar？

你可能会问，为什么选择 Pulsar 来做这次的 AI 实验？这背后其实有我的架构思考。在云原生时代，Pulsar 相比于 Kafka/RocketMQ 等传统消息系统，展现出了几个极具吸引力的特质：

1. 真正的云原生架构：Pulsar 的存储计算分离架构（Broker 无状态，BookKeeper 负责存储）带来了无与伦比的弹性和扩展性，这与 AI 应用对资源弹性需求不谋而合。

2. 强大的多租户能力：Pulsar 原生支持多租户，可以从逻辑上完美隔离不同业务线或团队的资源。这使得我们用 AI 创建和管理租户（createTenant）的操作变得极具现实意义。

3. 统一消息模型：Pulsar 在一套平台内同时支持流（Streaming）和队列（Queueing）两种模型，避免了企业内部维护多套消息系统的复杂性。一个 AI 助手就能统一管理所有消息场景，价值巨大。

将 AI 的智能决策能力与 Pulsar 强大的底层架构相结合，无疑会为未来的数据平台运维和管理带来革命性的变化。