Single AI agents are powerful. Multi-agent systems are transformative. This guide builds a complete, production-ready multi-agent AI system on Amazon EKS, the exact architecture used by financial services, healthcare, and enterprise SaaS teams in 2026. Five specialised agents (Customer, Payment, Fraud, Compliance, Customer Support), a Custom Agent Router, Bedrock Guardrails, RAG with OpenSearch, response verification, WAF protection, and human approval gates. Every component configured from scratch. Every line of code included.

What You Are Building

Look at the architecture diagram. Every component in it has a job, and they all work together to serve a user request safely, accurately, and at enterprise scale.

Here is the full data flow, before a single line of code:

User → App (mobile/web) → Route 53 (DNS lookup) → CloudFront (CDN, content closer to user) → WAF (blocks malicious traffic) → Cognito (authenticates user, issues token) → API Gateway (receives task + login token) → Private VPC Link (secure internal routing) → ALB (distributes to healthy agent pods) → EKS (runs the containerised agent application) → Custom Agent Router (chooses the right agent) → Specialist Agent (Customer / Payment / Fraud / Compliance / Support) → Company Knowledge (S3 → OpenSearch → Bedrock Knowledge Bases) → Model & Safety (Bedrock Guardrails → Claude via Bedrock) → Business Tools (AgentCore Gateway → Business APIs) → Human Approval (for sensitive actions — Set Functions → Human) → Response Verification (custom code on EKS checks the response) → If rejected: return to agent for correction → If accepted: send to user → User sees the response

That is the system. This guide builds every layer of it, step by step.

What you need:

  • AWS account with admin access
  • AWS CLI configured (aws configure)
  • kubectl installed
  • Python 3.11+
  • Docker
  • About 3–4 hours

AWS services used: Route 53, CloudFront, WAF, S3, Cognito, API Gateway, VPC, ALB, EKS, Amazon Bedrock, Bedrock Knowledge Bases, OpenSearch Serverless, Bedrock Guardrails, AgentCore Gateway, Lambda, Step Functions, CloudWatch, IAM

Step 1: Foundation VPC, EKS Cluster, and IAM

Everything runs inside a VPC. Start here.

1.1 Create the VPC

bash

REGION="eu-west-1" ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text) CLUSTER_NAME="multi-agent-prod" # Create VPC with public and private subnets aws ec2 create-vpc \ --cidr-block 10.0.0.0/16 \ --tag-specifications 'ResourceType=vpc,Tags=[{Key=Name,Value=multi-agent-vpc},{Key=Project,Value=multi-agent}]' \ --region $REGION VPC_ID=$(aws ec2 describe-vpcs \ --filters "Name=tag:Name,Values=multi-agent-vpc" \ --query 'Vpcs[0].VpcId' \ --output text \ --region $REGION) echo "VPC ID: $VPC_ID" # Enable DNS hostnames (required for EKS) aws ec2 modify-vpc-attribute \ --vpc-id $VPC_ID \ --enable-dns-hostnames \ --region $REGION # Create Internet Gateway for public subnets IGW_ID=$(aws ec2 create-internet-gateway \ --query 'InternetGateway.InternetGatewayId' \ --output text \ --region $REGION) aws ec2 attach-internet-gateway \ --internet-gateway-id $IGW_ID \ --vpc-id $VPC_ID \ --region $REGION # Create public subnets (for ALB and NAT Gateway) PUBLIC_SUBNET_1=$(aws ec2 create-subnet \ --vpc-id $VPC_ID \ --cidr-block 10.0.1.0/24 \ --availability-zone ${REGION}a \ --tag-specifications 'ResourceType=subnet,Tags=[{Key=Name,Value=public-1a},{Key=kubernetes.io/role/elb,Value=1}]' \ --query 'Subnet.SubnetId' \ --output text \ --region $REGION) PUBLIC_SUBNET_2=$(aws ec2 create-subnet \ --vpc-id $VPC_ID \ --cidr-block 10.0.2.0/24 \ --availability-zone ${REGION}b \ --tag-specifications 'ResourceType=subnet,Tags=[{Key=Name,Value=public-1b},{Key=kubernetes.io/role/elb,Value=1}]' \ --query 'Subnet.SubnetId' \ --output text \ --region $REGION) # Create private subnets (for EKS nodes — never directly exposed) PRIVATE_SUBNET_1=$(aws ec2 create-subnet \ --vpc-id $VPC_ID \ --cidr-block 10.0.10.0/24 \ --availability-zone ${REGION}a \ --tag-specifications 'ResourceType=subnet,Tags=[{Key=Name,Value=private-1a},{Key=kubernetes.io/role/internal-elb,Value=1}]' \ --query 'Subnet.SubnetId' \ --output text \ --region $REGION) PRIVATE_SUBNET_2=$(aws ec2 create-subnet \ --vpc-id $VPC_ID \ --cidr-block 10.0.11.0/24 \ --availability-zone ${REGION}b \ --tag-specifications 'ResourceType=subnet,Tags=[{Key=Name,Value=private-1b},{Key=kubernetes.io/role/internal-elb,Value=1}]' \ --query 'Subnet.SubnetId' \ --output text \ --region $REGION) echo "Subnets created: $PUBLIC_SUBNET_1, $PUBLIC_SUBNET_2, $PRIVATE_SUBNET_1, $PRIVATE_SUBNET_2"

1.2 Create the EKS Cluster

bash

# Create the EKS cluster using eksctl (simplest approach) # Install eksctl if not present: # brew install eksctl (macOS) # or download from https://eksctl.io cat > cluster-config.yaml << EOF apiVersion: eksctl.io/v1alpha5 kind: ClusterConfig metadata: name: $CLUSTER_NAME region: $REGION version: "1.31" tags: Project: multi-agent Environment: production vpc: id: $VPC_ID subnets: private: private-1a: id: $PRIVATE_SUBNET_1 private-1b: id: $PRIVATE_SUBNET_2 public: public-1a: id: $PUBLIC_SUBNET_1 public-1b: id: $PUBLIC_SUBNET_2 managedNodeGroups: - name: agent-nodes instanceType: m5.xlarge # 4 vCPU, 16GB — good for agent workloads minSize: 2 maxSize: 10 desiredCapacity: 3 privateNetworking: true # Nodes in private subnets only labels: role: agent tags: k8s.io/cluster-autoscaler/enabled: "true" k8s.io/cluster-autoscaler/$CLUSTER_NAME: "owned" iam: attachPolicyARNs: - arn:aws:iam::aws:policy/AmazonEKSWorkerNodePolicy - arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy - arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly withAddonPolicies: autoScaler: true cloudWatch: true addons: - name: vpc-cni - name: coredns - name: kube-proxy - name: aws-ebs-csi-driver EOF eksctl create cluster -f cluster-config.yaml # This takes 15-20 minutes # Get credentials when done: aws eks update-kubeconfig --region $REGION --name $CLUSTER_NAME # Verify cluster is running kubectl get nodes

1.3 Create Core IAM Roles

bash

# IAM role for the agent pods — allows Bedrock, S3, DynamoDB access cat > agent-pod-trust.json << EOF { "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": { "Federated": "arn:aws:iam::$ACCOUNT_ID:oidc-provider/$(aws eks describe-cluster \ --name $CLUSTER_NAME \ --query 'cluster.identity.oidc.issuer' \ --output text \ --region $REGION | sed 's/https:\/\///')" }, "Action": "sts:AssumeRoleWithWebIdentity", "Condition": { "StringEquals": { "$(aws eks describe-cluster \ --name $CLUSTER_NAME \ --query 'cluster.identity.oidc.issuer' \ --output text \ --region $REGION | sed 's/https:\/\///')": "system:serviceaccount:multi-agent:agent-service-account" } } }] } EOF aws iam create-role \ --role-name multi-agent-pod-role \ --assume-role-policy-document file://agent-pod-trust.json cat > agent-pod-policy.json << EOF { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream", "bedrock:ApplyGuardrail", "bedrock:Retrieve", "bedrock:RetrieveAndGenerate" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::multi-agent-company-docs", "arn:aws:s3:::multi-agent-company-docs/*" ] }, { "Effect": "Allow", "Action": [ "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:Query" ], "Resource": "arn:aws:dynamodb:$REGION:$ACCOUNT_ID:table/agent-sessions" }, { "Effect": "Allow", "Action": [ "cloudwatch:PutMetricData", "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "sns:Publish" ], "Resource": "arn:aws:sns:$REGION:$ACCOUNT_ID:human-approval-topic" } ] } EOF aws iam put-role-policy \ --role-name multi-agent-pod-role \ --policy-name agent-pod-permissions \ --policy-document file://agent-pod-policy.json echo "IAM role created: arn:aws:iam::$ACCOUNT_ID:role/multi-agent-pod-role"

Step 2: Company Knowledge S3, OpenSearch, and Bedrock Knowledge Base

The bottom-left of your architecture diagram. Every agent reads from here when it needs company information.

2.1 Create the S3 Document Store

bash

# S3 bucket for company documents aws s3 mb s3://multi-agent-company-docs-$ACCOUNT_ID --region $REGION # Upload sample company documents mkdir -p company-docs/{policies,products,procedures,compliance} cat > company-docs/policies/refund-policy.txt << 'EOF' REFUND POLICY — Last updated July 2026 Standard refunds: Customers may request a full refund within 30 days of purchase. Duplicate charges: Any duplicate charge is eligible for immediate refund regardless of date. Subscription cancellations: Pro-rated refunds available for annual subscriptions cancelled within 90 days. Fraud cases: Fraudulent charges are escalated to the fraud team and resolved within 24 hours. Approval required: Refunds above £500 require manager approval. EOF cat > company-docs/procedures/account-verification.txt << 'EOF' ACCOUNT VERIFICATION PROCEDURE KYC Requirements: - Identity verification required for accounts over £1,000 monthly limit - Photo ID + proof of address for business accounts - Enhanced due diligence for high-risk jurisdictions AML Checks: - All transactions over £10,000 flagged for review - Sanctions screening on all new accounts - PEP (Politically Exposed Person) screening monthly EOF cat > company-docs/products/subscription-tiers.txt << 'EOF' SUBSCRIPTION TIERS — 2026 Free Plan: Up to 100 API calls/month, community support Starter: £29/month, 10,000 API calls, email support Professional: £99/month, 100,000 API calls, priority support, SLA 99.9% Enterprise: Custom pricing, unlimited calls, dedicated support, SLA 99.99% Upgrades: Take effect immediately, charged pro-rated Downgrades: Take effect at next billing cycle EOF # Upload all documents to S3 aws s3 sync company-docs/ s3://multi-agent-company-docs-$ACCOUNT_ID/ \ --region $REGION echo "Company documents uploaded to S3"

2.2 Create OpenSearch Serverless (Vector Store)

bash

# Create OpenSearch Serverless collection for vector search # This is what "OpenSearch (Serverless Vector Store)" in your diagram refers to cat > opensearch-policy.json << EOF [{ "Rules": [ { "Resource": ["collection/multi-agent-knowledge"], "Permission": [ "aoss:CreateCollectionItems", "aoss:DeleteCollectionItems", "aoss:UpdateCollectionItems", "aoss:DescribeCollectionItems" ], "ResourceType": "collection" }, { "Resource": ["index/multi-agent-knowledge/*"], "Permission": [ "aoss:CreateIndex", "aoss:DeleteIndex", "aoss:UpdateIndex", "aoss:DescribeIndex", "aoss:ReadDocument", "aoss:WriteDocument" ], "ResourceType": "index" } ], "Principal": [ "arn:aws:iam::$ACCOUNT_ID:role/multi-agent-pod-role", "arn:aws:iam::$ACCOUNT_ID:role/AmazonBedrockExecutionRoleForKnowledgeBase" ] }] EOF aws opensearchserverless create-access-policy \ --name multi-agent-data-policy \ --type data \ --policy "$(cat opensearch-policy.json | jq -c .)" \ --region $REGION aws opensearchserverless create-collection \ --name multi-agent-knowledge \ --type VECTORSEARCH \ --description "Company knowledge vector store for multi-agent system" \ --region $REGION # Wait for collection to be active echo "Waiting for OpenSearch collection to become active..." aws opensearchserverless wait collection-active \ --id $(aws opensearchserverless list-collections \ --collection-filters name=multi-agent-knowledge \ --query 'collectionSummaries[0].id' \ --output text \ --region $REGION) \ --region $REGION || sleep 120 OPENSEARCH_ENDPOINT=$(aws opensearchserverless list-collections \ --collection-filters name=multi-agent-knowledge \ --query 'collectionSummaries[0].collectionEndpoint' \ --output text \ --region $REGION) echo "OpenSearch endpoint: $OPENSEARCH_ENDPOINT"

2.3 Create Bedrock Knowledge Base

bash

# Create IAM role for Bedrock Knowledge Base cat > bedrock-kb-trust.json << EOF { "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": {"Service": "bedrock.amazonaws.com"}, "Action": "sts:AssumeRole", "Condition": { "StringEquals": {"aws:SourceAccount": "$ACCOUNT_ID"} } }] } EOF aws iam create-role \ --role-name AmazonBedrockExecutionRoleForKnowledgeBase \ --assume-role-policy-document file://bedrock-kb-trust.json aws iam attach-role-policy \ --role-name AmazonBedrockExecutionRoleForKnowledgeBase \ --policy-arn arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess # Create the Knowledge Base KB_ID=$(aws bedrock-agent create-knowledge-base \ --name "multi-agent-company-knowledge" \ --description "Company knowledge base for multi-agent system" \ --role-arn arn:aws:iam::$ACCOUNT_ID:role/AmazonBedrockExecutionRoleForKnowledgeBase \ --knowledge-base-configuration '{ "type": "VECTOR", "vectorKnowledgeBaseConfiguration": { "embeddingModelArn": "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0" } }' \ --storage-configuration "{ \"type\": \"OPENSEARCH_SERVERLESS\", \"opensearchServerlessConfiguration\": { \"collectionArn\": \"arn:aws:aoss:$REGION:$ACCOUNT_ID:collection/multi-agent-knowledge\", \"vectorIndexName\": \"company-knowledge-index\", \"fieldMapping\": { \"vectorField\": \"embedding\", \"textField\": \"text\", \"metadataField\": \"metadata\" } } }" \ --region $REGION \ --query 'knowledgeBase.knowledgeBaseId' \ --output text) echo "Knowledge Base ID: $KB_ID" # Add S3 data source to the Knowledge Base aws bedrock-agent create-data-source \ --knowledge-base-id $KB_ID \ --name "company-documents-s3" \ --data-source-configuration "{ \"type\": \"S3\", \"s3Configuration\": { \"bucketArn\": \"arn:aws:s3:::multi-agent-company-docs-$ACCOUNT_ID\" } }" \ --region $REGION # Trigger initial ingestion aws bedrock-agent start-ingestion-job \ --knowledge-base-id $KB_ID \ --data-source-id $(aws bedrock-agent list-data-sources \ --knowledge-base-id $KB_ID \ --query 'dataSourceSummaries[0].dataSourceId' \ --output text \ --region $REGION) \ --region $REGION echo "Knowledge base ingestion started"

Step 3: Model and Safety Bedrock Guardrails

The centre-bottom of your diagram. This is what stands between the model and potentially harmful outputs.

bash

# Create Bedrock Guardrails — the safety layer for all agents GUARDRAIL_ID=$(aws bedrock create-guardrail \ --name "multi-agent-safety-guardrail" \ --description "Safety guardrails for all agents in the multi-agent system" \ --topic-policy-config '{ "topicsConfig": [ { "name": "Financial advice without authorisation", "definition": "Providing specific investment advice, portfolio recommendations, or financial planning guidance without appropriate authorisation", "examples": [ "You should invest all your savings in...", "I recommend you buy..." ], "type": "DENY" }, { "name": "Medical advice", "definition": "Providing medical diagnoses, treatment recommendations, or health guidance", "type": "DENY" }, { "name": "Legal advice", "definition": "Providing specific legal advice or legal interpretations", "type": "DENY" } ] }' \ --content-policy-config '{ "filtersConfig": [ {"type": "SEXUAL", "inputStrength": "HIGH", "outputStrength": "HIGH"}, {"type": "VIOLENCE", "inputStrength": "MEDIUM", "outputStrength": "HIGH"}, {"type": "HATE", "inputStrength": "HIGH", "outputStrength": "HIGH"}, {"type": "INSULTS", "inputStrength": "MEDIUM", "outputStrength": "MEDIUM"}, {"type": "MISCONDUCT", "inputStrength": "HIGH", "outputStrength": "HIGH"}, {"type": "PROMPT_ATTACK", "inputStrength": "HIGH", "outputStrength": "NONE"} ] }' \ --word-policy-config '{ "wordsConfig": [ {"text": "competitor_name_1"}, {"text": "confidential_product_code"} ], "managedWordListsConfig": [ {"type": "PROFANITY"} ] }' \ --sensitive-information-policy-config '{ "piiEntitiesConfig": [ {"type": "CREDIT_DEBIT_CARD_NUMBER", "action": "BLOCK"}, {"type": "CREDIT_DEBIT_CVV", "action": "BLOCK"}, {"type": "US_SOCIAL_SECURITY_NUMBER", "action": "ANONYMIZE"}, {"type": "UK_NATIONAL_INSURANCE_NUMBER", "action": "ANONYMIZE"}, {"type": "EMAIL", "action": "ANONYMIZE"}, {"type": "PHONE", "action": "ANONYMIZE"} ] }' \ --blocked-input-messaging "I cannot process this request as it contains content that violates our usage policy." \ --blocked-outputs-messaging "I cannot provide this response as it contains content that violates our policy. Please contact support for assistance." \ --region $REGION \ --query 'guardrailId' \ --output text) # Create guardrail version GUARDRAIL_VERSION=$(aws bedrock create-guardrail-version \ --guardrail-id $GUARDRAIL_ID \ --region $REGION \ --query 'version' \ --output text) echo "Guardrail ID: $GUARDRAIL_ID, Version: $GUARDRAIL_VERSION"

Step 4: The Five Specialist Agents Core Python Code

Now for the heart of the system. Each agent in your diagram does one specific job. Here is the complete Python implementation.

python

# agents/base_agent.py # Base class that all five agents inherit from import boto3 import json import logging import time import os from datetime import datetime, timezone from typing import Optional from dataclasses import dataclass logger = logging.getLogger(__name__) REGION = os.environ.get("AWS_REGION", "eu-west-1") MODEL_ID = "anthropic.claude-opus-4-7-20250514" GUARDRAIL_ID = os.environ.get("GUARDRAIL_ID") GUARDRAIL_VERSION = os.environ.get("GUARDRAIL_VERSION", "1") KNOWLEDGE_BASE_ID = os.environ.get("KNOWLEDGE_BASE_ID") bedrock = boto3.client('bedrock-runtime', region_name='us-east-1') bedrock_agent = boto3.client('bedrock-agent-runtime', region_name='us-east-1') cloudwatch = boto3.client('cloudwatch', region_name=REGION) @dataclass class AgentResponse: """Standardised response from any agent""" agent_type: str success: bool content: str requires_human_approval: bool = False approval_reason: str = "" knowledge_sources: list = None tokens_used: int = 0 latency_ms: int = 0 session_id: str = "" class BaseAgent: """ Base class for all five specialist agents. Each specialist agent: 1. Has a specific system prompt defining its role and limits 2. Retrieves company knowledge from the shared Knowledge Base 3. Calls Claude via Bedrock with Guardrails applied 4. Identifies whether human approval is required 5. Returns a structured AgentResponse """ agent_type: str = "base" system_prompt: str = "" high_value_threshold: float = 500.0 def __init__(self): self.bedrock = bedrock self.bedrock_agent = bedrock_agent self.cloudwatch = cloudwatch def retrieve_company_knowledge(self, query: str, num_results: int = 5) -> tuple[str, list]: """ Retrieve relevant company knowledge from Bedrock Knowledge Base. This is the RAG step — find relevant documents before calling the model. """ try: response = self.bedrock_agent.retrieve( knowledgeBaseId=KNOWLEDGE_BASE_ID, retrievalQuery={"text": query}, retrievalConfiguration={ "vectorSearchConfiguration": { "numberOfResults": num_results, "overrideSearchType": "HYBRID" } } ) results = response.get("retrievalResults", []) if not results: return "", [] # Format retrieved context context_parts = [] sources = [] for i, result in enumerate(results, 1): content = result["content"]["text"] source = result.get("location", {}).get("s3Location", {}).get("uri", "company-docs") score = result.get("score", 0) context_parts.append(f"[Source {i}: {source} (relevance: {score:.2f})]\n{content}") sources.append({"source": source, "score": score}) return "\n\n".join(context_parts), sources except Exception as e: logger.error(f"Knowledge retrieval failed: {e}") return "", [] def call_bedrock_with_guardrails( self, messages: list, additional_context: str = "", max_tokens: int = 1000 ) -> tuple[str, int]: """ Call Claude via Bedrock with Guardrails applied. Guardrails check both the input and the output. """ full_system = self.system_prompt if additional_context: full_system += f"\n\n## Relevant Company Knowledge:\n{additional_context}" body = { "anthropic_version": "bedrock-2023-05-31", "max_tokens": max_tokens, "system": full_system, "messages": messages } # Apply guardrails to the request if GUARDRAIL_ID: body["amazon-bedrock-guardrailConfig"] = { "guardrailIdentifier": GUARDRAIL_ID, "guardrailVersion": GUARDRAIL_VERSION, "trace": "enabled" } start_time = time.time() response = self.bedrock.invoke_model( modelId=MODEL_ID, body=json.dumps(body) ) latency_ms = int((time.time() - start_time) * 1000) result = json.loads(response['body'].read()) # Check if guardrails blocked the response guardrail_action = result.get("amazon-bedrock-guardrailAction", "NONE") if guardrail_action == "GUARDRAIL_INTERVENED": logger.warning(f"Guardrail intervened for {self.agent_type}") content = result.get("content", [{}])[0].get("text", "I cannot provide that response due to our content policy.") else: content = result["content"][0]["text"] total_tokens = result["usage"]["input_tokens"] + result["usage"]["output_tokens"] return content, total_tokens def check_requires_human_approval(self, request: str, response: str) -> tuple[bool, str]: """ Determine if this response requires human approval before being sent. Each specialist agent overrides this with its own rules. """ return False, "" def handle(self, request: str, session_id: str, user_id: str) -> AgentResponse: """Main handler — called by the Agent Router""" start_time = time.time() logger.info(f"{self.agent_type} handling request for session {session_id}") # Step 1: Retrieve relevant company knowledge (RAG) knowledge_context, sources = self.retrieve_company_knowledge(request) # Step 2: Build conversation messages messages = [{"role": "user", "content": f"User ID: {user_id}\n\nRequest: {request}"}] # Step 3: Call the model with guardrails content, tokens_used = self.call_bedrock_with_guardrails( messages=messages, additional_context=knowledge_context ) # Step 4: Check human approval requirement requires_approval, approval_reason = self.check_requires_human_approval(request, content) latency_ms = int((time.time() - start_time) * 1000) # Step 5: Publish metrics self.publish_metrics(tokens_used, latency_ms, requires_approval) return AgentResponse( agent_type=self.agent_type, success=True, content=content, requires_human_approval=requires_approval, approval_reason=approval_reason, knowledge_sources=sources, tokens_used=tokens_used, latency_ms=latency_ms, session_id=session_id ) def publish_metrics(self, tokens: int, latency: int, requires_approval: bool): """Publish CloudWatch metrics for cost and performance monitoring""" try: self.cloudwatch.put_metric_data( Namespace="MultiAgentSystem/Agents", MetricData=[ { "MetricName": "TokensConsumed", "Value": tokens, "Unit": "Count", "Dimensions": [{"Name": "AgentType", "Value": self.agent_type}] }, { "MetricName": "RequestLatencyMs", "Value": latency, "Unit": "Milliseconds", "Dimensions": [{"Name": "AgentType", "Value": self.agent_type}] }, { "MetricName": "HumanApprovalRequired", "Value": 1 if requires_approval else 0, "Unit": "Count", "Dimensions": [{"Name": "AgentType", "Value": self.agent_type}] } ] ) except Exception as e: logger.warning(f"Metrics publish failed: {e}")

python

# agents/specialist_agents.py # The five specialist agents from your architecture diagram from .base_agent import BaseAgent import re class CustomerAgent(BaseAgent): """ Handles customer account requests: - Account information queries - Profile updates - Subscription changes - Billing history """ agent_type = "customer" system_prompt = """You are the Customer Account Agent for our platform. Your responsibilities: - Answer questions about the customer's account, subscription, and billing history - Help customers update their profile information - Explain subscription tiers and help customers choose the right plan - Process subscription upgrades and downgrades per company policy Always: - Verify you are speaking about the correct account (reference User ID) - Be accurate about subscription terms and pricing from company documentation - Never process payments directly — route to Payment Agent - Never access other customers' information You have access to company knowledge about subscription tiers, policies, and procedures.""" def check_requires_human_approval(self, request: str, response: str) -> tuple[bool, str]: """Account changes above certain thresholds need approval""" # Account deletion always needs human confirmation if any(term in request.lower() for term in ["delete account", "close account", "cancel everything"]): return True, "Account deletion requires human confirmation and data retention review" return False, "" class PaymentAgent(BaseAgent): """ Handles approved payment tasks: - Refund processing - Payment dispute investigation - Invoice generation - Subscription billing issues """ agent_type = "payment" system_prompt = """You are the Payment Agent for our platform. Your responsibilities: - Process refund requests according to company refund policy - Investigate payment disputes - Clarify billing questions and invoice details - Process approved payment tasks Rules you MUST follow: - Refunds up to £500: you can approve and process directly - Refunds above £500: flag for human manager approval with full justification - Duplicate charges: always refund immediately, no approval needed - Suspected fraud: immediately escalate to Fraud Agent — do NOT process payment - Always reference the specific transaction ID in your response - Log all payment actions with timestamp and reason Company refund policy is available in your knowledge base.""" def check_requires_human_approval(self, request: str, response: str) -> tuple[bool, str]: """Large refunds and unusual payment patterns need human review""" # Extract amounts from request and response amounts = re.findall(r'£(\d+(?:,\d{3})*(?:\.\d{2})?)', request + response) for amount_str in amounts: amount = float(amount_str.replace(',', '')) if amount > self.high_value_threshold: return True, f"Refund of £{amount:,.2f} exceeds £{self.high_value_threshold:.0f} threshold — manager approval required" return False, "" class FraudAgent(BaseAgent): """ Investigates suspicious activity: - Unusual transaction patterns - Suspicious login attempts - KYC/AML checks - Sanctions screening """ agent_type = "fraud" system_prompt = """You are the Fraud Investigation Agent for our platform. Your responsibilities: - Investigate reports of suspicious account activity - Identify patterns consistent with fraud, money laundering, or identity theft - Check accounts against KYC (Know Your Customer) requirements - Flag potential AML (Anti-Money Laundering) concerns - Perform sanctions and PEP (Politically Exposed Person) screening Investigation framework: 1. Review the transaction pattern or suspicious activity reported 2. Check against company AML and KYC policies (in knowledge base) 3. Assign a risk level: LOW / MEDIUM / HIGH / CRITICAL 4. For MEDIUM and above: recommend specific actions 5. For HIGH and CRITICAL: ALWAYS escalate for human review You do NOT have authority to freeze accounts or block transactions. You gather evidence and make recommendations. Human team executes. Your knowledge base contains AML procedures, KYC requirements, and compliance documentation.""" def check_requires_human_approval(self, request: str, response: str) -> tuple[bool, str]: """Fraud findings always need human review before action""" risk_indicators = ["HIGH", "CRITICAL", "suspicious", "fraud", "money laundering", "sanctions", "freeze", "block", "escalate"] response_lower = response.lower() for indicator in risk_indicators: if indicator.lower() in response_lower: return True, f"Fraud investigation flagged risk indicator: '{indicator}' — compliance team review required" return False, "" class ComplianceAgent(BaseAgent): """ Handles regulatory compliance checks: - KYC and AML verification - Sanctions screening - Regulatory reporting requirements - Data protection queries (GDPR, etc.) """ agent_type = "compliance" system_prompt = """You are the Compliance Agent for our platform. Your responsibilities: - Verify customer compliance with KYC and AML requirements - Assess regulatory obligations for specific transactions or business activities - Advise on data protection requirements (GDPR, UK GDPR) - Check against sanctions lists and PEP databases - Advise on regulatory reporting thresholds Important boundaries: - You provide compliance INFORMATION and ASSESSMENT, not legal advice - For definitive legal interpretations, recommend customers consult a qualified lawyer - High-risk compliance findings MUST be reviewed by the compliance team (human) - Always cite the specific regulation or company policy you are referencing Your knowledge base contains regulatory guidelines, compliance procedures, and policy documents.""" def check_requires_human_approval(self, request: str, response: str) -> tuple[bool, str]: """Compliance actions always require human sign-off""" compliance_actions = ["report to", "file with", "notify regulator", "submit to fca", "suspicious activity report", "sar", "freeze", "restrict"] response_lower = response.lower() for action in compliance_actions: if action in response_lower: return True, f"Compliance action '{action}' requires regulatory compliance team approval" return False, "" class CustomerSupportAgent(BaseAgent): """ Handles general customer support requests: - Product questions - Technical troubleshooting - Feature guidance - Escalation routing """ agent_type = "customer_support" system_prompt = """You are the Customer Support Agent for our platform. Your responsibilities: - Answer product and feature questions - Guide customers through technical issues step by step - Explain how to use platform features effectively - Identify when issues need escalation to specialist agents Routing rules: - Billing and payment questions → route to Payment Agent - Account access problems → route to Customer Agent - Suspicious activity → route to Fraud Agent - Compliance questions → route to Compliance Agent - General product help → handle yourself Response guidelines: - Be friendly, clear, and patient - Use numbered steps for technical instructions - Always confirm the customer understood the solution - If you cannot resolve in 3 exchanges, offer to escalate Your knowledge base contains product documentation, FAQs, and troubleshooting guides.""" def check_requires_human_approval(self, request: str, response: str) -> tuple[bool, str]: """Escalation requests need human agent involvement""" escalation_terms = ["speak to a human", "talk to someone", "escalate", "manager", "supervisor", "not helpful", "frustrated"] request_lower = request.lower() for term in escalation_terms: if term in request_lower: return True, "Customer has requested human agent involvement" return False, ""

Step 5: The Custom Agent Router

The blue brain icon in the centre of your diagram. This is what receives every request and decides which specialist agent handles it.

python

# agents/router.py # The Custom Agent Router — decides which agent handles each request import boto3 import json import logging import os from .specialist_agents import ( CustomerAgent, PaymentAgent, FraudAgent, ComplianceAgent, CustomerSupportAgent ) from .base_agent import AgentResponse, bedrock logger = logging.getLogger(__name__) MODEL_ID = "anthropic.claude-3-5-haiku-20241022" # Fast model for routing decisions class AgentRouter: """ The Custom Agent Router coordinates the specialist agents. Responsibilities: 1. Receive user requests from the API 2. Classify the request and select the right agent 3. Dispatch to the selected agent 4. Receive the agent's response 5. Pass the response to Response Verification 6. Handle human approval workflows """ def __init__(self): self.agents = { "customer": CustomerAgent(), "payment": PaymentAgent(), "fraud": FraudAgent(), "compliance": ComplianceAgent(), "customer_support": CustomerSupportAgent() } # Use a fast, cheap model for routing — no need for Opus here self.bedrock = bedrock def classify_request(self, request: str, user_id: str) -> str: """ Classify which specialist agent should handle this request. Uses Claude Haiku for speed — routing is a simple classification task. """ response = self.bedrock.invoke_model( modelId=MODEL_ID, body=json.dumps({ "anthropic_version": "bedrock-2023-05-31", "max_tokens": 50, "temperature": 0.0, # No randomness — routing must be deterministic "system": """Classify this customer request into exactly ONE category. Return ONLY the category name, nothing else. Categories: - customer: Account info, profile updates, subscription changes, login help - payment: Refunds, billing disputes, invoice questions, payment processing - fraud: Suspicious activity, unusual transactions, security concerns, identity theft - compliance: KYC, AML, regulatory questions, data protection, GDPR - customer_support: General product help, technical issues, feature questions, anything else If multiple categories apply, pick the PRIMARY one.""", "messages": [{"role": "user", "content": f"Request: {request}"}] }) ) result = json.loads(response['body'].read()) agent_type = result["content"][0]["text"].strip().lower() # Validate the classification if agent_type not in self.agents: logger.warning(f"Unknown agent type '{agent_type}' — defaulting to customer_support") return "customer_support" logger.info(f"Request classified as: {agent_type}") return agent_type def route(self, request: str, session_id: str, user_id: str) -> AgentResponse: """ Main routing method. Classifies the request and dispatches to the correct specialist agent. """ logger.info(f"Router received request for session {session_id}") # Step 1: Classify the request agent_type = self.classify_request(request, user_id) # Step 2: Dispatch to the selected agent selected_agent = self.agents[agent_type] response = selected_agent.handle(request, session_id, user_id) return response

Step 6: Response Verification The Safety Net

The centre-right box in your diagram: "Response Verification Custom Code on Amazon EKS". This checks every response before it reaches the user.

python

# verification/response_verifier.py # Custom verification code running on EKS import re import json import logging from dataclasses import dataclass from typing import Optional logger = logging.getLogger(__name__) @dataclass class VerificationResult: accepted: bool rejection_reason: Optional[str] = None modified_content: Optional[str] = None class ResponseVerifier: """ Checks agent responses before they reach the user. This is a second safety layer — after Bedrock Guardrails, this code performs custom business-logic checks that are specific to the application. From the architecture: "Checks the proposed response before releasing it to the user" "Send rejection reason" back to agent if rejected "Agent-generated response accepted" if it passes """ # Patterns that indicate a response should be blocked BLOCKED_PATTERNS = [ r'\bpassword\b.{0,50}\bis\b', # Never reveal passwords r'\b4[0-9]{12}(?:[0-9]{3})?\b', # Credit card numbers r'\b[0-9]{3}-[0-9]{2}-[0-9]{4}\b', # SSN format r'<script[^>]*>', # XSS attempts in output r'SELECT .+ FROM .+ WHERE', # SQL in output (injection indicator) r'internal server error', # Exposing system errors to users r'stack trace', # Exposing technical details r'(api_key|secret_key)\s*[:=]', # Exposing credentials ] # Financial accuracy patterns — dollar amounts must be precise FINANCIAL_PATTERNS = [ r'approximately\s+£', # "Approximately £500" is not okay for financial responses r'around\s+£', r'roughly\s+£', ] # Patterns that require human review despite agent passing them ESCALATION_PATTERNS = [ r'legal action', r'regulatory authority', r'media', r'going public', r'formal complaint', ] def verify(self, response_content: str, agent_type: str, original_request: str) -> VerificationResult: """ Verify an agent response before sending it to the user. Returns: - VerificationResult with accepted=True if the response passes - VerificationResult with accepted=False and rejection_reason if blocked """ # Check 1: Blocked patterns — hard stop for pattern in self.BLOCKED_PATTERNS: if re.search(pattern, response_content, re.IGNORECASE): logger.warning(f"Response blocked: pattern '{pattern}' detected in {agent_type} response") return VerificationResult( accepted=False, rejection_reason=f"Response contains blocked content pattern. Rewrite without including {pattern}" ) # Check 2: Financial responses must be precise if agent_type == "payment": for pattern in self.FINANCIAL_PATTERNS: if re.search(pattern, response_content, re.IGNORECASE): return VerificationResult( accepted=False, rejection_reason="Financial responses must use exact amounts, not approximations. Rewrite with precise figures." ) # Check 3: Minimum quality — response must be substantive if len(response_content.strip()) < 50: return VerificationResult( accepted=False, rejection_reason="Response is too short to be helpful. Provide a complete, substantive answer." ) # Check 4: Response must be in correct language (English) # Simple heuristic — production should use a proper language detector english_word_ratio = sum( 1 for word in response_content.split() if word.lower() in {"the", "a", "is", "are", "was", "were", "have", "has", "be", "to", "of", "and", "in"} ) / max(len(response_content.split()), 1) # Check 5: No response should tell the user the agent "can't" do something # without offering an alternative path if "cannot" in response_content.lower() or "can't" in response_content.lower(): if not any(word in response_content.lower() for word in ["instead", "however", "alternatively", "suggest", "recommend", "please"]): return VerificationResult( accepted=False, rejection_reason="Response declines without offering an alternative. Rewrite to include a constructive path forward." ) # All checks passed logger.info(f"Response from {agent_type} accepted by verifier") return VerificationResult(accepted=True)

Step 7: Human Approval Workflow

The bottom-right of your diagram. When an agent flags that human review is needed, this system handles it.

python

# approval/human_approval.py # Human approval workflow using SNS + Step Functions import boto3 import json import uuid import logging import os from datetime import datetime, timezone logger = logging.getLogger(__name__) REGION = os.environ.get("AWS_REGION", "eu-west-1") ACCOUNT_ID = os.environ.get("AWS_ACCOUNT_ID") sns = boto3.client('sns', region_name=REGION) dynamodb = boto3.resource('dynamodb', region_name=REGION) approval_table = dynamodb.Table('human-approvals') class HumanApprovalWorkflow: """ Manages the human approval gate in the multi-agent system. From the architecture diagram: "Use human approval when required" "Set Functions → Human approval" "Return allow or block decision" """ APPROVAL_TOPIC_ARN = f"arn:aws:sns:{REGION}:{ACCOUNT_ID}:human-approval-topic" def request_approval( self, session_id: str, user_id: str, agent_type: str, original_request: str, proposed_response: str, approval_reason: str ) -> str: """ Submit a response for human approval. Returns an approval_id that can be used to check status. """ approval_id = str(uuid.uuid4()) # Store the approval request in DynamoDB approval_table.put_item(Item={ "approvalId": approval_id, "sessionId": session_id, "userId": user_id, "agentType": agent_type, "originalRequest": original_request, "proposedResponse": proposed_response, "approvalReason": approval_reason, "status": "PENDING", "createdAt": datetime.now(timezone.utc).isoformat(), "ttl": int(datetime.now(timezone.utc).timestamp()) + 3600 # 1 hour TTL }) # Notify the human approval team via SNS sns.publish( TopicArn=self.APPROVAL_TOPIC_ARN, Subject=f"APPROVAL REQUIRED: {agent_type.upper()} Agent — {approval_reason[:50]}", Message=json.dumps({ "approval_id": approval_id, "approval_url": f"https://admin.yourcompany.com/approvals/{approval_id}", "session_id": session_id, "user_id": user_id, "agent_type": agent_type, "reason": approval_reason, "original_request": original_request[:500], "proposed_response": proposed_response[:500], "created_at": datetime.now(timezone.utc).isoformat() }, indent=2), MessageAttributes={ "agent_type": { "DataType": "String", "StringValue": agent_type }, "priority": { "DataType": "String", "StringValue": "HIGH" if agent_type in ["fraud", "compliance"] else "MEDIUM" } } ) logger.info(f"Human approval requested: {approval_id} for session {session_id}") return approval_id def check_approval_status(self, approval_id: str) -> dict: """Check the status of a pending approval""" response = approval_table.get_item(Key={"approvalId": approval_id}) return response.get("Item", {"status": "NOT_FOUND"}) def process_approval_decision(self, approval_id: str, decision: str, approver_id: str, notes: str = "") -> bool: """ Record a human approval decision. Called by the approval UI when a human makes a decision. decision: "APPROVED" or "REJECTED" """ if decision not in ["APPROVED", "REJECTED"]: raise ValueError(f"Invalid decision: {decision}. Must be APPROVED or REJECTED") approval_table.update_item( Key={"approvalId": approval_id}, UpdateExpression="SET #s = :status, approverId = :approver, notes = :notes, decidedAt = :decided", ExpressionAttributeNames={"#s": "status"}, ExpressionAttributeValues={ ":status": decision, ":approver": approver_id, ":notes": notes, ":decided": datetime.now(timezone.utc).isoformat() } ) logger.info(f"Approval {approval_id} {decision} by {approver_id}") return decision == "APPROVED"

Step 8: The Main API Handler Bringing It All Together

python

# main.py # FastAPI application running on EKS — the entry point from API Gateway from fastapi import FastAPI, HTTPException, Header, Depends from pydantic import BaseModel from typing import Optional import logging import uuid import time import os from agents.router import AgentRouter from verification.response_verifier import ResponseVerifier from approval.human_approval import HumanApprovalWorkflow logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="Multi-Agent AI System", description="Production multi-agent AI on Amazon EKS", version="1.0.0" ) # Initialise core components router = AgentRouter() verifier = ResponseVerifier() approval_workflow = HumanApprovalWorkflow() MAX_VERIFICATION_RETRIES = 3 class AgentRequest(BaseModel): message: str user_id: str session_id: Optional[str] = None class AgentApiResponse(BaseModel): session_id: str agent_type: str response: str requires_human_approval: bool approval_id: Optional[str] = None knowledge_sources: list = [] processing_time_ms: int @app.get("/health") async def health_check(): """Health check endpoint for ALB target group""" return {"status": "healthy", "service": "multi-agent-system"} @app.post("/api/v1/agent", response_model=AgentApiResponse) async def handle_agent_request( request: AgentRequest, authorization: str = Header(...) ): """ Main endpoint — receives requests from API Gateway. Flow: 1. Validate authentication token 2. Route to correct specialist agent 3. Verify the response 4. Handle human approval if required 5. Return response to user """ start_time = time.time() # Generate session ID if not provided session_id = request.session_id or str(uuid.uuid4()) logger.info(f"Request received: session={session_id}, user={request.user_id}") # Step 1: Route to specialist agent agent_response = router.route( request=request.message, session_id=session_id, user_id=request.user_id ) # Step 2: Verify the response (up to 3 attempts) final_content = agent_response.content verification_passed = False for attempt in range(MAX_VERIFICATION_RETRIES): verification = verifier.verify( response_content=final_content, agent_type=agent_response.agent_type, original_request=request.message ) if verification.accepted: verification_passed = True break else: logger.warning( f"Verification failed (attempt {attempt + 1}): {verification.rejection_reason}" ) if attempt < MAX_VERIFICATION_RETRIES - 1: # Ask the agent to try again with the rejection reason correction_response = router.agents[agent_response.agent_type].call_bedrock_with_guardrails( messages=[ {"role": "user", "content": request.message}, {"role": "assistant", "content": final_content}, {"role": "user", "content": f"Your previous response was rejected: {verification.rejection_reason}. Please provide a corrected response."} ] ) final_content = correction_response[0] if not verification_passed: raise HTTPException( status_code=500, detail="Unable to generate an acceptable response. Please try again or contact support." ) # Step 3: Handle human approval if required approval_id = None if agent_response.requires_human_approval: approval_id = approval_workflow.request_approval( session_id=session_id, user_id=request.user_id, agent_type=agent_response.agent_type, original_request=request.message, proposed_response=final_content, approval_reason=agent_response.approval_reason ) # Return pending message to user while approval is in progress final_content = ( f"Your request has been received and is pending review by our team. " f"Reference number: {approval_id}. " f"We will respond within 1 business hour. " f"\n\nPreliminary assessment: {final_content}" ) processing_time_ms = int((time.time() - start_time) * 1000) logger.info( f"Request completed: session={session_id}, agent={agent_response.agent_type}, " f"time={processing_time_ms}ms, approval_required={agent_response.requires_human_approval}" ) return AgentApiResponse( session_id=session_id, agent_type=agent_response.agent_type, response=final_content, requires_human_approval=agent_response.requires_human_approval, approval_id=approval_id, knowledge_sources=agent_response.knowledge_sources or [], processing_time_ms=processing_time_ms )

Step 9: Deploy to EKS — Kubernetes Manifests

yaml

# k8s/namespace.yaml apiVersion: v1 kind: Namespace metadata: name: multi-agent labels: app: multi-agent-system

yaml

# k8s/service-account.yaml # IRSA — IAM Roles for Service Accounts # Links the Kubernetes service account to the IAM role we created in Step 1 apiVersion: v1 kind: ServiceAccount metadata: name: agent-service-account namespace: multi-agent annotations: # Replace with your actual role ARN eks.amazonaws.com/role-arn: arn:aws:iam::YOUR_ACCOUNT_ID:role/multi-agent-pod-role

yaml

# k8s/configmap.yaml apiVersion: v1 kind: ConfigMap metadata: name: agent-config namespace: multi-agent data: AWS_REGION: "eu-west-1" BEDROCK_REGION: "us-east-1" LOG_LEVEL: "INFO" MAX_VERIFICATION_RETRIES: "3"

yaml

# k8s/secrets.yaml # In production: use AWS Secrets Manager with External Secrets Operator # For initial setup: apiVersion: v1 kind: Secret metadata: name: agent-secrets namespace: multi-agent type: Opaque stringData: GUARDRAIL_ID: "YOUR_GUARDRAIL_ID" GUARDRAIL_VERSION: "1" KNOWLEDGE_BASE_ID: "YOUR_KB_ID"

yaml

# k8s/deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: multi-agent-system namespace: multi-agent labels: app: multi-agent version: v1 spec: replicas: 3 # 3 replicas across 2 AZs selector: matchLabels: app: multi-agent template: metadata: labels: app: multi-agent version: v1 spec: serviceAccountName: agent-service-account # IRSA for AWS API access containers: - name: multi-agent image: YOUR_ECR_REGISTRY/multi-agent:latest ports: - containerPort: 8000 envFrom: - configMapRef: name: agent-config - secretRef: name: agent-secrets resources: requests: cpu: "500m" memory: "1Gi" limits: cpu: "2000m" memory: "4Gi" readinessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 10 periodSeconds: 5 livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 securityContext: runAsNonRoot: true # Never run as root runAsUser: 1000 readOnlyRootFilesystem: true affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: labelSelector: matchLabels: app: multi-agent topologyKey: kubernetes.io/hostname # Spread across nodes

yaml

# k8s/service.yaml apiVersion: v1 kind: Service metadata: ...