StreamSets MCP Server

A comprehensive Model Context Protocol (MCP) server that provides seamless integration with StreamSets Control Hub APIs, enabling complete data pipeline management and creation through conversational AI.

🚀 Features

Pipeline Management (Read Operations)

Job Management: List, start, stop, and monitor job execution
Pipeline Operations: Browse, search, and analyze pipeline configurations
Connection Management: Manage data connections and integrations
Metrics & Analytics: Comprehensive performance and usage analytics
Enterprise Integration: Deployment management, security audits, and alerts

Pipeline Building (Write Operations) 🆕

Interactive Pipeline Creation: Build pipelines through conversation
Stage Library: Access to 25+ StreamSets stages (Origins, Processors, Destinations, Executors)
Visual Flow Management: Connect stages with data and event streams
Persistent Sessions: Pipeline builders persist across conversations
Smart Validation: Automatic validation of pipeline logic and connections

📊 API Coverage

44 Tools covering 9 StreamSets Services:

Job Runner API (11 tools) - Job lifecycle management
Pipeline Repository API (7 tools) - Pipeline CRUD operations
Connection API (4 tools) - Data connection management
Provisioning API (5 tools) - Infrastructure and deployment
Notification API (2 tools) - Alert and notification management
Topology API (1 tool) - System topology information
Metrics APIs (7 tools) - Performance and usage analytics
Security API (1 tool) - Security audit trails
Pipeline Builder (6 tools) - Interactive pipeline creation

🏗️ Pipeline Builder Capabilities

Create Complete Data Pipelines

# 1. Initialize a new pipeline builder sdc_create_pipeline_builder title="My ETL Pipeline" engine_type="data_collector" # 2. Browse available stages sdc_list_available_stages category="origins" # 3. Add stages to your pipeline sdc_add_pipeline_stage pipeline_id="pipeline_builder_1" stage_label="Dev Raw Data Source" sdc_add_pipeline_stage pipeline_id="pipeline_builder_1" stage_label="Expression Evaluator" sdc_add_pipeline_stage pipeline_id="pipeline_builder_1" stage_label="Trash" # 4. Connect stages with data flows sdc_connect_pipeline_stages pipeline_id="pipeline_builder_1" source_stage_id="stage_1" target_stage_id="stage_2" sdc_connect_pipeline_stages pipeline_id="pipeline_builder_1" source_stage_id="stage_2" target_stage_id="stage_3" # 5. Visualize your pipeline flow sdc_get_pipeline_flow pipeline_id="pipeline_builder_1" # 6. Build and publish (coming soon) # sdc_build_pipeline pipeline_id="pipeline_builder_1" # sdc_publish_pipeline pipeline_id="pipeline_builder_1"

Persistent Pipeline Sessions

Cross-Conversation: Continue building pipelines across multiple conversations
Auto-Save: All changes automatically saved to disk
Session Management: List, view, and delete pipeline builder sessions
Storage Location: ~/.streamsets_mcp/pipeline_builders/

🛠️ Installation

Prerequisites

Python 3.8+
StreamSets Control Hub account with API credentials
Claude Desktop (for MCP integration)

Setup

Clone the repository
git clone https://github.com/yourusername/streamsets-mcp-server.git cd streamsets-mcp-server
Install dependencies
pip install -r requirements.txt
Configure environment variables
export STREAMSETS_HOST_PREFIX="https://your-instance.streamsets.com" export STREAMSETS_CRED_ID="your-credential-id" export STREAMSETS_CRED_TOKEN="your-auth-token"
Test the server
python streamsets_server.py

Docker Deployment

Setup for MCP Integration

# Build the image docker build -t streamsets-mcp-server . # Create persistent volume for pipeline builders docker volume create streamsets-pipeline-data

Manual Testing

# Test run with volume persistence docker run --rm -it \ -e STREAMSETS_HOST_PREFIX="https://your-instance.streamsets.com" \ -e STREAMSETS_CRED_ID="your-credential-id" \ -e STREAMSETS_CRED_TOKEN="your-auth-token" \ -v streamsets-pipeline-data:/data \ streamsets-mcp-server

Claude Desktop Integration

Option 1: Direct Python (Local Development)

{ "mcpServers": { "streamsets": { "command": "python", "args": ["/path/to/streamsets_server.py"], "env": { "STREAMSETS_HOST_PREFIX": "https://your-instance.streamsets.com", "STREAMSETS_CRED_ID": "your-credential-id", "STREAMSETS_CRED_TOKEN": "your-auth-token" } } } }

Option 2: Docker with Persistence (Production)

{ "mcpServers": { "streamsets": { "command": "docker", "args": [ "run", "--rm", "-i", "-v", "streamsets-pipeline-data:/data", "-e", "STREAMSETS_HOST_PREFIX=https://your-instance.streamsets.com", "-e", "STREAMSETS_CRED_ID=your-credential-id", "-e", "STREAMSETS_CRED_TOKEN=your-auth-token", "streamsets-mcp-server" ] } } }

📖 Usage Examples

Job Management

# List all jobs sdc_list_jobs organization="your-org" status="ACTIVE" # Get detailed job information sdc_get_job_details job_id="your-job-id" # Start/stop jobs sdc_start_job job_id="your-job-id" sdc_stop_job job_id="your-job-id" # Bulk operations sdc_start_multiple_jobs job_ids="job1,job2,job3"

Pipeline Operations

# Search pipelines sdc_search_pipelines search_query="name==ETL*" # Get pipeline details sdc_get_pipeline_details pipeline_id="your-pipeline-id" # Export/import pipelines sdc_export_pipelines commit_ids="commit1,commit2"

Metrics & Analytics

# Job performance metrics sdc_get_job_metrics job_id="your-job-id" # System health overview sdc_get_job_count_by_status # Executor infrastructure metrics sdc_get_executor_metrics executor_type="COLLECTOR" label="prod" # Security audit trails sdc_get_security_audit_metrics org_id="your-org" audit_type="login"

🔧 Configuration

Environment Variables

Required (StreamSets Authentication)

STREAMSETS_HOST_PREFIX - StreamSets Control Hub URL
STREAMSETS_CRED_ID - API Credential ID
STREAMSETS_CRED_TOKEN - Authentication Token

Optional (Pipeline Builder Persistence)

PIPELINE_STORAGE_PATH - Custom storage directory for pipeline builders

Pipeline Builder Storage

Pipeline builders are automatically persisted across conversations and container restarts:

Storage Locations (Priority Order)

Custom Path: PIPELINE_STORAGE_PATH environment variable
Docker Volume: /data/pipeline_builders (when running in Docker)
Default Path: ~/.streamsets_mcp/pipeline_builders/

Configuration Options

Format: Pickle files for session persistence
Management: Automatic file management with error handling
Fallback: Memory-only mode if no writable storage available

Docker Persistence

When using Docker, pipeline builders persist in named volumes:

# Data persists in Docker volume 'streamsets-pipeline-data' docker volume create streamsets-pipeline-data # Run with persistent volume docker run --rm -it -v streamsets-pipeline-data:/data streamsets-mcp-server

Troubleshooting

No Persistence: Check storage directory permissions
Docker Issues: Ensure volume mounts are configured correctly
Memory Mode: Server logs will indicate if persistence is disabled

📚 Documentation

API Reference: See CLAUDE.md for detailed tool documentation
Stage Library: Built-in documentation for 25+ StreamSets stages
Configuration: custom.yaml for MCP server registry
Swagger Specs: API specifications in /swagger/ directory

🧪 Development

Project Structure

streamsets-mcp-server/ ├── streamsets_server.py # Main MCP server implementation ├── custom.yaml # MCP server configuration ├── CLAUDE.md # Comprehensive documentation ├── requirements.txt # Python dependencies ├── Dockerfile # Container deployment ├── swagger/ # API specifications └── README.md # This file

Adding New Tools

Define tool function with @mcp.tool() decorator
Add comprehensive error handling and logging
Update custom.yaml with tool metadata
Document in CLAUDE.md

Testing

# Syntax validation python -m py_compile streamsets_server.py # Tool count verification grep -c "@mcp.tool()" streamsets_server.py

🤝 Contributing

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

📝 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

StreamSets for the comprehensive Control Hub APIs
Anthropic for the Model Context Protocol framework
FastMCP for the Python MCP server implementation

📧 Support

For issues and questions:

Create an issue on GitHub
Check the documentation in CLAUDE.md
Review the API specifications in /swagger/

Transform your data pipeline workflows with conversational AI! 🚀

This server cannot be installed

security - not tested

license - not found

quality - not tested

How are these scores calculated?

remote-capable server

The server can be hosted and run remotely because it primarily relies on remote services or has no dependency on the local environment.

Enables complete StreamSets Control Hub integration through conversational AI, allowing users to manage data pipelines, monitor jobs, and interactively build new pipelines with 44 tools across 9 StreamSets services. Features persistent pipeline builder sessions that let users create complete ETL workflows through natural language conversations.