Project: End-to-End Contact Center Automation with Amazon Connect

Incoming call
     ↓
Amazon Connect IVR (Contact Flow)
     ↓ (route based on intent)
     ├── Self-service: Lambda resolves & plays response
     └── Agent queue: route to available agent
               ↓
         Live call begins
               ↓
         DeepGram (real-time transcript stream)
               ↓
         Claude scores quality + sentiment
               ↓
         Metrics → DynamoDB → WebSocket → Dashboard

contact-center/
├── infrastructure/
│   ├── main.tf           # Connect instance, queues, routing profiles
│   ├── lambda.tf         # Lambda functions and IAM
│   └── dynamodb.tf       # Tables for calls and agents
├── functions/
│   ├── onboard-client/   # Automate new client provisioning
│   ├── ivr-handler/      # Contact flow Lambda integrations
│   ├── call-events/      # Process Connect event stream
│   ├── transcription/    # DeepGram WebSocket proxy
│   └── quality-scorer/   # Claude-based call scoring
├── dashboard/            # React real-time dashboard
└── flows/                # Amazon Connect contact flow JSON exports

# infrastructure/main.tf

terraform {
  required_providers {
    aws = { source = "hashicorp/aws", version = "~> 5.0" }
  }
}

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

# ── Connect Instance ────────────────────────────────────────────────
resource "aws_connect_instance" "main" {
  instance_alias               = "my-contact-center"
  inbound_calls_enabled        = true
  outbound_calls_enabled       = true
  contact_flow_logs_enabled    = true
  contact_lens_enabled         = true   # AI-powered analytics
  early_media_enabled          = true
  identity_management_type     = "CONNECT_MANAGED"
}

# ── Hours of Operation ──────────────────────────────────────────────
resource "aws_connect_hours_of_operation" "business_hours" {
  instance_id = aws_connect_instance.main.id
  name        = "Business Hours"
  description = "Mon-Fri 9am-6pm EST"
  time_zone   = "America/New_York"

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

# ── Queues ──────────────────────────────────────────────────────────
resource "aws_connect_queue" "support" {
  instance_id           = aws_connect_instance.main.id
  name                  = "Support Queue"
  hours_of_operation_id = aws_connect_hours_of_operation.business_hours.hours_of_operation_id
  max_contacts          = 50
}

resource "aws_connect_queue" "sales" {
  instance_id           = aws_connect_instance.main.id
  name                  = "Sales Queue"
  hours_of_operation_id = aws_connect_hours_of_operation.business_hours.hours_of_operation_id
  max_contacts          = 30
}

# ── Routing Profile ─────────────────────────────────────────────────
resource "aws_connect_routing_profile" "agent" {
  instance_id               = aws_connect_instance.main.id
  name                      = "Standard Agent"
  default_outbound_queue_id = aws_connect_queue.support.queue_id
  description               = "Handles support and sales"

  media_concurrencies {
    channel     = "VOICE"
    concurrency = 1
  }

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

  queue_configs {
    channel  = "VOICE"
    delay    = 0
    priority = 2
    queue_id = aws_connect_queue.sales.queue_id
  }
}

# ── Lambda Integration ──────────────────────────────────────────────
resource "aws_connect_lambda_function_association" "ivr" {
  instance_id   = aws_connect_instance.main.id
  function_arn  = aws_lambda_function.ivr_handler.arn
}

# functions/ivr-handler/handler.py
import json
import boto3

dynamodb = boto3.resource("dynamodb")
clients_table = dynamodb.Table("Clients")

def lambda_handler(event: dict, context) -> dict:
    """
    Called from Amazon Connect contact flow.
    event['Details']['ContactData']['Attributes'] contains flow attributes.
    event['Details']['Parameters'] contains custom params set in the flow.
    """
    contact_id = event["Details"]["ContactData"]["ContactId"]
    phone_number = event["Details"]["ContactData"]["CustomerEndpoint"]["Address"]
    intent = event["Details"]["Parameters"].get("intent", "unknown")

    print(f"Contact {contact_id} | Phone {phone_number} | Intent {intent}")

    if intent == "account_balance":
        result = handle_account_balance(phone_number)
    elif intent == "appointment_booking":
        result = handle_appointment(phone_number, event["Details"]["Parameters"])
    elif intent == "after_hours":
        result = handle_after_hours(phone_number)
    else:
        result = { "action": "transfer_to_agent", "queue": "Support Queue" }

    return result


def handle_account_balance(phone: str) -> dict:
    response = clients_table.query(
        IndexName="PhoneIndex",
        KeyConditionExpression="phone = :p",
        ExpressionAttributeValues={":p": phone}
    )
    if not response["Items"]:
        return { "action": "play_message", "message_key": "account_not_found" }

    client = response["Items"][0]
    return {
        "action": "play_message",
        "message_key": "account_balance",
        "balance": str(client.get("balance_cents", 0)),
        "client_name": client.get("name", "valued customer"),
    }


def handle_appointment(phone: str, params: dict) -> dict:
    # Would integrate with calendar API here
    return {
        "action": "play_message",
        "message_key": "appointment_confirmed",
        "appointment_time": params.get("selected_slot", ""),
    }


def handle_after_hours(phone: str) -> dict:
    return {
        "action": "play_message",
        "message_key": "after_hours_ai_bot",
        "callback_enabled": "true",
    }

# functions/transcription/handler.py
"""
Proxies Amazon Connect audio stream to DeepGram for real-time transcription.
Triggered when a call starts via Connect Event Stream → Kinesis → Lambda.
"""
import asyncio
import json
import os
import boto3
import websockets
from deepgram import DeepgramClient, LiveTranscriptionEvents, LiveOptions

dynamodb = boto3.resource("dynamodb")
transcripts_table = dynamodb.Table("CallTranscripts")
sns = boto3.client("sns")


async def transcribe_call(contact_id: str, audio_stream_url: str):
    dg = DeepgramClient(os.environ["DEEPGRAM_API_KEY"])

    connection = dg.listen.live.v("1")
    transcript_buffer = []

    def on_message(self, result, **kwargs):
        sentence = result.channel.alternatives[0].transcript
        if sentence and result.is_final:
            speaker = "AGENT" if result.metadata.get("channel", 0) == 0 else "CUSTOMER"
            segment = {
                "speaker": speaker,
                "text": sentence,
                "start": result.start,
                "duration": result.duration,
                "confidence": result.channel.alternatives[0].confidence,
            }
            transcript_buffer.append(segment)
            print(f"[{speaker}] {sentence}")

    def on_close(self, close, **kwargs):
        # Save full transcript to DynamoDB
        transcripts_table.put_item(Item={
            "contactId": contact_id,
            "segments": transcript_buffer,
            "fullText": " ".join(s["text"] for s in transcript_buffer),
            "createdAt": __import__("datetime").datetime.utcnow().isoformat(),
        })
        # Trigger quality scoring
        sns.publish(
            TopicArn=os.environ["QUALITY_SCORE_TOPIC"],
            Message=json.dumps({ "contactId": contact_id }),
        )

    connection.on(LiveTranscriptionEvents.Transcript, on_message)
    connection.on(LiveTranscriptionEvents.Close, on_close)

    options = LiveOptions(
        model="nova-2",
        language="en-US",
        smart_format=True,
        punctuate=True,
        diarize=True,          # speaker separation
        multichannel=True,     # separate agent/customer channels
        encoding="mulaw",
        sample_rate=8000,
        channels=2,
    )

    connection.start(options)

    # Pipe Amazon Connect audio stream to DeepGram
    async with websockets.connect(audio_stream_url) as audio_ws:
        async for chunk in audio_ws:
            connection.send(chunk)

    connection.finish()

# functions/quality-scorer/handler.py
import json
import os
import boto3
from anthropic import Anthropic

dynamodb = boto3.resource("dynamodb")
transcripts_table = dynamodb.Table("CallTranscripts")
scores_table = dynamodb.Table("CallScores")
client = Anthropic()


def lambda_handler(event, context):
    for record in event["Records"]:
        message = json.loads(record["Sns"]["Message"])
        contact_id = message["contactId"]
        score_call(contact_id)


def score_call(contact_id: str):
    # Fetch transcript
    resp = transcripts_table.get_item(Key={"contactId": contact_id})
    if "Item" not in resp:
        return

    transcript = resp["Item"]
    full_text = transcript.get("fullText", "")
    segments = transcript.get("segments", [])

    # Build readable transcript with speakers
    formatted = "\n".join(
        f"[{s['speaker']}]: {s['text']}" for s in segments
    )

    prompt = f"""You are a call quality analyst. Evaluate this customer service call transcript and return a JSON score.

Transcript:
{formatted}

Score the call on these dimensions (0-10 each):
1. greeting_quality — Did the agent greet professionally?
2. problem_resolution — Was the customer's issue resolved?
3. empathy — Did the agent show empathy and understanding?
4. communication_clarity — Was the agent clear and easy to understand?
5. call_handling — Did the agent follow proper call handling procedures?
6. customer_sentiment — How did the customer feel? (0=very negative, 10=very positive)

Also provide:
- overall_score (0-100, weighted average)
- summary (2 sentences about the call quality)
- key_issue (the main problem discussed)
- improvement_tips (array of 2-3 specific coaching tips for the agent)
- escalation_risk (true/false — should this be reviewed by a supervisor?)

Return ONLY valid JSON, no other text."""

    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1024,
        messages=[{ "role": "user", "content": prompt }],
    )

    raw = message.content[0].text
    score_data = json.loads(raw)
    score_data["contactId"] = contact_id
    score_data["scoredAt"] = __import__("datetime").datetime.utcnow().isoformat()

    scores_table.put_item(Item=score_data)
    print(f"Scored call {contact_id}: {score_data['overall_score']}/100")

    # Alert supervisor if escalation risk
    if score_data.get("escalation_risk"):
        boto3.client("sns").publish(
            TopicArn=os.environ["ESCALATION_TOPIC"],
            Subject=f"Low Quality Call Alert: {contact_id}",
            Message=json.dumps({
                "contactId": contact_id,
                "score": score_data["overall_score"],
                "summary": score_data["summary"],
            }),
        )

# infrastructure/dynamodb.tf

resource "aws_dynamodb_table" "call_transcripts" {
  name         = "CallTranscripts"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "contactId"

  attribute { name = "contactId"; type = "S" }

  ttl {
    attribute_name = "expiresAt"
    enabled        = true
  }
}

resource "aws_dynamodb_table" "call_scores" {
  name         = "CallScores"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "contactId"

  attribute { name = "contactId"; type = "S" }
  attribute { name = "scoredAt";  type = "S" }
  attribute { name = "agentId";   type = "S" }

  global_secondary_index {
    name               = "AgentScoreIndex"
    hash_key           = "agentId"
    range_key          = "scoredAt"
    projection_type    = "ALL"
  }
}

resource "aws_dynamodb_table" "agents" {
  name         = "Agents"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "agentId"

  attribute { name = "agentId"; type = "S" }

  stream_enabled   = true
  stream_view_type = "NEW_AND_OLD_IMAGES"
}

// dashboard/src/App.tsx
"use client";
import { useState, useEffect, useCallback } from "react";
import { Phone, Star, TrendingUp, AlertTriangle, Users } from "lucide-react";

interface AgentMetric {
  agentId: string;
  name: string;
  callsToday: number;
  avgScore: number;
  activeCalls: number;
  escalations: number;
}

interface LiveCall {
  contactId: string;
  agentName: string;
  duration: number;   // seconds
  sentiment: "positive" | "neutral" | "negative";
  transcript: string; // latest segment
}

export default function Dashboard() {
  const [agents, setAgents]     = useState<AgentMetric[]>([]);
  const [liveCalls, setLiveCalls] = useState<LiveCall[]>([]);
  const [wsStatus, setWsStatus] = useState<"connecting" | "connected" | "error">("connecting");

  const connect = useCallback(() => {
    const ws = new WebSocket(process.env.NEXT_PUBLIC_DASHBOARD_WS_URL!);

    ws.onopen  = () => setWsStatus("connected");
    ws.onerror = () => setWsStatus("error");
    ws.onclose = () => { setWsStatus("connecting"); setTimeout(connect, 3000); };

    ws.onmessage = (e) => {
      const { type, data } = JSON.parse(e.data);
      if (type === "agent_metrics")  setAgents(data);
      if (type === "live_calls")     setLiveCalls(data);
      if (type === "call_scored") {
        setAgents(prev => prev.map(a =>
          a.agentId === data.agentId
            ? { ...a, avgScore: data.newAvgScore, escalations: a.escalations + (data.escalation ? 1 : 0) }
            : a
        ));
      }
    };
    return ws;
  }, []);

  useEffect(() => {
    const ws = connect();
    return () => ws.close();
  }, [connect]);

  const sentimentColor = (s: string) =>
    s === "positive" ? "text-emerald-500" : s === "negative" ? "text-red-400" : "text-amber-400";

  return (
    <div className="min-h-screen bg-background p-6">
      <div className="max-w-7xl mx-auto">

        {/* Header */}
        <div className="flex items-center justify-between mb-8">
          <div>
            <h1 className="text-2xl font-extrabold text-foreground">Contact Center Live</h1>
            <p className="text-sm text-muted-foreground mt-1">Real-time agent performance dashboard</p>
          </div>
          <span className={`flex items-center gap-2 text-xs font-semibold px-3 py-1.5 rounded-full
            ${wsStatus === "connected" ? "bg-emerald-500/10 text-emerald-500" : "bg-amber-500/10 text-amber-500"}`}>
            <span className={`w-2 h-2 rounded-full ${wsStatus === "connected" ? "bg-emerald-500 animate-pulse" : "bg-amber-500"}`} />
            {wsStatus === "connected" ? "Live" : "Reconnecting…"}
          </span>
        </div>

        {/* Stats */}
        <div className="grid grid-cols-2 lg:grid-cols-4 gap-4 mb-8">
          {[
            { icon: Phone,         label: "Active Calls",    value: liveCalls.length,                        color: "text-blue-500" },
            { icon: Users,         label: "Agents Online",   value: agents.filter(a => a.activeCalls > 0).length, color: "text-violet-500" },
            { icon: Star,          label: "Avg Score Today", value: agents.length
                ? Math.round(agents.reduce((s, a) => s + a.avgScore, 0) / agents.length) + "/100"
                : "—",                                                                                        color: "text-amber-500" },
            { icon: AlertTriangle, label: "Escalations",     value: agents.reduce((s, a) => s + a.escalations, 0), color: "text-red-400" },
          ].map((stat) => (
            <div key={stat.label} className="rounded-2xl border border-border bg-card p-5 text-center">
              <stat.icon className={`w-5 h-5 mx-auto mb-2 ${stat.color}`} />
              <p className="text-2xl font-extrabold text-foreground">{stat.value}</p>
              <p className="text-xs text-muted-foreground mt-0.5">{stat.label}</p>
            </div>
          ))}
        </div>

        <div className="grid lg:grid-cols-2 gap-6">

          {/* Agent scores */}
          <div className="rounded-2xl border border-border bg-card overflow-hidden">
            <div className="px-6 py-4 border-b border-border flex items-center gap-2">
              <TrendingUp className="w-4 h-4 text-primary" />
              <h2 className="font-bold text-foreground text-sm">Agent Performance</h2>
            </div>
            <div className="divide-y divide-border">
              {agents
                .sort((a, b) => b.avgScore - a.avgScore)
                .map((agent) => (
                <div key={agent.agentId} className="flex items-center gap-4 px-6 py-3">
                  <div className="flex-1">
                    <p className="font-semibold text-foreground text-sm">{agent.name}</p>
                    <p className="text-xs text-muted-foreground">{agent.callsToday} calls today</p>
                  </div>
                  {agent.activeCalls > 0 && (
                    <span className="text-[10px] font-bold bg-emerald-500/10 text-emerald-500 px-2 py-0.5 rounded-full">
                      On call
                    </span>
                  )}
                  {agent.escalations > 0 && (
                    <AlertTriangle className="w-3.5 h-3.5 text-red-400" />
                  )}
                  <div className="text-right shrink-0">
                    <p className={`font-extrabold text-sm ${
                      agent.avgScore >= 80 ? "text-emerald-500"
                      : agent.avgScore >= 60 ? "text-amber-500"
                      : "text-red-400"
                    }`}>
                      {agent.avgScore}/100
                    </p>
                    <p className="text-xs text-muted-foreground">avg score</p>
                  </div>
                </div>
              ))}
            </div>
          </div>

          {/* Live calls */}
          <div className="rounded-2xl border border-border bg-card overflow-hidden">
            <div className="px-6 py-4 border-b border-border flex items-center gap-2">
              <Phone className="w-4 h-4 text-emerald-500" />
              <h2 className="font-bold text-foreground text-sm">Live Calls</h2>
            </div>
            {liveCalls.length === 0 ? (
              <div className="px-6 py-8 text-center text-sm text-muted-foreground">
                No active calls right now
              </div>
            ) : (
              <div className="divide-y divide-border">
                {liveCalls.map((call) => (
                  <div key={call.contactId} className="px-6 py-3">
                    <div className="flex items-center justify-between mb-1">
                      <span className="font-semibold text-foreground text-sm">{call.agentName}</span>
                      <span className={`text-xs font-bold ${sentimentColor(call.sentiment)}`}>
                        {call.sentiment}
                      </span>
                    </div>
                    <p className="text-xs text-muted-foreground line-clamp-1 italic">
                      &ldquo;{call.transcript}&rdquo;
                    </p>
                    <p className="text-[10px] text-muted-foreground mt-1">
                      {Math.floor(call.duration / 60)}:{String(call.duration % 60).padStart(2, "0")} elapsed
                    </p>
                  </div>
                ))}
              </div>
            )}
          </div>

        </div>
      </div>
    </div>
  );
}

# functions/onboard-client/handler.py
"""
Automates provisioning a new client's contact center configuration:
- Create dedicated queue
- Assign phone number
- Configure routing
- Send welcome notification
"""
import boto3
import os

connect = boto3.client("connect")
sns = boto3.client("sns")
INSTANCE_ID = os.environ["CONNECT_INSTANCE_ID"]


def lambda_handler(event, context):
    body = event["body"] if isinstance(event.get("body"), dict) else __import__("json").loads(event.get("body", "{}"))

    client_name = body["clientName"]
    client_id   = body["clientId"]
    hours_id    = body.get("hoursOfOperationId", os.environ["DEFAULT_HOURS_ID"])

    # 1. Create dedicated queue
    queue = connect.create_queue(
        InstanceId=INSTANCE_ID,
        Name=f"{client_name} Queue",
        Description=f"Queue for client {client_id}",
        HoursOfOperationId=hours_id,
        MaxContacts=20,
        Tags={"clientId": client_id, "env": "production"},
    )
    queue_id = queue["Queue"]["QueueId"]

    # 2. Claim a phone number from inventory
    phone_number = connect.list_phone_numbers(
        InstanceId=INSTANCE_ID,
        PhoneNumberTypes=["DID"],
        PhoneNumberCountryCodes=["US"],
    )["PhoneNumberSummaryList"][0]

    connect.associate_phone_number_contact_flow(
        InstanceId=INSTANCE_ID,
        PhoneNumberId=phone_number["Id"],
        ContactFlowId=os.environ["MAIN_CONTACT_FLOW_ID"],
    )

    # 3. Tag phone number with client
    connect.tag_resource(
        resourceArn=phone_number["PhoneNumberArn"],
        tags={"clientId": client_id},
    )

    # 4. Notify via SNS
    sns.publish(
        TopicArn=os.environ["ONBOARDING_TOPIC"],
        Subject=f"Contact Center Ready: {client_name}",
        Message=__import__("json").dumps({
            "clientId":    client_id,
            "clientName":  client_name,
            "queueId":     queue_id,
            "phoneNumber": phone_number["PhoneNumber"],
        }),
    )

    return {
        "statusCode": 200,
        "body": __import__("json").dumps({
            "message":     "Contact center provisioned",
            "queueId":     queue_id,
            "phoneNumber": phone_number["PhoneNumber"],
        }),
    }

# Deploy infrastructure
cd infrastructure
terraform init
terraform plan -out=tfplan
terraform apply tfplan

# Deploy Lambda functions
cd ../functions/ivr-handler
pip install -r requirements.txt -t ./package
cd package && zip -r ../function.zip . && cd ..
zip -g function.zip handler.py
aws lambda update-function-code \
  --function-name ivr-handler \
  --zip-file fileb://function.zip

# Deploy all functions with a script
for fn in ivr-handler call-events transcription quality-scorer onboard-client; do
  cd ../functions/$fn
  pip install -r requirements.txt -t ./package 2>/dev/null
  cd package && zip -r ../function.zip . && cd ..
  zip -g function.zip handler.py
  aws lambda update-function-code \
    --function-name $fn \
    --zip-file fileb://function.zip
done