Apache Beam MCP服务器
一个模型上下文协议（MCP）服务器，用于管理跨不同运行程序的Apache Beam管道：Flink、Spark、Dataflow和Direct。
 ](https://github.com/llm-mcp/mcp-spec)  ](https://github.com/yourusername/beam-mcp-server/pkgs/container/beam-mcp-server) 
这是什么？
Apache Beam MCP服务器提供了一个标准化的API，用于管理不同跑步者之间的Apache Beam数据管道。它的设计目的是：
- 数据工程师：使用一致的API管理管道，而不考虑流道
- AI/LLM开发人员：通过MCP标准启用AI控制的数据管道
- DevOps团队：简化管道操作和监控
主要特点
- 多跑步者支持：一个用于Flink、Spark、Dataflow和Direct runner的API
- 符合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支持
使用预构建图像
预构建的Docker镜像可以在GitHub容器注册表上找到：
# 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 --latestDocker Compose
对于具有多种服务的本地开发（Flink、Spark、Prometheus、Grafana）：
docker-compose -f docker-compose.dev.yaml up -dKubernetes部署
该存储库包括用于将Beam MCP服务器部署到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服务器实现了所有标准的模型上下文协议（MCP）端点，为人工智能管理的数据管道提供了一个全面的框架：
/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公司：对每个pull请求运行测试、linting和类型检查
- 光盘：每次向main/master推送Docker镜像时，都会构建并推送Docker映像
- 部署：自动部署到开发和生产环境
监测和可观察性
Beam MCP服务器内置了对监控和可观察性的支持：
- 普罗米修斯指标：在以下位置显示指标
/metrics端点 - Grafana仪表板：用于监控的预配置仪表板
- 健康检查：在以下位置提供健康检查端点
/health - 日志记录：结构化JSON日志记录，便于与日志聚合系统集成
贡献
我们欢迎捐款！查看我们的 贡献指南 了解详情。
要运行测试，请执行以下操作：
# Run the regression tests
./scripts/run_regression_tests.sh许可证
此项目根据Apache许可证2.0获得许可。
MCP实施状态
MCP（模型上下文协议）的实现分为几个阶段：
第1阶段：核心连接生命周期（已完成）
- ✅ 连接初始化
- ✅ 连接状态管理
- ✅ 基本能力谈判
- ✅ 使用SSE的HTTP传输
- ✅ JSON-RPC消息处理
- ✅ 错误处理
第二阶段：全面谈判（已完成）
- ✅ 增强的功能兼容性检查
- ✅ 功能的语义版本兼容性
- ✅ 功能支持级别（必需、首选、可选、实验）
- ✅ 能力属性验证
- ✅ 基于能力的API端点控制
- ✅ 功能路由器与FastAPI集成
第3阶段：高级消息处理（已完成）
- ✅ 结构化消息类型
- ✅ 消息验证
- ✅ 改进了错误处理
- ✅ 批处理消息处理
第四阶段：生产优化（TODO）
- ⬜ 性能优化
- ⬜ 监控和指标
- ⬜ 高级安全功能
- ⬜ 高可用性支持
在构建与MCP服务器交互的客户端时，您必须遵循模型上下文协议。有关详细信息，请参阅 MCP协议合规性.