Apache Beam MCP Server

Apache Beam MCP 服务

一个用于在不同运行器（Flink、Spark、Dataflow 和 Direct）上管理 Apache Beam 流水线的模型上下文协议（MCP）服务。

这是什么？

Apache Beam MCP 服务提供了一个标准化 API，用于在不同运行器上管理 Apache Beam 数据流水线。它面向以下用户：

• 数据工程师：无论使用哪种运行器，都可以通过统一的 API 管理流水线
• AI/LLM 开发者：通过 MCP 标准实现由 AI 控制的数据流水线
• DevOps 团队：简化流水线操作与监控

主要功能

• 多运行器支持：通过一个 API 支持 Flink、Spark、Dataflow 和 Direct 运行器
• 符合 MCP 规范：遵循模型上下文协议，实现 AI 集成
• 流水线管理：创建、监控和控制数据流水线
• 易于扩展：添加新的运行器或自定义功能
• 可用于生产环境：包含 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 支持

使用预构建镜像

GitHub Container Registry 提供预构建的 Docker 镜像：

# 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 服务部署到 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 服务实现了所有标准模型上下文协议（MCP）端点，为 AI 管理的数据流水线提供全面的框架：

/tools 端点

管理用于流水线处理的 AI Agent 和模型：

# 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 协议合规性。