A2a MCP Server

🚀 MCP サーバーデモ

このプロジェクトは、MCP（モデル制御プロトコル）と Kafka をベースに構築された、本番環境で使用可能なタスク管理システムを展示しています。このシステムは、AI エージェントと Kafka ベースのタスクキューとの相互作用を実現し、タスク管理や通知処理、リアルタイムのイベント処理をサポートします。

🚀 クイックスタート

このシステムは、MCP を使用して AI エージェントと Kafka ベースのタスクキューを連携させる強力なタスク管理システムを提供します。以下の機能をサポートしています。

• タスク管理（タスクの作成、更新、完了）
• 通知処理
• Kafka を介したリアルタイムイベント処理

✨ 主な機能

• タスク管理：本番環境のタスクの作成、更新、優先度の調整、完了をサポート
• 通知システム：優先度付きのリアルタイム通知
• Kafka 統合：信頼性の高いメッセージキューとイベントフローの処理
• MCP ツール：AI エージェントに対するタスクと通知操作のためのインターフェース
• コンシューマーサービス：Kafka メッセージに基づくバックグラウンド処理

📦 インストール

# リポジトリをクローン
git clone https://github.com/yourusername/mcp-server-demo.git
cd mcp-server-demo

# 依存関係をインストール
pip install -e .

📚 ドキュメント

設定

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',
}

使用方法

サーバーの起動

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 コンシューマー：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 を送信してください。