Amazon Connect: Instances, Queues, Routing Profiles & Phone Numbers

Inbound Call
    ↓
Phone Number → Contact Flow (IVR logic)
    ↓
Queue (e.g., "appointment-scheduling")
    ↓
Routing Profile (maps agents to queues)
    ↓
Agent

# connect.tf

resource "aws_connect_instance" "main" {
  identity_management_type = "CONNECT_MANAGED"
  inbound_calls_enabled    = true
  outbound_calls_enabled   = true
  instance_alias           = "myoptometry-${var.env}"

  tags = { Environment = var.env }
}

# Enable call recording storage
resource "aws_connect_instance_storage_config" "recordings" {
  instance_id  = aws_connect_instance.main.id
  resource_type = "CALL_RECORDINGS"

  storage_config {
    storage_type = "S3"
    s3_config {
      bucket_name   = aws_s3_bucket.recordings.bucket
      bucket_prefix = "recordings/"
      encryption_config {
        encryption_type = "KMS"
        key_id          = aws_kms_key.connect.arn
      }
    }
  }
}

resource "aws_connect_hours_of_operation" "business_hours" {
  instance_id = aws_connect_instance.main.id
  name        = "Business Hours"
  description = "Monday to Friday 8AM-6PM EST"
  time_zone   = "America/New_York"

  config {
    day = "MONDAY"
    start_time { hours = 8;  minutes = 0 }
    end_time   { hours = 18; minutes = 0 }
  }
  config {
    day = "TUESDAY"
    start_time { hours = 8;  minutes = 0 }
    end_time   { hours = 18; minutes = 0 }
  }
  config {
    day = "WEDNESDAY"
    start_time { hours = 8;  minutes = 0 }
    end_time   { hours = 18; minutes = 0 }
  }
  config {
    day = "THURSDAY"
    start_time { hours = 8;  minutes = 0 }
    end_time   { hours = 18; minutes = 0 }
  }
  config {
    day = "FRIDAY"
    start_time { hours = 8;  minutes = 0 }
    end_time   { hours = 18; minutes = 0 }
  }
}

resource "aws_connect_queue" "scheduling" {
  instance_id           = aws_connect_instance.main.id
  name                  = "appointment-scheduling"
  description           = "Handles appointment booking calls"
  hours_of_operation_id = aws_connect_hours_of_operation.business_hours.hours_of_operation_id
  max_contacts          = 50   # max queued calls before overflow

  outbound_caller_config {
    outbound_caller_id_name = "Sunrise Eye Care"
  }
}

resource "aws_connect_queue" "insurance" {
  instance_id           = aws_connect_instance.main.id
  name                  = "insurance-verification"
  description           = "Insurance verification inquiries"
  hours_of_operation_id = aws_connect_hours_of_operation.business_hours.hours_of_operation_id
  max_contacts          = 20
}

resource "aws_connect_routing_profile" "general" {
  instance_id               = aws_connect_instance.main.id
  name                      = "general-agents"
  description               = "Handles scheduling and general inquiries"
  default_outbound_queue_id = aws_connect_queue.scheduling.queue_id

  media_concurrencies {
    channel     = "VOICE"
    concurrency = 1  # one call at a time
  }

  queue_configs {
    channel  = "VOICE"
    delay    = 0
    priority = 1
    queue_id = aws_connect_queue.scheduling.queue_id
  }

  queue_configs {
    channel  = "VOICE"
    delay    = 30  # only route here if primary queue has >30s wait
    priority = 2
    queue_id = aws_connect_queue.insurance.queue_id
  }
}

import boto3
import os

connect = boto3.client("amazon-connect", region_name=os.environ["AWS_REGION"])
INSTANCE_ID = os.environ["CONNECT_INSTANCE_ID"]


def create_clinic_queue(clinic_name: str, hours_id: str) -> str:
    """Create a dedicated queue for a new clinic."""
    response = connect.create_queue(
        InstanceId=INSTANCE_ID,
        Name=f"clinic-{clinic_name.lower().replace(' ', '-')}",
        Description=f"Queue for {clinic_name} patients",
        HoursOfOperationId=hours_id,
        MaxContacts=30,
        Tags={"clinic": clinic_name}
    )
    return response["QueueId"]


def create_clinic_routing_profile(clinic_name: str, queue_id: str, 
                                    default_queue_id: str) -> str:
    """Create a routing profile for clinic-specific agents."""
    response = connect.create_routing_profile(
        InstanceId=INSTANCE_ID,
        Name=f"routing-{clinic_name.lower().replace(' ', '-')}",
        Description=f"Routing for {clinic_name} agents",
        DefaultOutboundQueueId=queue_id,
        MediaConcurrencies=[{"Channel": "VOICE", "Concurrency": 1}],
        QueueConfigs=[
            {
                "QueueReference": {"QueueId": queue_id, "Channel": "VOICE"},
                "Priority": 1,
                "Delay": 0
            }
        ]
    )
    return response["RoutingProfileId"]


def claim_phone_number(toll_free: bool = False) -> dict:
    """Claim a US phone number for the Connect instance."""
    response = connect.search_available_phone_numbers(
        TargetArn=f"arn:aws:connect:{os.environ['AWS_REGION']}:{os.environ['ACCOUNT_ID']}:instance/{INSTANCE_ID}",
        PhoneNumberCountryCode="US",
        PhoneNumberType="TOLL_FREE" if toll_free else "DID"
    )
    
    if not response["AvailableNumbersList"]:
        raise Exception("No phone numbers available")
    
    number = response["AvailableNumbersList"][0]["PhoneNumber"]
    
    claim_response = connect.claim_phone_number(
        TargetArn=f"arn:aws:connect:{os.environ['AWS_REGION']}:{os.environ['ACCOUNT_ID']}:instance/{INSTANCE_ID}",
        PhoneNumber=number,
        Tags={"purpose": "clinic-inbound"}
    )
    
    return {
        "phone_number": number,
        "phone_number_id": claim_response["PhoneNumberId"]
    }


def onboard_clinic(clinic_name: str, hours_id: str) -> dict:
    """Full clinic onboarding: queue + routing profile + phone number."""
    queue_id = create_clinic_queue(clinic_name, hours_id)
    routing_id = create_clinic_routing_profile(clinic_name, queue_id, queue_id)
    phone = claim_phone_number()
    
    return {
        "clinic_name": clinic_name,
        "queue_id": queue_id,
        "routing_profile_id": routing_id,
        "phone_number": phone["phone_number"],
        "phone_number_id": phone["phone_number_id"]
    }

def get_queue_metrics(queue_ids: list[str]) -> list[dict]:
    """Get current queue depths and wait times."""
    response = connect.get_current_metric_data(
        InstanceId=INSTANCE_ID,
        Filters={
            "Queues": queue_ids,
            "Channels": ["VOICE"]
        },
        Groupings=["QUEUE"],
        CurrentMetrics=[
            {"Name": "CONTACTS_IN_QUEUE",      "Unit": "COUNT"},
            {"Name": "OLDEST_CONTACT_AGE",      "Unit": "SECONDS"},
            {"Name": "AGENTS_AVAILABLE",        "Unit": "COUNT"},
            {"Name": "AGENTS_ON_CALL",          "Unit": "COUNT"},
        ]
    )
    
    results = []
    for collection in response["MetricResults"]:
        queue_id = collection["Dimensions"]["Queue"]["Id"]
        metrics = {m["Metric"]["Name"]: m["Value"] for m in collection["Metrics"]}
        results.append({
            "queue_id": queue_id,
            "contacts_in_queue": metrics.get("CONTACTS_IN_QUEUE", 0),
            "oldest_contact_age_seconds": metrics.get("OLDEST_CONTACT_AGE", 0),
            "agents_available": metrics.get("AGENTS_AVAILABLE", 0),
            "agents_on_call": metrics.get("AGENTS_ON_CALL", 0),
        })
    
    return results

from datetime import datetime, timedelta

def get_daily_call_stats(queue_id: str, date: str) -> dict:
    """Get call volume and handle time for a specific date."""
    start = datetime.strptime(date, "%Y-%m-%d")
    end = start + timedelta(days=1)
    
    response = connect.get_metric_data(
        InstanceId=INSTANCE_ID,
        StartTime=start,
        EndTime=end,
        Filters={"Queues": [queue_id], "Channels": ["VOICE"]},
        Groupings=["QUEUE"],
        HistoricalMetrics=[
            {"Name": "CONTACTS_HANDLED",        "Unit": "COUNT",   "Statistic": "SUM"},
            {"Name": "CONTACTS_ABANDONED",       "Unit": "COUNT",   "Statistic": "SUM"},
            {"Name": "HANDLE_TIME",              "Unit": "SECONDS", "Statistic": "AVG"},
            {"Name": "QUEUE_ANSWER_TIME",        "Unit": "SECONDS", "Statistic": "AVG"},
            {"Name": "SERVICE_LEVEL",            "Unit": "PERCENT", "Statistic": "AVG",
             "Threshold": {"Comparison": "LT", "ThresholdValue": 20}},
        ]
    )
    
    metrics = {}
    for collection in response["MetricResults"]:
        for item in collection["Collections"]:
            metrics[item["Metric"]["Name"]] = item["Value"]
    
    return {
        "date": date,
        "queue_id": queue_id,
        "calls_handled": int(metrics.get("CONTACTS_HANDLED", 0)),
        "calls_abandoned": int(metrics.get("CONTACTS_ABANDONED", 0)),
        "avg_handle_time_seconds": round(metrics.get("HANDLE_TIME", 0), 1),
        "avg_wait_time_seconds": round(metrics.get("QUEUE_ANSWER_TIME", 0), 1),
        "service_level_pct": round(metrics.get("SERVICE_LEVEL", 0), 1),
    }