How to Build a KYC verification Agent Using CrewAI in Python for payments

A KYC verification agent for payments takes customer identity data, checks it against policy and external sources, and returns a decision with evidence. It matters because onboarding speed, fraud risk, and regulatory compliance all depend on getting this right without pushing sensitive data into uncontrolled workflows.

Architecture

• •

  Input normalizer

  • •Accepts customer name, DOB, address, document metadata, and jurisdiction.
  • •Converts raw payloads into a strict schema before any agent sees them.
• •

  Policy engine

  • •Encodes KYC rules by country, product type, and risk tier.
  • •Decides what checks are mandatory: sanctions, PEP, ID document validation, address verification.
• •

  CrewAI agents

  • •One agent handles document review.
  • •One agent handles sanctions/PEP reasoning.
  • •One agent consolidates findings into a final recommendation.
• •

  Tools layer

  • •Wraps external services like OCR, sanctions screening APIs, and internal customer profile lookups.
  • •Keeps network calls out of the agent prompt and inside deterministic Python functions.
• •

  Audit logger

  • •Persists inputs, tool outputs, decisions, timestamps, model version, and rationale.
  • •Required for payments compliance reviews and dispute handling.
• •

  Decision API

  • •Exposes the result to onboarding or payment orchestration systems.
  • •Returns approve, reject, or manual_review plus evidence.

Implementation

1) Define the data model and tools

Keep the agent away from raw JSON blobs. Use Pydantic for strict input validation and Python functions as tools so every external lookup is explicit and auditable.

from typing import List
from pydantic import BaseModel, Field
from crewai.tools import tool

class KYCRequest(BaseModel):
    customer_id: str
    full_name: str
    date_of_birth: str
    country: str
    address: str
    document_number: str | None = None
    risk_level: str = Field(default="standard")

class KYCResult(BaseModel):
    decision: str
    reasons: List[str]
    evidence: List[str]

@tool("screen_sanctions")
def screen_sanctions(full_name: str, country: str) -> str:
    # Replace with your sanctioned-party provider call.
    matches = []
    if "test" in full_name.lower():
        matches.append("Possible name similarity hit")
    return "clear" if not matches else "; ".join(matches)

@tool("verify_document")
def verify_document(document_number: str | None) -> str:
    if not document_number:
        return "missing_document"
    return "document_format_valid"

@tool("lookup_customer_profile")
def lookup_customer_profile(customer_id: str) -> str:
    # Replace with internal CRM / core banking lookup.
    return f"customer_id={customer_id}; residency=verified; account_age=18_months"

2) Build the agents and tasks

Use separate agents for narrow responsibilities. In payments workflows, this keeps reasoning bounded and makes audit trails easier to defend.

from crewai import Agent, Task, Crew, Process

kyc_analyst = Agent(
    role="KYC Analyst",
    goal="Assess identity risk for payment onboarding using available evidence.",
    backstory="You review identity checks for regulated payment products.",
    tools=[screen_sanctions, verify_document, lookup_customer_profile],
    verbose=True,
)

compliance_reviewer = Agent(
    role="Compliance Reviewer",
    goal="Apply payment KYC policy and produce a final decision.",
    backstory="You enforce KYC standards for card acquiring and wallet onboarding.",
    verbose=True,
)

review_task = Task(
    description=(
        "Review the customer's identity data. "
        "Check sanctions exposure, document status, and customer profile. "
        "Return a concise assessment with approve/reject/manual_review."
    ),
    expected_output="A structured KYC assessment with decision and reasons.",
    agent=kyc_analyst,
)

decision_task = Task(
    description=(
        "Using the analyst output, decide whether the customer should be approved "
        "for payment onboarding. Include compliance reasons and audit-friendly evidence."
    ),
    expected_output="Final decision with reasons and evidence.",
    agent=compliance_reviewer,
)

3) Run the crew and shape the result

For production systems you want one orchestration path that returns a predictable object. CrewAI’s Crew plus Process.sequential is enough for a first pass because each task depends on the previous one.

def run_kyc_check(request: KYCRequest) -> dict:
    crew = Crew(
        agents=[kyc_analyst, compliance_reviewer],
        tasks=[review_task, decision_task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff(inputs={
        "customer_id": request.customer_id,
        "full_name": request.full_name,
        "date_of_birth": request.date_of_birth,
        "country": request.country,
        "address": request.address,
        "document_number": request.document_number or "",
        "risk_level": request.risk_level,
    })

    return {
        "customer_id": request.customer_id,
        "crew_output": str(result),
    }

4) Add a deterministic policy layer around the agent

Do not let the model invent policy. Let code enforce hard rules such as missing documents or high-risk jurisdictions before you call the crew.

HIGH_RISK_COUNTRIES = {"IR", "KP", "SY"}

def precheck_policy(req: KYCRequest) -> list[str]:
    violations = []
    if req.country in HIGH_RISK_COUNTRIES:
        violations.append("high_risk_jurisdiction")
    if not req.document_number:
        violations.append("missing_identity_document")
    return violations

def kyc_pipeline(req: KYCRequest) -> dict:
    violations = precheck_policy(req)
    
    if violations:
        return {
            "decision": "manual_review",
            "reasons": violations,
            "evidence": ["policy_precheck_failed"],
        }

# Example usage:
# req = KYCRequest(...)
# print(kyc_pipeline(req))

Production Considerations

• •

  Deploy close to your data boundary

  • •Keep PII inside your approved region.
  • •If your payments stack has EU customers, run the agent in an EU region and keep tool calls regional too.
• •

  Log everything needed for audit

  • •Store input hashes, tool outputs, timestamps, model version, prompt version, and final decision.
  • •Regulators care about why you approved or rejected a payer.
• •

  Add guardrails before execution

  • •Block unsupported countries, missing documents, or incomplete profiles in code.
  • •Use manual review routing for ambiguous cases instead of forcing an automated approval.
• •

  Monitor false positives and reviewer overrides

  • •Track sanction-screen hit rates, manual review rate, approval latency, and downstream chargeback/fraud signals.
  • •A good KYC system is not just accurate; it is explainable under pressure.

Common Pitfalls

• •

  Putting raw compliance policy inside prompts

  • •Mistake: asking the model to “know” your AML/KYC rules.
  • •Fix: enforce hard rules in Python first; let the agent reason over evidence only.
• •

  Skipping auditability

  • •Mistake: returning only approved or rejected.
  • •Fix: persist reasons from each tool call plus final rationale so compliance can reconstruct the path later.
• •

  Letting external tools return unstructured text only

  • •Mistake: feeding free-form API responses directly into the next task.
  • •Fix: normalize tool outputs into stable fields like status, match_score, source, and checked_at.

If you’re building this for payments at scale, keep the model narrow. The code should decide what must happen; CrewAI should help interpret evidence and produce a defensible outcome.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.