from mcp.server.fastmcp import FastMCP

# Initialize the server with a name
mcp = FastMCP("DataEngineering")

@mcp.tool()
def query_database(query: str, database: str = "production") -> str:
    """Execute a SQL query against the specified database.

    Args:
        query: The SQL query to execute
        database: Target database name (default: production)

    Returns:
        Query results as formatted text
    """
    # In a real implementation, you would connect to your database here
    return f"Query executed on {database}: {query}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "data-engineering-server",
  version: "1.0.0",
});

server.registerTool(
  "query_database",
  {
    description: "Execute a SQL query against the specified database",
    inputSchema: {
      query: z.string().describe("The SQL query to execute"),
      database: z.string().default("production").describe("Target database name"),
    },
  },
  async ({ query, database }) => {
    // In a real implementation, connect to your database here
    return {
      content: [{ type: "text", text: `Query executed on ${database}: ${query}` }],
    };
  }
);

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main();

@mcp.tool()
def get_table_schema(table_name: str) -> str:
    """Get the schema definition for a database table.

    Returns column names, data types, and constraints for the specified table.
    Useful for understanding table structure before writing queries.

    Args:
        table_name: Fully qualified table name (e.g., 'analytics.orders')

    Returns:
        Schema definition as formatted text
    """
    # Implementation here
    return f"Schema for {table_name}: id INT PRIMARY KEY, name VARCHAR(255)..."

from pydantic import BaseModel, Field

class ValidationResult(BaseModel):
    """Result of a data quality validation check."""
    table_name: str
    row_count: int = Field(description="Total rows examined")
    null_count: int = Field(description="Number of null values found")
    duplicate_count: int = Field(description="Number of duplicate rows")
    is_valid: bool = Field(description="Whether the table passed validation")

@mcp.tool()
def run_data_quality_check(table_name: str) -> ValidationResult:
    """Run comprehensive data quality checks on a table.

    Validates completeness, uniqueness, and data integrity.

    Args:
        table_name: The table to validate

    Returns:
        Detailed validation results
    """
    # Run actual checks against your database
    return ValidationResult(
        table_name=table_name,
        row_count=10000,
        null_count=5,
        duplicate_count=0,
        is_valid=True
    )

from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field

class Environment(str, Enum):
    PRODUCTION = "production"
    STAGING = "staging"
    DEVELOPMENT = "development"

class QueryParams(BaseModel):
    """Parameters for database queries."""
    limit: int = Field(default=100, ge=1, le=10000, description="Maximum rows to return")
    timeout_seconds: int = Field(default=30, ge=1, le=300, description="Query timeout")

@mcp.tool()
def run_query(
    query: str,
    environment: Environment = Environment.PRODUCTION,
    params: Optional[QueryParams] = None
) -> str:
    """Execute a read-only SQL query.

    Args:
        query: SQL SELECT statement to execute
        environment: Target environment
        params: Optional query parameters

    Returns:
        Query results as JSON
    """
    effective_params = params or QueryParams()
    # Execute with validated parameters
    return f"Results from {environment.value} (limit: {effective_params.limit})"

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("DataEngineering")

@mcp.tool()
async def process_large_dataset(dataset_id: str, ctx: Context) -> str:
    """Process a large dataset with progress updates.

    Args:
        dataset_id: Identifier of the dataset to process
        ctx: MCP context for progress reporting (injected automatically)

    Returns:
        Processing summary
    """
    await ctx.info(f"Starting processing of {dataset_id}")

    total_batches = 10
    for i in range(total_batches):
        # Simulate batch processing
        await ctx.report_progress(
            progress=(i + 1) / total_batches,
            total=1.0,
            message=f"Processing batch {i + 1}/{total_batches}"
        )

    await ctx.info("Processing complete")
    return f"Processed dataset {dataset_id}: 10 batches, 100,000 rows"

@mcp.resource("schema://{table_name}")
def get_table_schema(table_name: str) -> str:
    """Expose table schema as a readable resource.

    URI pattern: schema://analytics.orders
    """
    # Fetch from your metadata store
    return f"""
    Table: {table_name}
    Columns:
      - id: INT (Primary Key)
      - customer_id: INT (Foreign Key)
      - total_amount: DECIMAL(10,2)
      - created_at: TIMESTAMP
    """

@mcp.resource("config://pipelines/{pipeline_id}")
def get_pipeline_config(pipeline_id: str) -> str:
    """Expose pipeline configuration."""
    return f"Pipeline {pipeline_id} configuration: schedule=daily, owner=data-team"

@mcp.prompt(title="Data Quality Report")
def data_quality_prompt(table_name: str) -> str:
    """Generate a data quality analysis prompt.

    Args:
        table_name: Table to analyze
    """
    return f"""Analyze the data quality of the '{table_name}' table.

Please check:
1. Completeness: What percentage of required fields are populated?
2. Uniqueness: Are there duplicate records?
3. Freshness: When was the data last updated?
4. Validity: Do values conform to expected formats and ranges?

Provide a summary with specific recommendations for improvement."""

@mcp.prompt(title="Schema Review")
def schema_review_prompt(table_name: str, changes: str) -> str:
    """Generate a schema change review prompt."""
    return f"""Review the proposed schema changes for {table_name}:

{changes}

Evaluate:
- Backward compatibility with existing queries
- Impact on downstream dependencies
- Performance implications
- Data migration requirements"""

if __name__ == "__main__":
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "my-data-server": {
      "command": "uv",
      "args": ["run", "/path/to/server.py"],
      "env": {
        "DATABASE_URL": "postgresql://localhost/analytics"
      }
    }
  }
}

mcp = FastMCP("RemoteDataServer", stateless_http=True)

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

{
  "mcpServers": {
    "remote-data-server": {
     "type": "http",
      "url": "https://mcp.yourcompany.com",
      "headers": {
        "Authorization": "Bearer ${MCP_API_KEY}"
      }
    }
  }
}

# BAD - breaks JSON-RPC communication
print("Debug: processing query")  # Goes to stdout!
print(f"Error: {e}")              # Also stdout!

# GOOD - use logging to stderr
import logging
import sys

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    stream=sys.stderr  # Explicitly send to stderr
)
logger = logging.getLogger(__name__)

logger.info("Debug: processing query")   # Goes to stderr
logger.error(f"Error: {e}")              # Also stderr

@mcp.tool()
async def my_tool(query: str, ctx: Context) -> str:
    await ctx.info("Processing query")  # Proper MCP logging
    # ...

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("DataCatalogMCP")

# Simulated catalog data - replace with your actual catalog API
CATALOG = {
    "sales.orders": {
        "description": "Order transactions from all channels",
        "columns": ["order_id", "customer_id", "total_amount", "created_at"],
        "owner": "sales-team",
        "tags": ["pii", "financial"],
        "upstream": ["raw.shopify_orders", "raw.pos_transactions"],
        "downstream": ["analytics.revenue_daily", "ml.churn_features"]
    },
    "sales.customers": {
        "description": "Customer master data with demographics",
        "columns": ["customer_id", "email", "segment", "lifetime_value"],
        "owner": "marketing-team",
        "tags": ["pii"],
        "upstream": ["raw.crm_contacts"],
        "downstream": ["sales.orders", "analytics.cohorts"]
    }
}

@mcp.tool()
def search_tables(
    query: str,
    tags: list[str] | None = None,
    limit: int = 10
) -> str:
    """Search for tables in the data catalog.

    Args:
        query: Search term to match against table names and descriptions
        tags: Optional list of tags to filter by (e.g., ['pii', 'financial'])
        limit: Maximum number of results to return

    Returns:
        Matching tables with descriptions
    """
    results = []
    for table_name, metadata in CATALOG.items():
        # Simple search matching
        if query.lower() in table_name.lower() or query.lower() in metadata["description"].lower():
            if tags is None or any(t in metadata["tags"] for t in tags):
                results.append({
                    "name": table_name,
                    "description": metadata["description"],
                    "owner": metadata["owner"],
                    "tags": metadata["tags"]
                })
    return json.dumps(results[:limit], indent=2)

@mcp.tool()
def get_table_details(table_name: str) -> str:
    """Get detailed metadata for a specific table.

    Args:
        table_name: Fully qualified table name (e.g., 'sales.orders')

    Returns:
        Complete table metadata including columns, owner, and tags
    """
    if table_name not in CATALOG:
        return json.dumps({"error": f"Table '{table_name}' not found in catalog"})

    metadata = CATALOG[table_name]
    return json.dumps({
        "table": table_name,
        **metadata
    }, indent=2)

@mcp.tool()
def get_data_lineage(
    table_name: str,
    direction: str = "both",
    depth: int = 2
) -> str:
    """Trace data lineage for a table.

    Args:
        table_name: Table to trace lineage for
        direction: 'upstream', 'downstream', or 'both'
        depth: How many levels to traverse (1-5)

    Returns:
        Lineage graph showing data flow
    """
    if table_name not in CATALOG:
        return json.dumps({"error": f"Table '{table_name}' not found"})

    metadata = CATALOG[table_name]
    lineage = {"table": table_name}

    if direction in ("upstream", "both"):
        lineage["upstream"] = metadata.get("upstream", [])
    if direction in ("downstream", "both"):
        lineage["downstream"] = metadata.get("downstream", [])

    return json.dumps(lineage, indent=2)

if __name__ == "__main__":
    mcp.run(transport="stdio")

from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum
import json

mcp = FastMCP("PipelineMonitorMCP")

class PipelineStatus(str, Enum):
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    PENDING = "pending"

class PipelineRun(BaseModel):
    """Details of a pipeline execution."""
    pipeline_id: str
    run_id: str
    status: PipelineStatus
    started_at: str
    finished_at: str | None = None
    duration_seconds: int | None = None
    error_message: str | None = None

# Simulated pipeline data - replace with Airflow/Dagster/Prefect API calls
PIPELINES = {
    "etl_daily_sales": {
        "last_run": PipelineRun(
            pipeline_id="etl_daily_sales",
            run_id="run_20260114_001",
            status=PipelineStatus.SUCCESS,
            started_at="2026-01-14T06:00:00Z",
            finished_at="2026-01-14T06:45:00Z",
            duration_seconds=2700
        ),
        "schedule": "0 6 * * *",
        "owner": "data-platform"
    },
    "sync_inventory": {
        "last_run": PipelineRun(
            pipeline_id="sync_inventory",
            run_id="run_20260114_001",
            status=PipelineStatus.FAILED,
            started_at="2026-01-14T07:00:00Z",
            finished_at="2026-01-14T07:15:00Z",
            duration_seconds=900,
            error_message="Connection timeout to inventory service"
        ),
        "schedule": "0 */2 * * *",
        "owner": "supply-chain"
    }
}

@mcp.tool()
def get_pipeline_status(pipeline_id: str) -> PipelineRun:
    """Get the current status of a data pipeline.

    Args:
        pipeline_id: Unique identifier for the pipeline

    Returns:
        Latest run details including status, timing, and any errors
    """
    if pipeline_id not in PIPELINES:
        return PipelineRun(
            pipeline_id=pipeline_id,
            run_id="unknown",
            status=PipelineStatus.PENDING,
            started_at=datetime.now().isoformat(),
            error_message=f"Pipeline '{pipeline_id}' not found"
        )

    return PIPELINES[pipeline_id]["last_run"]

@mcp.tool()
def list_failed_pipelines(hours: int = 24) -> str:
    """List all pipelines that failed recently.

    Args:
        hours: Look back period in hours (default: 24)

    Returns:
        List of failed pipelines with error details
    """
    failures = []
    for pipeline_id, data in PIPELINES.items():
        if data["last_run"].status == PipelineStatus.FAILED:
            failures.append({
                "pipeline": pipeline_id,
                "run_id": data["last_run"].run_id,
                "error": data["last_run"].error_message,
                "failed_at": data["last_run"].finished_at,
                "owner": data["owner"]
            })

    return json.dumps(failures, indent=2)

@mcp.tool()
def list_all_pipelines() -> str:
    """List all registered pipelines with their current status.

    Returns:
        Summary of all pipelines
    """
    summary = []
    for pipeline_id, data in PIPELINES.items():
        summary.append({
            "pipeline": pipeline_id,
            "status": data["last_run"].status.value,
            "schedule": data["schedule"],
            "owner": data["owner"],
            "last_run": data["last_run"].started_at
        })

    return json.dumps(summary, indent=2)

@mcp.tool()
async def trigger_pipeline(pipeline_id: str, ctx: Context) -> str:
    """Trigger a manual run of a pipeline.

    Args:
        pipeline_id: Pipeline to trigger
        ctx: Context for progress reporting

    Returns:
        Confirmation with new run ID
    """
    if pipeline_id not in PIPELINES:
        return json.dumps({"error": f"Pipeline '{pipeline_id}' not found"})

    await ctx.info(f"Triggering pipeline: {pipeline_id}")

    # In real implementation, call your orchestrator's API
    new_run_id = f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

    return json.dumps({
        "status": "triggered",
        "pipeline": pipeline_id,
        "run_id": new_run_id,
        "message": f"Pipeline {pipeline_id} triggered successfully"
    }, indent=2)

if __name__ == "__main__":
    mcp.run(transport="stdio")

from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel, Field
from enum import Enum
import json

mcp = FastMCP("DataQualityMCP")

class CheckType(str, Enum):
    COMPLETENESS = "completeness"
    UNIQUENESS = "uniqueness"
    FRESHNESS = "freshness"
    VALIDITY = "validity"
    CONSISTENCY = "consistency"

class QualityCheckResult(BaseModel):
    """Result of a single quality check."""
    check_name: str
    check_type: CheckType
    table_name: str
    passed: bool
    score: float = Field(ge=0, le=1, description="Score between 0 and 1")
    rows_checked: int
    issues_found: int
    details: str | None = None

# Simulated quality scores - replace with Great Expectations/Elementary/custom API
QUALITY_SCORES = {
    "sales.orders": {
        "overall": 0.94,
        "completeness": 0.99,
        "uniqueness": 0.95,
        "freshness": 0.88,
        "validity": 0.96
    },
    "sales.customers": {
        "overall": 0.87,
        "completeness": 0.92,
        "uniqueness": 0.98,
        "freshness": 0.75,
        "validity": 0.85
    }
}

@mcp.tool()
async def run_quality_check(
    table_name: str,
    check_type: CheckType,
    ctx: Context
) -> QualityCheckResult:
    """Run a specific data quality check on a table.

    Args:
        table_name: Table to validate
        check_type: Type of quality check to run
        ctx: Context for progress reporting

    Returns:
        Detailed check results
    """
    await ctx.info(f"Running {check_type.value} check on {table_name}")

    # Simulate check execution
    await ctx.report_progress(progress=0.5, total=1.0, message="Scanning table...")

    # In real implementation, call your quality framework
    score = QUALITY_SCORES.get(table_name, {}).get(check_type.value, 0.5)
    passed = score >= 0.9

    await ctx.report_progress(progress=1.0, total=1.0, message="Check complete")

    return QualityCheckResult(
        check_name=f"{check_type.value}_{table_name.replace('.', '_')}",
        check_type=check_type,
        table_name=table_name,
        passed=passed,
        score=score,
        rows_checked=100000,
        issues_found=int((1 - score) * 100000),
        details=f"{'Passed' if passed else 'Failed'}: {score:.1%} of rows meet {check_type.value} criteria"
    )

@mcp.tool()
def get_quality_score(table_name: str) -> str:
    """Get the overall data quality score for a table.

    Args:
        table_name: Table to get scores for

    Returns:
        Quality scores across all dimensions
    """
    if table_name not in QUALITY_SCORES:
        return json.dumps({
            "error": f"No quality scores found for '{table_name}'",
            "available_tables": list(QUALITY_SCORES.keys())
        }, indent=2)

    scores = QUALITY_SCORES[table_name]
    return json.dumps({
        "table": table_name,
        "overall_score": scores["overall"],
        "dimensions": {
            "completeness": scores["completeness"],
            "uniqueness": scores["uniqueness"],
            "freshness": scores["freshness"],
            "validity": scores.get("validity", "not measured")
        },
        "status": "healthy" if scores["overall"] >= 0.9 else "needs attention"
    }, indent=2)

@mcp.tool()
def list_quality_issues(min_severity: str = "warning") -> str:
    """List tables with data quality issues.

    Args:
        min_severity: Minimum severity to include ('warning' or 'critical')

    Returns:
        Tables that need attention
    """
    issues = []
    threshold = 0.8 if min_severity == "critical" else 0.9

    for table_name, scores in QUALITY_SCORES.items():
        if scores["overall"] < threshold:
            # Find the worst dimension
            dimensions = {k: v for k, v in scores.items() if k != "overall"}
            worst_dim = min(dimensions, key=dimensions.get)

            issues.append({
                "table": table_name,
                "overall_score": scores["overall"],
                "severity": "critical" if scores["overall"] < 0.8 else "warning",
                "primary_issue": worst_dim,
                "issue_score": dimensions[worst_dim]
            })

    return json.dumps(sorted(issues, key=lambda x: x["overall_score"]), indent=2)

@mcp.tool()
async def run_full_validation(table_name: str, ctx: Context) -> str:
    """Run all quality checks on a table.

    Args:
        table_name: Table to validate
        ctx: Context for progress reporting

    Returns:
        Complete validation report
    """
    await ctx.info(f"Starting full validation of {table_name}")

    results = []
    checks = list(CheckType)

    for i, check_type in enumerate(checks):
        await ctx.report_progress(
            progress=(i + 1) / len(checks),
            total=1.0,
            message=f"Running {check_type.value} check..."
        )

        result = await run_quality_check(table_name, check_type, ctx)
        results.append({
            "check": result.check_type.value,
            "passed": result.passed,
            "score": result.score
        })

    overall_passed = all(r["passed"] for r in results)

    return json.dumps({
        "table": table_name,
        "validation_passed": overall_passed,
        "checks": results,
        "summary": f"{'All checks passed' if overall_passed else 'Some checks failed'}"
    }, indent=2)

if __name__ == "__main__":
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "my-data-server": {
      "command": "uv",
      "args": ["run", "/full/path/to/server.py"],
      "env": {
        "DATABASE_URL": "your-connection-string"
      }
    }
  }
}

my-data-mcp-server/
├── pyproject.toml      # Dependencies and metadata
├── server.py           # Main server file
├── tools/              # Tool implementations (optional, for large servers)
│   ├── __init__.py
│   ├── catalog.py
│   └── quality.py
├── tests/              # Unit tests for business logic
│   └── test_tools.py
└── README.md           # Setup and usage instructions