Athena Workgroups for Cost Control in MCP Servers

Creating isolated workgroups for MCP tool categories

Each MCP tool (or tool category) gets its own workgroup, its own S3 result prefix, and its own cost limits:

import boto3

athena = boto3.client("athena", region_name="us-east-1")

def create_mcp_workgroup(
    name: str,
    result_bucket: str,
    result_prefix: str,
    bytes_scanned_limit: int = 10 * 1024**3,  # default 10 GB per query
    description: str = "",
) -> None:
    """Create an isolated Athena workgroup for an MCP tool or tool group."""
    athena.create_work_group(
        Name=name,
        Description=description,
        Configuration={
            "ResultConfiguration": {
                "OutputLocation": f"s3://{result_bucket}/{result_prefix}/",
                "EncryptionConfiguration": {
                    "EncryptionOption": "SSE_S3",   # SSE_KMS for regulated data
                },
            },
            # EnforceWorkGroupConfiguration: True means the workgroup's OutputLocation
            # overrides whatever OutputLocation the client passes in start_query_execution.
            # Without this, any caller can route results to any bucket.
            "EnforceWorkGroupConfiguration": True,
            "PublishCloudWatchMetricsEnabled": True,
            "BytesScannedCutoffPerQuery": bytes_scanned_limit,
            "RequesterPaysEnabled": False,
            "EngineVersion": {
                "SelectedEngineVersion": "Athena engine version 3",
            },
        },
        Tags=[
            {"Key": "mcp-tool", "Value": name},
            {"Key": "cost-center", "Value": "analytics"},
        ],
    )

# Create workgroups for different MCP tool categories
create_mcp_workgroup(
    name="mcp-reporting-tools",
    result_bucket="my-athena-results",
    result_prefix="reporting",
    bytes_scanned_limit=50 * 1024**3,  # 50 GB — reporting can be expensive
    description="Workgroup for MCP reporting and dashboard tools",
)

create_mcp_workgroup(
    name="mcp-exploration-tools",
    result_bucket="my-athena-results",
    result_prefix="exploration",
    bytes_scanned_limit=5 * 1024**3,   # 5 GB — keep ad-hoc exploration cheap
    description="Workgroup for MCP ad-hoc query and exploration tools",
)

create_mcp_workgroup(
    name="mcp-etl-jobs",
    result_bucket="my-athena-results",
    result_prefix="etl",
    bytes_scanned_limit=500 * 1024**3,  # 500 GB — ETL jobs need higher limits
    description="Workgroup for MCP-triggered ETL and CTAS operations",
)

Routing queries to the correct workgroup from MCP handlers

MCP tool handlers specify the workgroup in start_query_execution. Because EnforceWorkGroupConfiguration=True is set, the workgroup's OutputLocation overrides any client-supplied value:

import time

def run_mcp_query(
    sql: str,
    workgroup: str,
    timeout_seconds: int = 300,
) -> tuple[list[dict], dict]:
    """Run Athena query in a specific workgroup; return rows + cost metadata."""
    start_resp = athena.start_query_execution(
        QueryString=sql,
        WorkGroup=workgroup,
        # ResultConfiguration is ignored when EnforceWorkGroupConfiguration=True
        # but boto3 requires it — use a placeholder that will be overridden
        ResultConfiguration={"OutputLocation": "s3://ignored/by-workgroup/"},
    )
    qid = start_resp["QueryExecutionId"]

    deadline = time.time() + timeout_seconds
    while time.time() < deadline:
        exec_resp = athena.get_query_execution(QueryExecutionId=qid)
        state = exec_resp["QueryExecution"]["Status"]["State"]

        if state == "SUCCEEDED":
            stats = exec_resp["QueryExecution"]["Statistics"]
            cost_metadata = {
                "query_execution_id": qid,
                "data_scanned_bytes": stats.get("DataScannedInBytes", 0),
                "engine_execution_ms": stats.get("EngineExecutionTimeInMillis", 0),
                "total_execution_ms": stats.get("TotalExecutionTimeInMillis", 0),
                # Athena pricing: $5 per TB scanned (rounded up to 10 MB minimum)
                "estimated_cost_usd": max(stats.get("DataScannedInBytes", 0), 10 * 1024**2)
                    / (1024**4) * 5.0,
            }
            rows = fetch_all_results(qid)
            return rows, cost_metadata

        elif state in ("FAILED", "CANCELLED"):
            reason = exec_resp["QueryExecution"]["Status"].get("StateChangeReason", "")
            # BytesScannedCutoffPerQuery exceeded → CANCELLED with specific reason
            if "BytesScannedCutoffPerQuery" in reason:
                raise RuntimeError(
                    f"Query cancelled: exceeded workgroup data scan limit. "
                    f"Add a WHERE clause or partition filter to reduce scanned data."
                )
            raise RuntimeError(f"Athena query {state}: {reason}")

        time.sleep(2)

    athena.stop_query_execution(QueryExecutionId=qid)
    raise TimeoutError(f"Athena query exceeded {timeout_seconds}s timeout")


def fetch_all_results(qid: str) -> list[dict]:
    """Paginate through Athena query results."""
    paginator = athena.get_paginator("get_query_results")
    rows = []
    columns = None
    for page in paginator.paginate(QueryExecutionId=qid):
        result_rows = page["ResultSet"]["Rows"]
        if columns is None:
            columns = [col["VarCharValue"] for col in result_rows[0]["Data"]]
            result_rows = result_rows[1:]
        for row in result_rows:
            rows.append({columns[i]: cell.get("VarCharValue") for i, cell in enumerate(row["Data"])})
    return rows

Per-query cost tracking and alerting

Every call to get_query_execution on a completed query returns DataScannedInBytes. Log this per MCP tool call to build a real-time cost dashboard:

import boto3

cloudwatch = boto3.client("cloudwatch", region_name="us-east-1")

def emit_query_cost_metrics(
    tool_name: str,
    workgroup: str,
    data_scanned_bytes: int,
    engine_execution_ms: int,
) -> None:
    """Emit per-tool Athena cost metrics to CloudWatch for alerting."""
    cloudwatch.put_metric_data(
        Namespace="MCPServer/AthenaCosts",
        MetricData=[
            {
                "MetricName": "DataScannedBytes",
                "Dimensions": [
                    {"Name": "ToolName", "Value": tool_name},
                    {"Name": "WorkGroup", "Value": workgroup},
                ],
                "Value": data_scanned_bytes,
                "Unit": "Bytes",
            },
            {
                "MetricName": "QueryDurationMs",
                "Dimensions": [
                    {"Name": "ToolName", "Value": tool_name},
                    {"Name": "WorkGroup", "Value": workgroup},
                ],
                "Value": engine_execution_ms,
                "Unit": "Milliseconds",
            },
        ],
    )

# Athena also publishes workgroup-level metrics to CloudWatch automatically
# (when PublishCloudWatchMetricsEnabled=True) in the AWS/Athena namespace:
# - DMLQueriesRunning, DMLQueriesFailed, DMLQueriesSucceeded
# - BytesScannedPerQuery (distribution)
# Create alarms on BytesScannedPerQuery to detect runaway queries before the bill lands

For multi-tenant MCP servers where each tenant's queries should be tracked separately, create one workgroup per tenant tier (or per tenant if the tenant count is small). Tag workgroups with tenant identifiers for AWS Cost Explorer chargeback reports. The ListWorkGroups API returns all workgroups — iterate these to build a per-workgroup cost summary at the end of each billing period.

Workgroup query result encryption and compliance

For MCP servers handling regulated data, workgroups enforce query result encryption at rest — results written to S3 are always encrypted, regardless of what the client code requests:

# For HIPAA/PCI workloads: use SSE_KMS with a dedicated CMK
kms = boto3.client("kms", region_name="us-east-1")
key_response = kms.create_key(
    Description="Athena result encryption key for MCP medical tools",
    Tags=[{"TagKey": "compliance", "TagValue": "hipaa"}],
)
kms_key_arn = key_response["KeyMetadata"]["Arn"]

athena.create_work_group(
    Name="mcp-medical-tools",
    Configuration={
        "ResultConfiguration": {
            "OutputLocation": "s3://my-secure-results/medical/",
            "EncryptionConfiguration": {
                "EncryptionOption": "SSE_KMS",
                "KmsKey": kms_key_arn,
            },
            "AclConfiguration": {
                "S3AclOption": "BUCKET_OWNER_FULL_CONTROL",
            },
        },
        "EnforceWorkGroupConfiguration": True,
        "PublishCloudWatchMetricsEnabled": True,
        "BytesScannedCutoffPerQuery": 10 * 1024**3,
    },
)

# Update existing workgroup — e.g. raise the scan limit for a specific tool
athena.update_work_group(
    WorkGroup="mcp-exploration-tools",
    ConfigurationUpdates={
        "BytesScannedCutoffPerQuery": 20 * 1024**3,  # raise from 5 GB to 20 GB
        "EnforceWorkGroupConfiguration": True,
    },
)

Workgroup isolation means that compromised MCP tool credentials (a leaked API key) can only read query results from that tool's dedicated result prefix — not from other tools' result buckets. Combine workgroup isolation with IAM policies that grant athena:GetQueryResults only on the workgroup's specific result bucket prefix for defense-in-depth.