A2a MCP Server

🚀 MCP 服務器演示

本項目是一個基於 MCP（模型控制協議）和 Kafka 構建的生產就緒任務管理系統，它實現了 AI 代理與 Kafka 任務隊列的交互，能高效管理任務並處理通知。

🚀 快速開始

本系統支持任務管理、通知處理以及通過 Kafka 即時處理事件。以下是快速啟動系統的步驟：

1. 克隆倉庫並安裝依賴：

# 克隆倉庫
git clone https://github.com/yourusername/mcp-server-demo.git
cd mcp-server-demo

# 安裝依賴項
pip install -e .

2. 更新 kafka_config.py 中的 Kafka 配置為您的實際 Kafka 集群詳細信息：

KAFKA_CONFIG = {
    'bootstrap.servers': 'your-kafka-bootstrap-servers',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'SCRAM-SHA-512',
    'sasl.username': 'your-username',
    'sasl.password': 'your-password',
}

3. 啟動服務器：

python main.py

4. 加載測試數據：

python kafka_test_data.py

✨ 主要特性

• 任務管理：可創建、更新、調整優先級和完成生產任務。
• 通知系統：具備即時通知功能，且帶有優先級級別。
• Kafka 集成：實現可靠的消息隊列和事件流處理。
• MCP 工具：為 AI 代理提供友好型接口，用於任務和通知操作。
• 消費者服務：基於 Kafka 消息進行背景處理。

📦 安裝指南

環境要求

• Python 3.13+
• Kafka 集群（本地或 AWS MSK）
• Confluent Kafka Python 客戶端

安裝步驟

# 克隆倉庫
git clone https://github.com/yourusername/mcp-server-demo.git
cd mcp-server-demo

# 安裝依賴項
pip install -e .

💻 使用示例

基礎用法

啟動服務器

python main.py

加載測試數據

python kafka_test_data.py

高級用法

該系統為 AI 代理暴露了一系列 MCP 工具，可用於任務和通知管理。

任務管理

• fetch_queue：獲取待處理任務列表
• change_task_priority：更新任務優先級
• pickup_task：標記任務為進行中
• complete_task：標記任務為完成
• get_task_details：獲取任務詳細信息
• check_task_status：檢查任務當前狀態

通知管理

• check_notification_count：獲取未讀通知數量
• get_notification_list：獲取過濾後的通知列表
• mark_notification_as_read：標記通知為已讀

📚 詳細文檔

架構

該系統包含以下組件：

• MCP 服務器：為 AI 代理提供交互接口。
• Kafka 生產者：向 Kafka 主題發送消息。
• Kafka 消費者：從 Kafka 主題處理消息。
• 任務服務：負責任務管理的業務邏輯。
• 通知服務：負責通知處理的業務邏輯。

項目結構

mcp-server-demo/
├── main.py                # MCP 服務器初始化
├── kafka_config.py        # Kafka 配置
├── consumer_service.py    # Kafka 消費者服務
├── task_service.py        # 任務管理邏輯
├── notification_service.py # 通知處理邏輯
└── kafka_test_data.py     # 測試數據生成器

📄 許可證

本項目採用 MIT 許可證。

貢獻

歡迎貢獻！請隨意提交 Pull Request。