Athena Federated Queries from MCP Server Tools

Registering a federated data source for MCP tool access

The standard path for RDS/Aurora sources uses the AthenaJdbcConnector SAR (Serverless Application Repository) app. Deploy it once, then register it as an Athena data source:

import boto3
import time

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

# Register the deployed connector Lambda as an Athena data source
# The connector Lambda was deployed from SAR: AthenaJdbcConnector
athena.create_data_catalog(
    Name="rds-orders-catalog",
    Type="LAMBDA",
    Description="Federated access to RDS orders database via JDBC connector",
    Parameters={
        # ARN of the connector Lambda from SAR deployment
        "function": "arn:aws:lambda:us-east-1:123456789012:function:athena-rds-connector",
    },
)

# Verify the catalog is registered
response = athena.get_data_catalog(Name="rds-orders-catalog")
print(f"Catalog status: {response['DataCatalog']['Type']}")
# Output: Catalog status: LAMBDA

The connector Lambda needs a spill bucket — when a query result exceeds the Lambda 6 MB response limit, Athena spills intermediate data to S3 and continues. Configure this via the Lambda environment variable spill_bucket:

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

lambda_client.update_function_configuration(
    FunctionName="athena-rds-connector",
    Environment={
        "Variables": {
            "spill_bucket": "my-athena-spill-bucket",    # dedicated spill bucket
            "spill_prefix": "federation-spill/",          # objects auto-expire via lifecycle rule
            "jdbc_connection_string": "jdbc:postgresql://rds-host:5432/orders",
            "secret_manager_enabled": "true",
            "secret_name": "rds/orders/credentials",     # Secrets Manager secret for DB creds
        }
    },
)

# Add lifecycle rule to expire spill objects — they accumulate fast
s3 = boto3.client("s3")
s3.put_bucket_lifecycle_configuration(
    Bucket="my-athena-spill-bucket",
    LifecycleConfiguration={
        "Rules": [{
            "ID": "expire-federation-spill",
            "Filter": {"Prefix": "federation-spill/"},
            "Status": "Enabled",
            "Expiration": {"Days": 1},    # spill objects have no value after query completes
        }]
    },
)

Querying federated sources from an MCP tool handler

Once the data source is registered, MCP tool handlers query it using the three-part "catalog"."database"."table" notation alongside regular S3 tables in the AwsDataCatalog:

import boto3
import json
import time

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

def run_federated_query(sql: str, workgroup: str = "primary") -> list[dict]:
    """MCP tool: run Athena query spanning S3 + federated RDS source."""
    # Start the query — Athena fans out to multiple data sources automatically
    start_response = athena.start_query_execution(
        QueryString=sql,
        WorkGroup=workgroup,
        ResultConfiguration={
            "OutputLocation": "s3://my-athena-results/",
        },
    )
    query_execution_id = start_response["QueryExecutionId"]

    # Poll until terminal state
    while True:
        status_response = athena.get_query_execution(
            QueryExecutionId=query_execution_id
        )
        state = status_response["QueryExecution"]["Status"]["State"]

        if state == "SUCCEEDED":
            break
        elif state in ("FAILED", "CANCELLED"):
            reason = status_response["QueryExecution"]["Status"].get(
                "StateChangeReason", "unknown"
            )
            raise RuntimeError(f"Athena query {state}: {reason}")

        time.sleep(2)  # poll at 2s intervals — federated queries are slower than pure S3

    # Paginate through results
    rows = []
    paginator = athena.get_paginator("get_query_results")
    page_iter = paginator.paginate(QueryExecutionId=query_execution_id)

    columns = None
    for page in page_iter:
        result_rows = page["ResultSet"]["Rows"]
        if columns is None:
            # First row in first page is the column header row
            columns = [col["VarCharValue"] for col in result_rows[0]["Data"]]
            result_rows = result_rows[1:]   # skip header row
        for row in result_rows:
            rows.append({
                columns[i]: cell.get("VarCharValue", None)
                for i, cell in enumerate(row["Data"])
            })

    return rows

# Cross-source join: S3 data lake (orders) + RDS live database (customers)
# "AwsDataCatalog" is the default catalog for S3/Glue tables
# "rds-orders-catalog" is the federated connector registered above
sql = """
SELECT
    o.order_id,
    o.order_date,
    o.total_amount,
    c.customer_name,
    c.email
FROM "AwsDataCatalog"."analytics"."orders" AS o
JOIN "rds-orders-catalog"."public"."customers" AS c
    ON o.customer_id = c.id
WHERE o.order_date >= DATE '2026-09-01'
  AND o.status = 'completed'
ORDER BY o.total_amount DESC
LIMIT 100
"""
results = run_federated_query(sql)

Federated queries that cross data sources always serialize through Athena's query coordinator. Cross-source JOINs pull the smaller side (typically the RDS dimension table) into Athena's memory as a broadcast hash join. For MCP tools, keep the federated side of the JOIN small — scan filters on the federated source are pushed down to the connector Lambda, so a predicate like WHERE c.id IN (...) is translated to a parameterized SQL query against RDS rather than a full table scan.

DynamoDB and custom connectors

The AthenaDynamoDBConnector SAR app enables Athena queries against DynamoDB tables without ETL. It converts SQL predicates to DynamoDB KeyConditionExpression and FilterExpression calls. Key behaviors an MCP tool author needs to know:

# DynamoDB connector: SAR app "AthenaDynamoDBConnector"
# After deployment, register it:
athena.create_data_catalog(
    Name="ddb-events-catalog",
    Type="LAMBDA",
    Parameters={
        "function": "arn:aws:lambda:us-east-1:123456789012:function:athena-ddb-connector",
    },
)

# DynamoDB table schema must be defined in Glue Data Catalog
# (connector reads Glue metadata to project column names/types)
# Query example — equality on partition key is pushed down to DDB KeyCondition
sql = """
SELECT event_type, event_timestamp, payload
FROM "ddb-events-catalog"."events_db"."user_events"
WHERE user_id = '550e8400-e29b-41d4-a716-446655440000'
  AND event_timestamp >= '2026-09-01T00:00:00Z'
"""

# WARNING: queries without partition key equality result in a full DDB table scan
# via Scan API — expensive and slow for large tables.
# Always filter on the partition key when querying DynamoDB via Athena federation.

For MCP tools that need to query proprietary APIs or on-premise databases, build a custom connector using the athena-federation-sdk Java library. The connector Lambda implements two handlers: MetadataHandler (lists databases, tables, columns) and RecordHandler (streams data rows for a given split). Splits are the unit of parallelism — each split becomes one Lambda invocation. A well-written connector partitions data into 100–500 splits for large queries.

Performance patterns for MCP analytics tools

Federated queries are slower than native S3 queries because each split invokes a Lambda function and waits for it to return data. Tuning strategies for MCP tools that call federated sources frequently:

# Pattern 1: Materialized S3 snapshot — run federation query nightly,
# write results to S3 Parquet, then serve MCP tool reads from S3 only
def refresh_materialized_snapshot() -> None:
    """Scheduled job: refresh S3 copy of RDS dimension tables."""
    sql = """
    CREATE TABLE "AwsDataCatalog"."analytics"."customers_snapshot"
    WITH (
        format = 'PARQUET',
        parquet_compression = 'SNAPPY',
        external_location = 's3://my-datalake/snapshots/customers/',
        partitioned_by = ARRAY['snapshot_date']
    )
    AS SELECT *, CAST(CURRENT_DATE AS VARCHAR) AS snapshot_date
       FROM "rds-orders-catalog"."public"."customers"
    """
    run_federated_query(sql, workgroup="etl-workgroup")

# MCP tools then query the S3 snapshot instead of hitting RDS directly
# — sub-second latency vs 5-30 seconds for live federation queries

# Pattern 2: Federated query with explicit split count hint
# (connector-specific — check connector docs for SPLIT_FACTOR hint)
sql_with_hints = """
SELECT /*+ SPLIT_FACTOR=50 */ order_id, total
FROM "rds-orders-catalog"."public"."large_orders_table"
WHERE created_at >= DATE '2026-09-01'
"""

# Pattern 3: Result set S3 caching in MCP tool layer
# Store QueryExecutionId → result mapping with a TTL
# Subsequent identical queries re-use the existing S3 result file
import hashlib

def run_with_cache(sql: str, cache_ttl_seconds: int = 3600) -> list[dict]:
    """Cache Athena results by SQL hash to avoid redundant federated queries."""
    sql_hash = hashlib.sha256(sql.encode()).hexdigest()[:16]
    # Check S3 for cached result (implementation detail: store metadata in DDB or SSM)
    cached_id = lookup_cached_execution(sql_hash)
    if cached_id:
        # Re-use existing result file — no charge for GetQueryResults on completed queries
        return fetch_query_results(cached_id)
    results = run_federated_query(sql)
    store_cached_execution(sql_hash, results, ttl=cache_ttl_seconds)
    return results

For MCP tools that must query live operational data (not snapshots), keep the federated side of every query as narrow as possible. Push WHERE predicates that match the operational source's indexes — the connector translates these to the native query language. A federated query that hits a DynamoDB table without a partition key predicate invokes a full-table Scan across all shards, which is both slow and expensive.