Automated Secret Rotation for MCP Servers

The four-stage rotation lifecycle

When Secrets Manager triggers rotation, it invokes the Lambda function four times in sequence. Each invocation receives an event with a Step field naming the current stage. Your handler must dispatch to the correct logic for each step:

import boto3
import json

secrets_client = boto3.client("secretsmanager")

def handler(event: dict, context) -> None:
    arn = event["SecretId"]
    token = event["ClientRequestToken"]  # version token for AWSPENDING
    step = event["Step"]

    if step == "createSecret":
        create_secret(secrets_client, arn, token)
    elif step == "setSecret":
        set_secret(secrets_client, arn, token)
    elif step == "testSecret":
        test_secret(secrets_client, arn, token)
    elif step == "finishSecret":
        finish_secret(secrets_client, arn, token)
    else:
        raise ValueError(f"Unknown rotation step: {step}")

The ClientRequestToken is the version token that Secrets Manager assigned to the AWSPENDING version. Every Secrets Manager API call in your rotation function should use this token when operating on the new version to ensure idempotency — if the Lambda is retried, the same version token prevents duplicate versions from being created.

Implementing each rotation stage

import secrets
import string

def create_secret(client, arn: str, token: str) -> None:
    # Idempotency check — if AWSPENDING already exists with this token, skip
    try:
        client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
        return  # Already created in a previous (failed) invocation
    except client.exceptions.ResourceNotFoundException:
        pass  # Expected — create it now

    # Get the current secret to use as a template for the new value
    current = json.loads(
        client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")["SecretString"]
    )

    # Generate a new password (32 chars, alphanumeric + selected symbols)
    alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
    new_password = "".join(secrets.choice(alphabet) for _ in range(32))

    new_secret = {**current, "password": new_password}

    client.put_secret_value(
        SecretId=arn,
        ClientRequestToken=token,  # links this value to the AWSPENDING version
        SecretString=json.dumps(new_secret),
        VersionStages=["AWSPENDING"],
    )


def set_secret(client, arn: str, token: str) -> None:
    pending = json.loads(
        client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")["SecretString"]
    )
    current = json.loads(
        client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")["SecretString"]
    )

    # Connect to database with CURRENT credentials, update the password
    conn = connect_db(current["host"], current["port"], current["username"], current["password"])
    try:
        conn.execute(
            "ALTER USER %s WITH PASSWORD %s",
            (pending["username"], pending["password"])
        )
        conn.commit()
    finally:
        conn.close()


def test_secret(client, arn: str, token: str) -> None:
    # MUST use AWSPENDING — not AWSCURRENT
    pending = json.loads(
        client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")["SecretString"]
    )

    # Verify the new credential works
    conn = connect_db(pending["host"], pending["port"], pending["username"], pending["password"])
    try:
        conn.execute("SELECT 1")
    finally:
        conn.close()
    # If connect_db raises, Secrets Manager marks the rotation as failed


def finish_secret(client, arn: str, token: str) -> None:
    # Get the current version ID so we can demote it
    metadata = client.describe_secret(SecretId=arn)
    current_version_id = None
    for version_id, stages in metadata["VersionIdsToStages"].items():
        if "AWSCURRENT" in stages and version_id != token:
            current_version_id = version_id
            break

    if current_version_id is None:
        return  # Already finished (idempotent)

    # Promote AWSPENDING to AWSCURRENT, demote old AWSCURRENT to AWSPREVIOUS
    client.update_secret_version_stage(
        SecretId=arn,
        VersionStage="AWSCURRENT",
        MoveToVersionId=token,
        RemoveFromVersionId=current_version_id,
    )

SingleUser vs MultiUser rotation strategies

For database credentials, there are two rotation strategies with fundamentally different availability characteristics:

MultiUser rotation requires a master secret — a separate secret containing admin credentials for the database, used by the rotation Lambda to create and modify the alternating users. The master secret ARN is set in the rotating secret's metadata:

aws secretsmanager rotate-secret \
  --secret-id "arn:aws:secretsmanager:us-east-1:123456789012:secret:prod/mcp-server/database-AbCdEf" \
  --rotation-lambda-arn "arn:aws:lambda:us-east-1:123456789012:function:mcp-secrets-rotation" \
  --rotation-rules AutomaticallyAfterDays=30 \
  --rotate-immediately-on-update

For RDS-specific rotation, AWS provides managed rotation Lambda functions per database engine (PostgreSQL, MySQL, Oracle, SQL Server) that implement both SingleUser and MultiUser strategies.

Rotation schedule expressions

The --rotation-rules parameter accepts either a simple day count or a cron expression for precise scheduling:

# Rotate every 30 days (any time in the 30-day window)
--rotation-rules AutomaticallyAfterDays=30

# Rotate on a specific schedule (cron expression)
--rotation-rules ScheduleExpression="cron(0 2 1 * ? *)"
# Meaning: at 02:00 UTC on the 1st of every month

# Via CDK
secret.add_rotation_schedule(
    "RotationSchedule",
    rotation_lambda=rotation_fn,
    automatically_after=Duration.days(30),
)

For MCP server credentials that are sensitive (OAuth client secrets, payment processor API keys), use a cron expression to schedule rotation at off-peak hours when traffic is lowest — this minimizes the window in which cached credentials are stale.

Monitoring rotation with EventBridge

Secrets Manager emits EventBridge events for each rotation outcome. Catch these to alert on failed rotations before they cause production incidents:

# EventBridge rule to catch rotation failures
{
  "source": ["aws.secretsmanager"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventName": ["RotateSecret"],
    "errorCode": [{"exists": true}]
  }
}

# Or use the dedicated Secrets Manager events (CloudTrail must be enabled)
{
  "source": ["aws.secretsmanager"],
  "detail-type": ["Secret Rotation Event"],
  "detail": {
    "eventName": ["RotationFailed"]
  }
}

Route failed rotation events to an SNS topic → Slack channel. A rotation failure means the old credential is still in use — this is a security gap that needs immediate attention, not something to discover at the next daily standup.

Common failures