AI Agents in Practice: Automated Customer Onboarding & Verification Agent (Prompts + Code)

Introduction

This article introduces an Automated Customer Onboarding & Verification Agent, a critical component in streamlining the customer onboarding process for businesses. This agent automates the collection of customer information, performs identity verification, checks for compliance with regulations (such as KYC—Know Your Customer), and ensures that customer data is properly processed and stored. The agent can issue a receipt for each completed action, ensuring transparency and security. By automating repetitive tasks, this agent improves efficiency and reduces human error.

The Use Case

Customer onboarding typically involves multiple stages: capturing basic information, verifying identity, checking for compliance with relevant policies, and setting up user accounts or subscriptions. The agent automates each of these steps, ensuring that the correct information is collected, verification checks are conducted according to the latest regulations, and compliance is assured. After completing these tasks, the agent either approves the onboarding or requests additional information or manual verification if required.

Prompt Contract (agent interface)

# file: contracts/onboarding_verification_v1.yaml
role: "CustomerOnboardingAgent"
scope: >
  Collect customer details, verify identity, ensure compliance with regulatory requirements, and create customer accounts.
  Ask once if critical fields are missing (customer_id, full_name, email, identity_document).
  Propose tool calls; never assert success without a receipt.
output:
  type: object
  required: [summary, decision, customer_data, next_steps, tool_proposals]
  properties:
    summary: {type: string, maxWords: 100}
    decision: {type: string, enum: ["approve", "reject", "need_approval", "need_more_info"]}
    customer_data:
      type: object
      required: [customer_id, full_name, email, identity_verified]
      properties:
        customer_id: {type: string}
        full_name: {type: string}
        email: {type: string}
        identity_verified: {type: boolean}
    citations: {type: array, items: {type: string}}
    next_steps: {type: array, items: {type: string}, maxItems: 6}
    tool_proposals:
      type: array
      items:
        type: object
        required: [name, args, preconditions, idempotency_key]
        properties:
          name: {type: string, enum: [CollectCustomerData, VerifyIdentity, CheckCompliance, CreateAccount, RequestApproval]}
          args: {type: object}
          preconditions: {type: string}
          idempotency_key: {type: string}
policy_id: "onboarding_policy_v4"
citation_rule: "1–2 minimal-span claim_ids per factual sentence"
decoding:
  narrative: {top_p: 0.92, temperature: 0.72, stop: ["\n\n## "]}
  bullets:   {top_p: 0.82, temperature: 0.45}

Example claims (context provided to the model)

[
  {"claim_id":"policy:kyc:identity_check","text":"All customers must complete an identity check before account creation.",
   "effective_date":"2025-01-01","source_id":"doc:onboarding_policy_v4","span":"complete an identity check"},
  {"claim_id":"policy:email:format","text":"Customer email must follow a valid format (e.g., [email protected]).",
   "effective_date":"2025-01-01","source_id":"doc:onboarding_policy_v4","span":"valid email format"},
  {"claim_id":"policy:compliance:customer_data","text":"Customer data must be stored securely and in compliance with GDPR.",
   "effective_date":"2025-01-01","source_id":"doc:onboarding_policy_v4","span":"stored securely and in compliance with GDPR"}
]

Tool Interfaces (typed, with receipts)

# tools.py
from pydantic import BaseModel
from typing import Optional, List, Dict
from datetime import datetime

class CollectCustomerDataArgs(BaseModel):
    customer_id: str
    full_name: str
    email: str
    identity_document: str

class VerifyIdentityArgs(BaseModel):
    customer_id: str
    identity_document: str

class CheckComplianceArgs(BaseModel):
    customer_id: str
    data_compliance: bool

class CreateAccountArgs(BaseModel):
    customer_id: str
    full_name: str
    email: str

class RequestApprovalArgs(BaseModel):
    customer_id: str
    approver_id: str

class ToolReceipt(BaseModel):
    tool: str
    ok: bool
    ref: str
    message: str = ""
    data: Optional[Dict] = None

# adapters.py  (demo logic)
from tools import *
from datetime import datetime

CUSTOMERS = {
    "C001": {"full_name": "John Doe", "email": "[email protected]", "identity_document": "id12345", "identity_verified": False},
    "C002": {"full_name": "Jane Smith", "email": "[email protected]", "identity_document": "id54321", "identity_verified": False}
}

def collect_customer_data(a: CollectCustomerDataArgs) -> ToolReceipt:
    return ToolReceipt(tool="CollectCustomerData", ok=True, ref=f"data-{a.customer_id}",
                       message="Customer data collected", data={"customer_id": a.customer_id})

def verify_identity(a: VerifyIdentityArgs) -> ToolReceipt:
    customer = CUSTOMERS.get(a.customer_id)
    if not customer:
        return ToolReceipt(tool="VerifyIdentity", ok=False, ref="customer-not-found", message="Customer not found")
    customer["identity_verified"] = True
    return ToolReceipt(tool="VerifyIdentity", ok=True, ref=f"identity-{a.customer_id}", message="Identity verified")

def check_compliance(a: CheckComplianceArgs) -> ToolReceipt:
    if not a.data_compliance:
        return ToolReceipt(tool="CheckCompliance", ok=False, ref="compliance-violation", message="Data compliance check failed")
    return ToolReceipt(tool="CheckCompliance", ok=True, ref="compliance-ok", message="Data complies with regulations")

def create_account(a: CreateAccountArgs) -> ToolReceipt:
    customer = CUSTOMERS.get(a.customer_id)
    if not customer:
        return ToolReceipt(tool="CreateAccount", ok=False, ref="customer-not-found", message="Customer not found")
    customer["account_created"] = True
    return ToolReceipt(tool="CreateAccount", ok=True, ref=f"account-{a.customer_id}", message="Account created")

def request_approval(a: RequestApprovalArgs) -> ToolReceipt:
    return ToolReceipt(tool="RequestApproval", ok=True, ref=f"approval-{a.customer_id}",
                       message=f"Approval requested for {a.customer_id}'s account.", data={"approver_id": a.approver_id})

Agent Loop (proposal → verification → execution → receipts)

# agent_onboarding.py
import uuid, json
from typing import Any, Dict, List
from tools import *
from adapters import *

ALLOWED_TOOLS = {"CollectCustomerData", "VerifyIdentity", "CheckCompliance", "CreateAccount", "RequestApproval"}

def new_idem() -> str:
    return f"idem-{uuid.uuid4()}"

def verify_proposal(p: Dict[str, Any]) -> str:
    required = {"name","args","preconditions","idempotency_key"}
    if not required.issubset(p): return "Missing proposal fields"
    if p["name"] not in ALLOWED_TOOLS: return "Tool not allowed"
    return ""

def execute(p: Dict[str, Any]) -> ToolReceipt:
    n, a = p["name"], p["args"]
    if n == "CollectCustomerData":    return collect_customer_data(CollectCustomerDataArgs(**a))
    if n == "VerifyIdentity":         return verify_identity(VerifyIdentityArgs(**a))
    if n == "CheckCompliance":        return check_compliance(CheckComplianceArgs(**a))
    if n == "CreateAccount":          return create_account(CreateAccountArgs(**a))
    if n == "RequestApproval":        return request_approval(RequestApprovalArgs(**a))
    return ToolReceipt(tool=n, ok=False, ref="none", message="Unknown tool")

# --- Model shim returning a plan per contract (replace with your LLM call) ---
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], customer_data: Dict[str,Any]) -> Dict[str,Any]:
    decision = "approve" if customer_data["identity_verified"] else "need_approval"
    return {
      "summary": f"Customer {customer_data['full_name']} onboarding check complete.",
      "decision": decision,
      "customer_data": {
        "customer_id": customer_data["customer_id"],
        "full_name": customer_data["full_name"],
        "email": customer_data["email"],
        "identity_verified": customer_data["identity_verified"]
      },
      "citations": ["policy:kyc:identity_check","policy:email:format","policy:compliance:customer_data"],
      "next_steps": ["Verify identity", "Check compliance", "Create account", "Request approval"],
      "tool_proposals": [
        {"name":"CollectCustomerData","args":{"customer_id":customer_data["customer_id"], "full_name":customer_data["full_name"], "email":customer_data["email"], "identity_document":customer_data["identity_document"]},
         "preconditions":"Collect customer details.","idempotency_key": new_idem()},
        {"name":"VerifyIdentity","args":{"customer_id":customer_data["customer_id"], "identity_document":customer_data["identity_document"]},
         "preconditions":"Verify customer identity.","idempotency_key": new_idem()},
        {"name":"CheckCompliance","args":{"customer_id":customer_data["customer_id"], "data_compliance":True},
         "preconditions":"Ensure data compliance.","idempotency_key": new_idem()},
        {"name":"CreateAccount","args":{"customer_id":customer_data["customer_id"], "full_name":customer_data["full_name"], "email":customer_data["email"]},
         "preconditions":"Create customer account.","idempotency_key": new_idem()}
      ]
    }

def render_response(model_json: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
    idx = {r.tool:r for r in receipts}
    lines = [model_json["summary"], ""]
    lines.append(f"Decision: {model_json['decision']}")
    lines.append(f"Customer Details: {model_json['customer_data']}")
    lines.append("")
    lines.append("Next steps:")
    for s in model_json["next_steps"]:
        lines.append(f"• {s}")
    if idx.get("CreateAccount") and idx["CreateAccount"].ok:
        lines.append(f"\nAccount created: {idx['CreateAccount'].message}")
    if idx.get("RequestApproval") and idx["RequestApproval"].ok:
        lines.append(f"Approval requested: {idx['RequestApproval'].message}")
    lines.append("\nCitations: " + ", ".join(model_json["citations"]))
    return "\n".join(lines)

def handle(customer_data: Dict[str,Any]) -> str:
    contract = open("contracts/onboarding_verification_v1.yaml").read()
    claims: List[Dict[str,Any]] = []  # load real claims from policy
    plan = call_model(contract, claims, customer_data)

    receipts: List[ToolReceipt] = []
    for prop in plan["tool_proposals"]:
        reason = verify_proposal(prop)
        if reason:
            receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason)); continue
        rec = execute(prop)
        receipts.append(rec)
        if not rec.ok and prop["name"] in {"CreateAccount"}:
            break
    return render_response(plan, receipts)

if __name__ == "__main__":
    example_customer_data = {
      "customer_id":"C003",
      "full_name":"Alice Johnson",
      "email":"[email protected]",
      "identity_document":"id67890",
      "identity_verified": False
    }
    print(handle(example_customer_data))

The Prompt You’d Send to the Model (concise and testable)

System:
You are CustomerOnboardingAgent. Follow the contract:
- Ask once if customer_id, full_name, email, or identity_document is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a receipt.
- Output JSON with keys: summary, decision, customer_data, citations[], next_steps[], tool_proposals[].

Claims (eligible only):
[ ... JSON array of onboarding policy claims like above ... ]

User:
Onboard this new customer:
{"customer_id":"C003","full_name":"Alice Johnson","email":"[email protected]","identity_document":"id67890"}

How to adapt quickly

Integrate identity verification systems, compliance checks (e.g., GDPR), and customer account creation with your internal CRM and KYC systems. Add idempotency to ensure no accidental double submissions or customer record creation. Load claims from company policies regarding data protection, email formats, and document requirements. Add validation layers to ensure compliance with legal regulations, such as data residency rules. Ship the contract, policy bundle, and decoder settings behind a feature flag to enable easy testing and canary deployments with rollback support.