Automating Enterprise SFTP Ingestion with AWS Transfer Family

import os, json, boto3

transfer = boto3.client("transfer")

CONNECTOR_ID   = os.environ["CONNECTOR_ID"]
REMOTE_DIR     = os.environ["REMOTE_DIR"]        # e.g. /exports/daily
LISTING_OUTPUT = os.environ["LISTING_OUTPUT"]    # e.g. /my-bucket/_listings


def lambda_handler(event, context):
    """Ask the connector what is actually on the remote server.

    Deliberately does NOT guess a filename from today's date: vendors are late,
    rename files, and backfill. The listing is written to S3 and a
    'SFTP Connector Directory Listing Completed' event follows.
    """
    resp = transfer.start_directory_listing(
        ConnectorId=CONNECTOR_ID,
        RemoteDirectoryPath=REMOTE_DIR,
        OutputDirectoryPath=LISTING_OUTPUT,
        MaxItems=1000,                 # one level deep only; no recursion
    )
    print(json.dumps({"listing_id": resp["ListingId"], "output_file": resp["OutputFileName"]}))
    return {"listingId": resp["ListingId"], "outputFileName": resp["OutputFileName"]}

{
  "source": "aws.transfer",
  "detail-type": "SFTP Connector Directory Listing Completed",
  "detail": {
    "operation": "LIST",
    "connector-id": "c-0123456789abcdef0",
    "listing-id": "6666abcd-11aa-22bb-cc33-EXAMPLE0000",
    "remote-directory-path": "/exports/daily",
    "output-file-location": {
      "domain": "S3",
      "bucket": "delantech-sftp-landing-zone",
      "key": "_listings/c-0123456789abcdef0-6666abcd-11aa-22bb-cc33-EXAMPLE0000.json"
    },
    "status-code": "COMPLETED"
  }
}

Skipping files already in the landing zone is what makes a re-run safe: the schedule can fire twice and the second run queues nothing.

import os, json, boto3

s3       = boto3.client("s3")
transfer = boto3.client("transfer")

CONNECTOR_ID = os.environ["CONNECTOR_ID"]
LOCAL_DIR    = os.environ["LOCAL_DIR"]       # /bucket/prefix in S3

# The landing zone is described by LOCAL_DIR, so derive it rather than carrying
# a second, separately-configurable bucket name that can drift out of step.
LANDING_BUCKET, _, LANDING_PREFIX = LOCAL_DIR.lstrip("/").partition("/")

MAX_PATHS_PER_CALL = 10          # hard AWS limit on RetrieveFilePaths


def lambda_handler(event, context):
    """Triggered by the directory-listing-completed event.

    Reads the listing JSON the connector wrote, decides what is new, and asks
    for exactly those paths. Everything already in the landing zone is skipped,
    which is what makes a re-run safe.
    """
    # The EVENT gives output-file-location {domain, bucket, key}. Do not reach
    # for output-file-name — that belongs to the StartDirectoryListing API
    # RESPONSE, not to this event, and is not present here.
    location = event["detail"]["output-file-location"]

    listing = json.loads(
        s3.get_object(Bucket=location["bucket"], Key=location["key"])["Body"].read()
    )
    remote_files = [f["filePath"] for f in listing.get("files", [])]

    wanted = [p for p in remote_files if p.endswith(".csv") and not already_landed(p)]
    if not wanted:
        print("nothing new on the remote server")
        return {"queued": 0}

    # StartFileTransfer accepts a MAXIMUM OF 10 paths per call, so chunk.
    # Passing the whole list fails validation the first morning the vendor
    # ships eleven files — which is exactly when you are not watching.
    transfer_ids = []
    for i in range(0, len(wanted), MAX_PATHS_PER_CALL):
        batch = wanted[i:i + MAX_PATHS_PER_CALL]
        resp = transfer.start_file_transfer(
            ConnectorId=CONNECTOR_ID,
            RetrieveFilePaths=batch,
            LocalDirectoryPath=LOCAL_DIR,
        )
        transfer_ids.append(resp["TransferId"])

    print(json.dumps({"transfer_ids": transfer_ids, "count": len(wanted)}))
    return {"transferIds": transfer_ids, "queued": len(wanted)}


def already_landed(remote_path):
    """True if this file is already in the landing zone."""
    key = f"{LANDING_PREFIX}/{os.path.basename(remote_path)}".lstrip("/")
    try:
        s3.head_object(Bucket=LANDING_BUCKET, Key=key)
        return True
    except s3.exceptions.ClientError:
        return False

{
  "version": "0",
  "source": "aws.transfer",
  "detail-type": "SFTP Connector File Retrieve Completed",
  "detail": {
    "operation": "RETRIEVE",
    "connector-id": "c-0123456789abcdef0",
    "transfer-id": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111",
    "file-transfer-id": "11112222-3333-4444-5555-EXAMPLE22222",
    "url": "sftp://partner.example.com",
    "file-path": "/exports/daily/orders_2026-04-25.csv",
    "local-directory-path": "/delantech-sftp-landing-zone/imports",
    "status-code": "COMPLETED",
    "bytes": 63533
  }
}

Failures carry both failure-code (a stable enum such as RETRIEVE_FILE_NOT_FOUND or CONNECTION_ERROR) and a human-readable failure-message. Alert on the message; branch on the code.

{
  "source": "aws.transfer",
  "detail-type": "SFTP Connector File Retrieve Failed",
  "detail": {
    "operation": "RETRIEVE",
    "connector-id": "c-0123456789abcdef0",
    "transfer-id": "a1b2c3d4-5678-90ab-cdef-EXAMPLE11111",
    "file-path": "/exports/daily/orders_2026-04-25.csv",
    "status-code": "FAILED",
    "failure-code": "RETRIEVE_FILE_NOT_FOUND",
    "failure-message": "File not found on remote server"
  }
}

{
  "source": ["aws.transfer"],
  "detail-type": ["SFTP Connector File Retrieve Completed"],
  "detail": {
    "connector-id": ["c-0123456789abcdef0"]
  }
}

{
  "source": ["aws.transfer"],
  "detail-type": ["SFTP Connector File Retrieve Failed"]
}

input_paths = {
  destination = "$.detail.local-directory-path"
  source      = "$.detail.file-path"
  transferId  = "$.detail.transfer-id"
}

input_template = {
  "pipeline":    "delantech-sftp-ingestion",
  "destination": "<destination>",
  "source":      "<source>",
  "transferId":  "<transferId>"
}

Note the notifier swallows its own errors. If Slack is unreachable you want the transfer failure recorded, not a second failure obscuring it.

import os, json, urllib.request, urllib.error

WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]


def lambda_handler(event, context):
    # The detail is FLAT — one event per file, no file-transfers[] array.
    d = event.get("detail", {})
    connector   = d.get("connector-id", "unknown")
    transfer_id = d.get("transfer-id", "n/a")
    file_path   = d.get("file-path", "n/a")
    code        = d.get("failure-code", "UNKNOWN")
    failure     = d.get("failure-message") or "No reason supplied by the remote host."

    body = {
        "text": f"SFTP ingestion failure: {code}",
        "attachments": [{
            "color": "#dc2626",
            "fields": [
                {"title": "Connector",    "value": connector,   "short": True},
                {"title": "Transfer ID",  "value": transfer_id, "short": True},
                {"title": "File",         "value": file_path,   "short": False},
                {"title": "Failure code", "value": code,        "short": True},
                {"title": "Reason",       "value": failure,     "short": False},
            ],
        }],
    }

    req = urllib.request.Request(
        WEBHOOK,
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            return {"slack_status": r.getcode()}
    except urllib.error.URLError as e:
        # Never let the notifier mask the original failure — log and move on.
        print(f"could not reach Slack: {e}")
        return {"slack_status": "unreachable"}

Trimmed to the parts that matter: the Lambda function and role definitions follow the usual pattern and are omitted.

provider "aws" {
  region = "us-east-1"
}

resource "aws_s3_bucket" "landing" {
  bucket = "delantech-sftp-landing-zone"
}

# --- IAM for the connector: write to S3, read the SFTP credentials ---
resource "aws_iam_role" "connector" {
  name = "delantech-sftp-connector-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect    = "Allow"
      Action    = "sts:AssumeRole"
      Principal = { Service = "transfer.amazonaws.com" }
    }]
  })
}

resource "aws_iam_role_policy" "connector" {
  name = "delantech-sftp-connector-policy"
  role = aws_iam_role.connector.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect   = "Allow"
        Action   = ["s3:PutObject", "s3:GetObject", "s3:ListBucket"]
        Resource = [aws_s3_bucket.landing.arn, "${aws_s3_bucket.landing.arn}/*"]
      },
      {
        Effect   = "Allow"
        Action   = ["secretsmanager:GetSecretValue"]
        Resource = [aws_secretsmanager_secret.sftp_creds.arn]
      }
    ]
  })
}

resource "aws_secretsmanager_secret" "sftp_creds" {
  name = "delantech/sftp/partner-credentials"
}

resource "aws_transfer_connector" "partner" {
  url          = "sftp://partner.example.com"
  access_role  = aws_iam_role.connector.arn

  sftp_config {
    user_secret_id      = aws_secretsmanager_secret.sftp_creds.arn
    trusted_host_keys   = [var.partner_host_key]
  }
}

# --- Route the two outcomes to two different places ---
resource "aws_cloudwatch_event_rule" "retrieve_completed" {
  name = "delantech-sftp-retrieve-completed"
  event_pattern = jsonencode({
    source        = ["aws.transfer"]
    "detail-type" = ["SFTP Connector File Retrieve Completed"]
    detail        = { "connector-id" = [aws_transfer_connector.partner.id] }
  })
}

resource "aws_cloudwatch_event_target" "to_step_functions" {
  rule      = aws_cloudwatch_event_rule.retrieve_completed.name
  target_id = "StepFunctions"
  arn       = aws_sfn_state_machine.processor.arn
  role_arn  = aws_iam_role.events_invoke.arn

  input_transformer {
    input_paths = {
      destination = "$.detail.local-directory-path"
      source      = "$.detail.file-path"
      transferId  = "$.detail.transfer-id"
    }
    input_template = jsonencode({
      pipeline    = "delantech-sftp-ingestion"
      destination = "<destination>"
      source      = "<source>"
      transferId  = "<transferId>"
    })
  }
}

resource "aws_cloudwatch_event_rule" "retrieve_failed" {
  name = "delantech-sftp-retrieve-failed"
  event_pattern = jsonencode({
    source        = ["aws.transfer"]
    "detail-type" = ["SFTP Connector File Retrieve Failed"]
  })
}

resource "aws_cloudwatch_event_target" "to_slack" {
  rule      = aws_cloudwatch_event_rule.retrieve_failed.name
  target_id = "SlackNotifier"
  arn       = aws_lambda_function.slack_notifier.arn
}

resource "aws_lambda_permission" "allow_events" {
  statement_id  = "AllowExecutionFromEventBridge"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.slack_notifier.function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.retrieve_failed.arn
}

# --- Schedule ---
resource "aws_scheduler_schedule" "daily" {
  name                         = "delantech-sftp-daily"
  schedule_expression          = "cron(0 6 * * ? *)"
  schedule_expression_timezone = "Europe/London"
  flexible_time_window { mode = "OFF" }

  target {
    arn      = aws_lambda_function.list_remote.arn
    role_arn = aws_iam_role.scheduler.arn
  }
}