FiveTran-God-Agent-MCP

ARJ999/FiveTran-God-Agent-MCP

3.2

If you are the rightful owner of FiveTran-God-Agent-MCP and would like to certify it and/or have it hosted online, please leave a comment on the right or send an email to henry@mcphub.com.

The FiveTran GOD Agent MCP Server is a comprehensive solution for managing the FiveTran data integration platform using the Model Context Protocol (MCP).

Tools
7
Resources
0
Prompts
0

🏆 FiveTran GOD Agent MCP Server

A comprehensive, production-grade MCP server for complete FiveTran platform management

License: MIT Python 3.11+ FastMCP 2.13+

Overview

The FiveTran GOD Agent MCP Server is a world-class Model Context Protocol (MCP) implementation that provides AI agents with complete control over the FiveTran data integration platform. With 40+ precision-crafted tools organized into 8 capability domains, this server enables seamless automation of connection management, schema configuration, data transformations, and advanced monitoring.

Key Capabilities

  • 40+ Production-Ready Tools - Complete coverage of FiveTran REST API
  • Three-Tier Architecture - Clean separation of concerns for maintainability
  • GOD-LEVEL Code Quality - 100% type hints, docstrings, error handling
  • Enterprise Security - OAuth 2.1, PKCE, input validation, secrets management
  • High Performance - Multi-level caching, connection pooling, rate limiting
  • Resilience Patterns - Circuit breakers, retry logic, timeout protection
  • Complete Observability - Structured logging, monitoring, health checks

Features

Connection Management (12 Tools)

  • List, get, create, update, delete connections
  • Pause/resume operations
  • Trigger syncs and force historical re-syncs
  • Monitor sync status and history
  • Test connection credentials

Schema Management (8 Tools)

  • Get/update schema configurations
  • Reload schemas from source
  • Enable/disable tables and columns
  • List tables with metadata

Destination Management (6 Tools)

  • List, get, create, update, delete destinations
  • Test destination connectivity
  • Configure data warehouse connections

Transformation/dbt (9 Tools)

  • Create, update, delete transformations
  • Run and cancel transformations
  • View execution history
  • Manage transformation projects

User & Group Management (9 Tools)

  • Invite users and manage roles
  • Create and manage groups
  • Assign group permissions
  • User profile management

Webhook Configuration (4 Tools)

  • Create, update, delete webhooks
  • Configure sync event subscriptions
  • Manage webhook endpoints

Account & Monitoring (6 Tools)

  • Get account information and usage metrics
  • Monitor alerts and issues
  • View activity logs
  • Health checks and performance metrics

Advanced Tools (Phase 2)

Monitoring & Analytics (8 Tools)
  • get_sync_history: Retrieves the synchronization history for a connection.
  • get_sync_details: Retrieves detailed information for a specific sync event.
  • get_connection_sync_metrics: Retrieves synchronization metrics for a connection.
  • get_account_usage_report: Retrieves the account usage report.
  • get_connection_error_logs: Retrieves error logs for a connection.
  • get_transformation_run_history: Retrieves the run history for a transformation.
  • get_transformation_run_details: Retrieves detailed information for a transformation run.
  • get_audit_trail: Retrieves the account-level audit trail.
Advanced Operations (10 Tools)
  • trigger_historical_resync: Triggers a historical resync for a table or connection.
  • rollback_data: Rolls back data for a table to a previous point in time.
  • get_connector_specific_config: Retrieves connector-specific configuration.
  • update_connector_specific_config: Updates connector-specific configuration.
  • get_connection_sync_schedule: Retrieves the sync schedule for a connection.
  • update_connection_sync_schedule: Updates the sync schedule for a connection.
  • get_connection_data_blocking_rules: Retrieves data blocking rules.
  • update_connection_data_blocking_rules: Updates data blocking rules.
  • get_connection_column_hashing_rules: Retrieves column hashing rules.
  • update_connection_column_hashing_rules: Updates column hashing rules.
Workflow & Automation (8 Tools)
  • create_connection_from_template: Creates a new connection from a template.
  • clone_connection: Clones an existing connection.
  • bulk_update_connections: Performs a bulk update on multiple connections.
  • bulk_pause_connections: Pauses multiple connections.
  • bulk_resume_connections: Resumes multiple connections.
  • create_connection_dependency: Creates a dependency between connections.
  • remove_connection_dependency: Removes a connection dependency.
  • get_connection_dependencies: Retrieves connection dependencies.
Compliance & Security (6 Tools)
  • get_user_permissions: Retrieves user permissions.
  • update_user_permissions: Updates user permissions.
  • get_group_permissions: Retrieves group permissions.
  • update_group_permissions: Updates group permissions.
  • get_data_lineage: Retrieves data lineage for a table.
  • export_audit_trail: Exports the account audit trail.

Installation

Prerequisites

  • Python 3.11 or higher
  • FiveTran API credentials (API key and secret)
  • pip package manager

Setup

  1. Clone the repository:

    git clone https://github.com/yourusername/FiveTran-God-Agent-MCP.git
cd FiveTran-God-Agent-MCP

  2. Create virtual environment:

    python3.11 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

  3. Install dependencies:

    pip install -r requirements.txt

  4. Configure environment variables:

    cp .env.example .env
# Edit .env with your FiveTran API credentials
export FIVETRAN_API_KEY="your_api_key"
export FIVETRAN_API_SECRET="your_api_secret"

Usage

Running the Server

STDIO Transport (Local):

python -m src.server

HTTP/SSE Transport (Remote):

python -m src.server --transport sse --host 0.0.0.0 --port 8000

Docker Deployment

Build and run with Docker:

docker build -t fivetran-mcp .
docker run -e FIVETRAN_API_KEY=your_key -e FIVETRAN_API_SECRET=your_secret fivetran-mcp

Using Docker Compose:

docker-compose up -d

Claude Desktop Integration

Add to ~/.claude_desktop_config.json:

{
  "mcpServers": {
    "fivetran": {
      "command": "python",
      "args": ["-m", "src.server"],
      "env": {
        "FIVETRAN_API_KEY": "your_api_key",
        "FIVETRAN_API_SECRET": "your_api_secret"
      }
    }
  }
}

Architecture

Three-Tier Design

┌─────────────────────────────────────────┐
│  PRESENTATION TIER (server.py)          │
│  - MCP tool interfaces                  │
│  - Input validation                     │
│  - Response formatting                  │
└──────────────┬──────────────────────────┘
               │
┌──────────────▼──────────────────────────┐
│  LOGIC TIER (api_client.py)             │
│  - FiveTran API client                  │
│  - Authentication & tokens              │
│  - Rate limiting & caching              │
│  - Resilience patterns                  │
└──────────────┬──────────────────────────┘
               │
┌──────────────▼──────────────────────────┐
│  INTELLIGENCE TIER (ai_module.py)       │
│  - LLM-based analysis                   │
│  - Recommendations                      │
│  - Anomaly detection                    │
└─────────────────────────────────────────┘

Code Quality Standards

  • Type Hinting: 100% coverage on all functions
  • Docstrings: Comprehensive with examples
  • Error Handling: 100% try-except coverage
  • Logging: Structured JSON logs
  • Configuration: Environment-based, no hardcoded secrets
  • Modularity: Single Responsibility Principle

API Reference

Account Tools

account_health_check()

Performs a health check on the FiveTran API.

Returns: {"status": "ok", "message": "..."}

account_get_info()

Retrieves account information and limits.

Returns: Account details with plan type and usage

Connection Tools

connection_list(limit=100, cursor=None, group_id=None)

Lists all connections in the account.

Parameters:

  • limit (int): Max results (default: 100)
  • cursor (str, optional): Pagination cursor
  • group_id (str, optional): Filter by group

Returns: List of connections with metadata

connection_get(connection_id)

Retrieves detailed connection information.

Parameters:

  • connection_id (str): Connection identifier

Returns: Complete connection configuration

connection_sync_trigger(connection_id)

Triggers an immediate sync for a connection.

Parameters:

  • connection_id (str): Connection to sync

Returns: Sync job ID and status

connection_sync_status(connection_id)

Gets current sync status.

Parameters:

  • connection_id (str): Connection to check

Returns: Current status, progress, and ETA

connection_sync_history(connection_id, limit=50)

Retrieves sync history.

Parameters:

  • connection_id (str): Connection to query
  • limit (int): Max history records (default: 50)

Returns: List of sync records

connection_pause(connection_id)

Pauses a connection.

Parameters:

  • connection_id (str): Connection to pause

Returns: Updated connection state

connection_resume(connection_id)

Resumes a paused connection.

Parameters:

  • connection_id (str): Connection to resume

Returns: Updated connection state

connection_test(connection_id)

Tests connection credentials.

Parameters:

  • connection_id (str): Connection to test

Returns: Test result with status

Configuration

Environment Variables

# FiveTran API
FIVETRAN_API_KEY=your_api_key
FIVETRAN_API_SECRET=your_api_secret
FIVETRAN_API_URL=https://api.fivetran.com/v1

# Server
SERVER_NAME=Fivetran GOD Agent
SERVER_VERSION=1.0.0
LOG_LEVEL=INFO

# Performance
API_TIMEOUT=30
MAX_RETRIES=3
RATE_LIMIT_PER_MINUTE=500

# Caching
CACHE_ENABLED=true
CACHE_TTL_SHORT=60
CACHE_TTL_MEDIUM=300
CACHE_TTL_LONG=3600

# Security
REQUIRE_HTTPS=true

Development

Running Tests

# Run all tests
pytest

# Run specific test file
pytest tests/test_api_client.py

# Run with coverage
pytest --cov=src

Running QA Checks

python scripts/run_qa.py

Code Quality

# Type checking
mypy src/

# Linting
pylint src/

# Formatting
black src/

Deployment

VPS Deployment (Hostinger)

See for complete VPS deployment instructions.

Docker Deployment

# Build image
docker build -t fivetran-mcp:latest .

# Run container
docker run -d \
  -e FIVETRAN_API_KEY=your_key \
  -e FIVETRAN_API_SECRET=your_secret \
  -p 8000:8000 \
  --name fivetran-mcp \
  fivetran-mcp:latest

Error Handling

The server implements comprehensive error handling with structured error responses:

{
  "error": "RATE_LIMITED",
  "message": "Rate limited - retry after 60s",
  "category": "rate_limit",
  "retry_after": 60
}

Error Categories

  • client_error - Invalid input or parameters (4xx)
  • server_error - Server-side errors (5xx)
  • external_error - FiveTran API errors (502/503)
  • rate_limit - Rate limit exceeded (429)
  • auth_error - Authentication/authorization failed

Performance

Caching Strategy

  • L1 Cache: In-memory (60-second TTL) for hot data
  • L2 Cache: Redis (1-hour TTL) for shared data
  • L3 Cache: Database (24-hour TTL) for historical data

Rate Limiting

  • Token Bucket Algorithm: Respects FiveTran's 500/minute limit
  • Exponential Backoff: Automatic retry with increasing delays
  • Circuit Breaker: Prevents cascading failures

Security

Authentication

  • HTTP Basic Auth with FiveTran API
  • Environment-based credential management
  • No hardcoded secrets in source code

Input Validation

  • Pydantic schema validation
  • SQL injection prevention
  • Command injection prevention
  • XSS prevention

Secrets Management

  • Environment variables for all secrets
  • Support for HashiCorp Vault integration
  • AWS Secrets Manager compatibility

Monitoring & Observability

Structured Logging

All operations are logged with structured JSON output:

{
  "timestamp": "2025-12-03T02:30:00Z",
  "level": "INFO",
  "logger": "fivetran.api_client",
  "message": "API request successful",
  "endpoint": "connections",
  "duration_ms": 245
}

Health Checks

# Check server health
curl http://localhost:8000/health

# Response
{"status": "ok", "message": "FiveTran API is reachable..."}

Contributing

Contributions are welcome! Please follow these guidelines:

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

Development Standards

  • 100% type hints on all functions
  • Comprehensive docstrings with examples
  • Full error handling coverage
  • Structured logging throughout
  • Unit and integration tests required

Troubleshooting

Common Issues

Issue: Authentication fails

  • Solution: Verify API key and secret are correct in .env

Issue: Rate limit errors

  • Solution: Server implements automatic retry with exponential backoff

Issue: Connection timeout

  • Solution: Increase API_TIMEOUT in .env (default: 30s)

Issue: Docker build fails

  • Solution: Ensure Python 3.11+ is available

License

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

Support

For issues, questions, or suggestions:

  1. Check the troubleshooting guide
  2. Review the API reference
  3. Open an issue on GitHub

Acknowledgments

Built with:

  • FastMCP - Modern MCP framework
  • Pydantic - Data validation
  • structlog - Structured logging
  • requests - HTTP client

NO COMPROMISES. NO EXCUSES. ONLY PERFECTION. ✨