Apache Beam MCP Server

Apache Beam MCP Server

Flink、Spark、Dataflow、Direct など、さまざまなランナーで Apache Beam パイプラインを管理するための Model Context Protocol（MCP）サーバーです。

これは何ですか？

Apache Beam MCP Server は、さまざまなランナーで Apache Beam データパイプラインを管理するための標準化された API を提供します。次のユーザー向けに設計されています。

• データエンジニア：ランナーにかかわらず、一貫した API でパイプラインを管理できます
• AI／LLM 開発者：MCP 標準を介して、AI によるデータパイプラインの制御を可能にします
• DevOps チーム：パイプラインの運用と監視を簡素化します

主な機能

• 複数ランナーのサポート：Flink、Spark、Dataflow、Direct ランナー向けの単一 API
• MCP 準拠：AI 連携のための Model Context Protocol に準拠
• パイプライン管理：データパイプラインの作成、監視、制御
• 拡張が容易：新しいランナーやカスタム機能を追加可能
• 本番環境対応：Docker／Kubernetes へのデプロイ、監視、スケーリングに対応

クイックスタート

インストール

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

# Create a virtual environment
python -m venv beam-mcp-venv
source beam-mcp-venv/bin/activate  # On Windows: beam-mcp-venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

サーバーの起動

# With the Direct runner (no external dependencies)
python main.py --debug --port 8888

# With Flink runner (if you have Flink installed)
CONFIG_PATH=config/flink_config.yaml python main.py --debug --port 8888

最初のジョブを実行

# Create test input
echo "This is a test file for Apache Beam WordCount example" > /tmp/input.txt

# Submit a job using curl
curl -X POST http://localhost:8888/api/v1/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "job_name": "test-wordcount",
    "runner_type": "direct",
    "job_type": "BATCH",
    "code_path": "examples/pipelines/wordcount.py",
    "pipeline_options": {
      "input_file": "/tmp/input.txt",
      "output_path": "/tmp/output"
    }
  }'

Docker のサポート

ビルド済みイメージの使用

ビルド済みの Docker イメージは GitHub Container Registry で入手できます。

# Pull the latest image
docker pull ghcr.io/yourusername/beam-mcp-server:latest

# Run the container
docker run -p 8888:8888 \
  -v $(pwd)/config:/app/config \
  -e GCP_PROJECT_ID=your-gcp-project \
  -e GCP_REGION=us-central1 \
  ghcr.io/yourusername/beam-mcp-server:latest

独自のイメージのビルド

# Build the image
./scripts/build_and_push_images.sh

# Build and push to a registry
./scripts/build_and_push_images.sh --registry your-registry --push --latest

Docker Compose

複数のサービス（Flink、Spark、Prometheus、Grafana）を使用したローカル開発向けです。

docker-compose -f docker-compose.dev.yaml up -d

Kubernetes へのデプロイ

このリポジトリには、Beam MCP Server を Kubernetes にデプロイするための Kubernetes マニフェストが含まれています。

# Deploy using kubectl
kubectl apply -k kubernetes/

# Deploy using Helm
helm install beam-mcp ./helm/beam-mcp-server \
  --namespace beam-mcp \
  --create-namespace

詳しいデプロイ手順については、Kubernetes デプロイガイドを参照してください。

MCP 標準エンドポイント

Beam MCP Server は、すべての標準 Model Context Protocol（MCP）エンドポイントを実装し、AI が管理するデータパイプラインのための包括的なフレームワークを提供します。

/tools エンドポイント

パイプライン処理用の AI エージェントとモデルを管理します。

# Register a sentiment analysis tool
curl -X POST "http://localhost:8888/api/v1/tools/" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "sentiment-analyzer",
    "description": "Analyzes sentiment in text data",
    "type": "transformation",
    "parameters": {
      "text_column": {
        "type": "string",
        "description": "Column containing text to analyze"
      }
    }
  }'

/resources エンドポイント

データセットやその他のパイプラインリソースを管理します。

# Register a dataset
curl -X POST "http://localhost:8888/api/v1/resources/" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Customer Transactions",
    "description": "Daily customer transaction data",
    "resource_type": "dataset",
    "location": "gs://analytics-data/transactions/*.csv"
  }'

/contexts エンドポイント

パイプラインの実行環境を定義します。

# Create a Dataflow execution context
curl -X POST "http://localhost:8888/api/v1/contexts/" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Dataflow Prod",
    "description": "Production Dataflow environment",
    "context_type": "dataflow",
    "parameters": {
      "region": "us-central1",
      "project": "beam-analytics-prod"
    }
  }'

これらの MCP 標準エンドポイントは Beam のコア機能とシームレスに連携し、データパイプライン管理のための完全なソリューションを提供します。詳しい例やユースケースについては、MCP プロトコル準拠を参照してください。

ドキュメント

• 開発者向けクイックスタート - 開発環境のセットアップ方法
• システム設計 - アーキテクチャと実装の詳細
• MCP プロトコル準拠 - MCP プロトコルの実装詳細
• ユーザーガイドと LLM 連携 - サーバーの使用方法と LLM 連携に関する包括的なガイド
• Kubernetes へのデプロイ - Kubernetes へのデプロイガイド
• クラウド最適化 - クラウド環境の最適化ガイド
• ローカル環境の要件 - ローカルテストのセットアップ要件
• トラブルシューティングガイド - よくある問題と解決策
• 貢献ガイド - 貢献方法
• テストの README - テストに関する情報

Python クライアントの例

import requests

# Get available runners
headers = {"MCP-Session-ID": "my-session-123"}
runners = requests.get("http://localhost:8888/api/v1/runners", headers=headers).json()

# Create a job
job = requests.post(
    "http://localhost:8888/api/v1/jobs",
    headers=headers,
    json={
        "job_name": "wordcount-example",
        "runner_type": "flink",
        "job_type": "BATCH",
        "code_path": "examples/pipelines/wordcount.py",
        "pipeline_options": {
            "parallelism": 2,
            "input_file": "/tmp/input.txt",
            "output_path": "/tmp/output"
        }
    }
).json()

# Monitor job status
job_id = job["data"]["job_id"]
status = requests.get(f"http://localhost:8888/api/v1/jobs/{job_id}", headers=headers).json()

CI/CD パイプライン

このリポジトリには、継続的インテグレーションとデプロイのための GitHub Actions ワークフローが含まれています。

• CI：すべてのプルリクエストでテスト、リンティング、型チェックを実行します
• CD：main/master へのプッシュごとに Docker イメージをビルドしてプッシュします
• デプロイ：開発環境と本番環境に自動的にデプロイします

モニタリングと可観測性

Beam MCP Server には、モニタリングと可観測性のサポートが組み込まれています。

• Prometheus メトリクス：/metrics エンドポイントでメトリクスを公開します
• Grafana ダッシュボード：モニタリング用の事前設定済みダッシュボードです
• ヘルスチェック：/health にヘルスチェックエンドポイントを提供します
• ロギング：ログ集約システムと簡単に統合できる、構造化された JSON ロギングです

コントリビューション

コントリビューションを歓迎します！詳しくは、コントリビューションガイドをご覧ください。

テストを実行するには：

# Run the regression tests
./scripts/run_regression_tests.sh

ライセンス

このプロジェクトは Apache License 2.0 の下でライセンスされています。

MCP 実装ステータス

MCP（Model Context Protocol）の実装は、複数のフェーズに分かれています。

フェーズ 1：コア接続ライフサイクル（完了）

• ✅ 接続の初期化
• ✅ 接続状態の管理
• ✅ 基本的な機能ネゴシエーション
• ✅ SSE による HTTP トランスポート
• ✅ JSON-RPC メッセージの処理
• ✅ エラー処理

フェーズ 2：完全な機能ネゴシエーション（完了）

• ✅ 機能の互換性チェックの強化
• ✅ 機能のセマンティックバージョン互換性
• ✅ 機能のサポートレベル（必須、優先、任意、実験的）
• ✅ 機能プロパティの検証
• ✅ 機能ベースの API エンドポイント制御
• ✅ FastAPI との機能ルーター統合

フェーズ 3：高度なメッセージ処理（完了）

• ✅ 構造化メッセージ型
• ✅ メッセージの検証
• ✅ エラー処理の改善
• ✅ バッチメッセージ処理

フェーズ 4：本番環境向けの最適化（未着手）

• ⬜ パフォーマンスの最適化
• ⬜ モニタリングとメトリクス
• ⬜ 高度なセキュリティ機能
• ⬜ 高可用性のサポート

MCP サーバーとやり取りするクライアントを構築する際は、Model Context Protocol に従う必要があります。詳しくは、MCP プロトコル準拠をご覧ください。