AI导哈

任务管理系统

一个基于MCP和Kafka的生产级任务管理系统，支持任务管理、实时通知处理和事件流处理。

  • by never2average
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 never2average/a2a-mcp-server

使用说明

MCP服务器演示

使用MCP（模型控制协议）和Kafka构建的生产就绪任务管理系统。

概述

该项目展示了一个强大的任务管理系统，该系统使用MCP使AI代理能够与基于Kafka的任务队列进行交互。该系统允许：

  • 任务管理（创建、更新、完成任务）
  • 通知处理
  • 通过Kafka进行实时事件处理

特性

  • 任务管理：创建、更新、优先排序和完成生产任务
  • 通知系统：具有优先级的实时通知
  • Kafka集成：可靠的消息队列和事件流
  • MCP工具：用于任务和通知操作的AI友好界面
  • 消费者服务：Kafka消息的后台处理

需求

  • Python 3.13+
  • Kafka集群（本地或AWS MSK）
  • 融合的Kafka Python客户端

安装

# Clone the repository
git clone https://github.com/yourusername/mcp-server-demo.git
cd mcp-server-demo

# Install dependencies
pip install -e .

配置

更新Kafka配置 kafka_config.py 使用您实际的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',
}

用法

启动服务器

python main.py

加载测试数据

要用示例任务和通知填充系统：

python kafka_test_data.py

MCP工具

该系统为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主题的消息
  • 任务服务：任务管理的业务逻辑
  • 通知服务：通知处理的业务逻辑

发展

项目结构

mcp-server-demo/
├── main.py                # MCP server initialization
├── kafka_config.py        # Kafka configuration
├── consumer_service.py    # Kafka consumer services
├── task_service.py        # Task management logic
├── notification_service.py # Notification handling logic
└── kafka_test_data.py     # Test data generator

许可证

MIT许可证

贡献

欢迎投稿！请随时提交拉取请求。

相关 MCP