由 Anthropic 推出的模型上下文协议（MCP）是一项新标准，它通过提供一种安全、一致的方式将 AI 智能体与外部工具和数据源连接起来，从而简化了人工智能（AI）集成。

当我们看到 MCP 的潜力时，我们立即开始探索如何将实时数据流引入其中。凭借我们长期以来对开源和开放标准的支持，构建一个 MCP 服务器是自然而然的选择。

通过这个服务器，我们实现了两件大事：

• 我们让 AI 智能体可以直接访问新鲜的实时数据，确保它们始终基于最新信息运行。

• 我们通过自然语言让管理 Confluent 变得更加容易，使用户能够配置主题、执行 Flink SQL，并与他们的数据基础设施交互，而无需应对复杂命令。

这篇博客文章将深入介绍我们构建了什么、它如何工作，以及为什么实时数据对 AI 智能体的未来至关重要。

什么是模型上下文协议？

如今，将 AI 系统与不同平台集成通常需要定制构建的解决方案，这些方案复杂、耗时且难以维护。开发者经常需要编写专门代码来将数据拉取到 AI 智能体中，手动集成 API（应用程序编程接口）、数据库和外部工具。现有框架如 LangChain 和 LlamaIndex 为这些集成提供了机制，但它们通常需要为每个系统使用一次性连接器，从而产生脆弱且难以扩展的解决方案。

MCP 通过提供一种通用标准来简化这些连接，从而解决了这个问题。开发者无需为每个工具或数据库重新发明集成方式，而是可以使用 MCP 作为 AI 模型与外部数据源之间的标准化桥梁。

MCP 的核心在于标准化智能体检索和与外部数据交互的方式。它为 AI 系统提供了一种结构化的方式来促进这些交互：

• 从数据库拉取客户记录

• 从云存储检索文档

• 基于实时输入执行工作流

通过建立通用协议，MCP 消除了构建 AI 驱动工作流中的大量复杂性和冗余。这使开发者能够专注于智能体逻辑，而非底层的集成挑战。

MCP 如何工作？

MCP 采用客户端-服务器架构，AI 驱动的应用程序（如 Claude Desktop 等客户端）与 MCP 服务器交互，以访问外部数据、工具和结构化提示。这种结构化方法确保智能体能够检索实时信息、执行预定义操作，并与外部系统保持安全、一致的交互。

从高层来看，MCP 通过在其服务器中定义三个关键组件来工作：

• 工具（Tools）– AI 智能体可调用的函数，用于执行特定操作，例如发起 API 请求或执行命令（如查询天气 API）。

• 资源（Resources）– AI 智能体可访问的数据源，类似于 REST API 端点。这些提供结构化数据，不执行额外计算。

• 提示（Prompts）– 预定义模板，指导 AI 模型以最优方式使用工具和资源。

MCP 服务器充当数据网关，通过标准化接口将这些工具、资源和提示暴露给 AI 应用程序。客户端（通常是助手或自动化工具等 AI 驱动系统）使用 JSON-RPC 2.0 与这些服务器通信，这是一种轻量级消息协议，确保安全、双向的通信。

在实践中，AI 驱动的客户端将：

• 连接到 MCP 服务器 – 客户端发起连接以发现可用的工具、资源和提示。

• 查询或调用工具/资源 – 基于用户输入，客户端通过 MCP 服务器请求数据或执行预定义函数。

• 处理并返回响应 – AI 智能体处理检索到的数据，应用上下文推理，并交付适当的响应或操作。

这种结构化方法降低了集成复杂性，改善了安全性和治理，并确保智能体以准确、最新的信息运行。

通过 Confluent 将实时数据引入 MCP

在 Confluent，我们多年来一直帮助企业将实时数据集成到他们的系统中，而 MCP 提供了一个机会，以标准化的方式将这一能力扩展到 AI 智能体。

AI 系统需要同时访问实时数据和历史数据才能有效运作。Confluent 已经提供了 120 多个预构建连接器，可以轻松地从数据库、事件系统和软件即服务（SaaS）应用程序中流式传输数据。通过添加 MCP 支持，我们让智能体可以直接访问这些数据源，而无需一次性集成。

这降低了管理独立数据管道的复杂性，并确保 AI 驱动的工作流基于最新的可用信息运行。

Confluent MCP Server 实现

我们构建了一个直接连接到 Confluent 的 MCP 服务器，允许智能体使用自然语言与实时数据交互。

用户无需手动配置主题、编写 Flink SQL 或管理连接器，只需用自然语言发出命令，服务器就会将其转换为可执行操作。

它能做什么：

• 管理 Kafka 主题（创建、删除、更新）

• 生产和消费消息

• 执行 Flink SQL 查询

• 管理和配置连接器

• 标记和组织主题

该服务器让智能体更容易连接到流经其中的数据，并有助于自动化配置和管理。

在此查看：https://github.com/confluentinc/mcp-confluent

Confluent MCP 服务器中的每个工具都通过名称、描述和输入参数进行定义，使智能体能够以结构化方式与 Confluent 资源交互。

当前实现包含 20 个内置工具，添加新功能也很简单：只需定义一个新工具及其 schema 和执行逻辑。这使得在新用例出现时扩展系统变得非常容易。

与 MCP 和 Confluent 的交互示例

借助 MCP，智能体可以使用自然语言与 Confluent 交互，无需手动配置。

为了展示这一点，我们将 Confluent MCP 服务器与 Goose 集成，Goose 是 Block 推出的一个开源 AI 框架。Goose 被设计为一个在机 AI 智能体，帮助开发者自动化复杂任务。与 Claude Desktop 客户端一样，它原生支持 MCP，因此可以轻松连接外部系统。

以下演示了 Goose 客户端如何通过我们的 MCP 服务器与 Confluent 交互。

1. 列出所有 Apache Kafka ® 主题

用户：列出所有 Kafka 主题。

2. 将对话保存到 Apache Kafka ®

用户：你能保存我们的对话历史吗？

对话历史已保存到“claude-conversations”主题。

3. 从可用主题中检索示例数据

用户：帮我从“cool”和“hello”主题中采样一些数据。

4. 标记包含 PII 的主题

用户：让我们创建一个名为 PII（个人可识别信息）的标签。分析我们刚刚创建的主题，如果它们包含 PII，则相应地打上标签。

## 5. 更改 PII 主题的保留时间

用户：让我们将标记为 PII 的主题的保留时间更改为“1 天”。

这些示例展示了 MCP 如何通过允许自然语言驱动的数据操作来简化与 Confluent 的交互。用户无需编写 CLI 命令或 API 调用，即可使用英语管理 Kafka 主题、检查数据并执行策略。

## 添加新工具

Confluent MCP 服务器中的每个工具都通过名称、描述和输入参数进行定义，使智能体能够以结构化方式与 Confluent 资源交互。如果您希望添加不受支持的新功能，可以通过以下三个简单步骤完成：

• 在 ToolName 枚举中添加一个新值。 ‎ export enum ToolName { CREATE_TOPICS = "create-topics" , } ‎

• 将新工具添加到 ToolFactory 类中的 handlers 映射中，将其与将要创建的类处理器关联起来。在本例中，我们将开发一个名为 CreateTopicsHandler 的类。 ‎ export class ToolFactory { private static handlers : Map < ToolName , ToolHandler > = new Map ( [ [ ToolName . CREATE_TOPICS , new CreateTopicsHandler ( ) ] , ] ) ; ‎

• 开发一个继承 BaseToolHandler 抽象类的类。这需要实现两个方法：getToolConfig：定义工具及其在 MCP 客户端中的呈现方式。handle：当工具被调用时调用此方法。 ‎ ... export class CreateTopicsHandler extends BaseToolHandler { async handle ( clientManager : ClientManager , toolArguments : Record < string , unknown > , ) : Promise < CallToolResult > { const { topicNames } = createTopicArgs . parse ( toolArguments ) ; const success = await ( await clientManager . getAdminClient ( ) ) . createTopics ( { topics : topicNames . map ( ( name ) => ( { topic : name } ) ) , } ) ; if ( ! success ) { return this . createResponse ( ` Failed to create Kafka topics: ${ topicNames . join ( "," ) } ` , true , ) ; } return this . createResponse ( ` Created Kafka topics: ${ topicNames . join ( "," ) } ` ) ; } getToolConfig ( ) : ToolConfig { return { name : ToolName . CREATE_TOPICS , description : "Create new topic(s) in the Kafka cluster." , inputSchema : createTopicArgs . shape , } ; } } ‎

通过遵循这些步骤，您可以轻松地向 MCP 服务器添加新工具，使 AI 智能体能够更有效地与 Confluent 资源交互。这种灵活性确保服务器能够适应不断变化的需求并支持日益增长的用例范围，使其成为希望将 AI 与实时数据集成在一起的组织的宝贵资产。

## 以实时数据赋能智能体

AI 智能体的能力取决于它们所能访问的数据。

如果它们基于陈旧、过时的信息做出决策，其洞察力很快就会失去相关性。为了有效运作，智能体需要实时访问跨异构系统的数据，确保它们始终使用最新的上下文。

这正是 MCP 和我们的 Confluent MCP 服务器发挥作用的地方。

通过为智能体提供一种标准化方式，使其能够连接实时数据流、结构化数据集和外部工具，我们消除了对定制、一次性集成的需求。智能体可以检索实时数据、执行操作，并基于最新可用信息做出决策，而开发者无需为每个数据源编写定制代码。

连接实时数据与存储数据

借助 Tableflow，我们可以通过统一实时数据和存储数据，进一步推进这一能力。

• 跨向量存储、数据库和数据湖执行联邦搜索

• 从单一界面访问历史数据集和实时流数据

• 将 Apache Iceberg™️ 和 Delta Tables 与事件流一起查询，确保对结构化和半结构化数据的无缝访问

这意味着 AI 智能体不再局限于过去的快照。相反，它们可以对实时事件做出反应，将其与存储的知识关联起来，并交付及时、有依据的洞察。

无需自定义集成

由于 Confluent 已经提供 120+ 预构建连接器，AI 智能体可以与企业系统、SaaS 平台、数据库和事件流互连，而无需额外开发工作。MCP 服务器抽象了复杂性，使所有这些数据都可以通过通用接口立即访问。

通过这种方法，开发者无需构建和维护脆弱的自定义管道。相反，他们可以专注于设计智能体。

总结

如果你正在构建 AI 驱动的应用程序，并希望了解 MCP 如何简化你的工作流，请试用我们的 MCP 服务器，并告诉我们你的想法。

在 GitHub 上查看它，进行实验，并分享你的反馈。我们很乐意听到你的声音！

Apache ®、Apache Kafka ® 和 Apache Flink ® 是 Apache Software Foundation 在美国和/或其他国家/地区的商标。使用这些商标并不意味着 Apache Software Foundation 的认可。所有其他商标均为其各自所有者的财产。

• Edward 是 Confluent 的一名 Staff Partner Innovation Engineer，专精于分布式系统和事件驱动架构。他设计并实现企业级 Kafka 生态系统、云原生解决方案以及跨云提供商的实时数据管道。他以架构复杂系统而闻名——从协议级工程到 AI 应用程序——同时倡导技术卓越并自动化工作流。

  
  

转载声明：本文为转载整理，原文版权归原作者及原站点所有。

原文来源：Confluent Blog

原创地址：https://www.confluent.io/blog/ai-agents-using-anthropic-mcp/