How to Build a compliance checking Agent Using CrewAI in Python for payments

A compliance checking agent for payments reviews a transaction, merchant profile, and policy set, then flags whether the payment can proceed, needs manual review, or must be blocked. It matters because payment teams need fast decisions without losing control over AML, sanctions, PCI scope, data residency, and auditability.

Architecture

• •

  Input normalizer

  • •Converts raw payment events into a structured payload.
  • •Pulls out amount, currency, merchant country, customer country, MCC, card-present flag, and beneficiary details.
• •

  Policy context store

  • •Holds internal compliance rules plus external constraints like sanctions lists and jurisdiction-specific thresholds.
  • •Keeps versioned policies so every decision can be replayed later.
• •

  Compliance analyst agent

  • •Uses CrewAI Agent to evaluate the transaction against policy.
  • •Produces a decision with explicit reasons and risk signals.
• •

  Evidence retrieval tool

  • •Fetches supporting facts from your internal systems: KYC status, merchant risk score, prior alerts, watchlist hits.
  • •Keeps the agent grounded in system-of-record data.
• •

  Decision workflow

  • •Uses CrewAI Task and Crew to orchestrate review steps.
  • •Separates analysis from final recommendation so you can add human approval later.
• •

  Audit logger

  • •Stores input snapshot, policy version, model output, tool calls, and final disposition.
  • •Required for investigations and regulator queries.

Implementation

1) Install CrewAI and define the compliance schema

Use a strict Pydantic schema so the agent returns structured output you can persist and act on. For payments, that structure should include a decision and an audit-friendly explanation.

from pydantic import BaseModel
from typing import Literal

class ComplianceDecision(BaseModel):
    decision: Literal["approve", "review", "block"]
    risk_level: Literal["low", "medium", "high"]
    reasons: list[str]
    policy_refs: list[str]

2) Build a tool that fetches payment evidence

In production, this tool should call your internal APIs or data warehouse. Keep it read-only; the agent should never mutate customer or merchant records.

from crewai.tools import BaseTool

class FetchPaymentEvidenceTool(BaseTool):
    name: str = "fetch_payment_evidence"
    description: str = "Fetch KYC, sanctions screening, merchant risk score, and transaction context."

    def _run(self, transaction_id: str) -> str:
        # Replace with real API calls
        evidence = {
            "transaction_id": transaction_id,
            "merchant_country": "GB",
            "customer_country": "NG",
            "amount": 2500,
            "currency": "USD",
            "merchant_risk_score": 82,
            "kyc_status": "verified",
            "sanctions_hit": False,
            "watchlist_hit": False,
            "policy_version": "payments-compliance-v7"
        }
        return str(evidence)

3) Create the compliance agent and task

Use a narrow role. Don’t ask the model to “be smart”; tell it exactly what to check. The prompt should force explicit references to policy and evidence.

from crewai import Agent, Task

compliance_agent = Agent(
    role="Payments Compliance Analyst",
    goal="Assess payment transactions against compliance policy and return a structured decision.",
    backstory=(
        "You review payment transactions for AML, sanctions exposure, fraud indicators,"
        " PCI-sensitive handling issues, and jurisdictional restrictions."
    ),
    tools=[FetchPaymentEvidenceTool()],
    verbose=True,
)

compliance_task = Task(
    description=(
        "Review transaction {transaction_id}. "
        "Use the evidence tool if needed. "
        "Return a compliance decision with reasons and policy references. "
        "Block transactions with sanctions hits or clear policy violations. "
        "Recommend manual review for high-risk cross-border payments or incomplete KYC."
    ),
    expected_output="A structured compliance decision matching the ComplianceDecision schema.",
    agent=compliance_agent,
)

4) Run the crew and parse the result

This is the actual orchestration pattern you want in service code. Keep the crew small for deterministic behavior.

from crewai import Crew, Process

crew = Crew(
    agents=[compliance_agent],
    tasks=[compliance_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"transaction_id": "txn_10492"})
print(result)

If you want stronger structure in your application layer, validate the output before persisting it:

decision = ComplianceDecision.model_validate_json(str(result))
print(decision.decision)
print(decision.policy_refs)

Production Considerations

• •

  Persist every decision with immutable metadata

  • •Store transaction snapshot, policy version, model output, tool responses, timestamp, and reviewer identity.
  • •You need this for chargeback disputes, SAR/STR workflows, and regulator audits.
• •

  Keep payment data residency in mind

  • •If your platform processes EU or regional payments, route prompts and tool outputs through approved regions only.
  • •Don’t send PANs or sensitive cardholder data into the agent unless you’ve reduced scope under PCI controls.
• •

  Add hard guardrails outside the model

  • •Sanctions hits should block before generation.
  • •Threshold rules like velocity limits or prohibited corridors should be enforced in code before the crew runs.
• •

  Monitor false positives by corridor and merchant segment

  • •Track approval rate vs review rate by country pair, MCC, amount band, and payment rail.
  • •A compliant system that blocks too much will kill conversion; tune with real outcomes.

Common Pitfalls

• •

  Letting the model make final decisions without deterministic rules

  • •Fix this by pre-screening for hard failures like sanctions matches or unsupported jurisdictions.
  • •The agent should recommend; code should enforce critical blocks.
• •

  Passing raw sensitive fields into prompts

  • •Don’t include full PANs, CVVs, or unnecessary PII.
  • •Tokenize or redact before calling Crew.kickoff().
• •

  Skipping policy versioning

  • •If you don’t pin policy_version, you can’t explain why a payment was approved last week but blocked today.
  • •Version policies alongside code deployments.
• •

  Using vague outputs

  • •“Looks risky” is useless in production.
  • •Require explicit reasons and policy_refs, then reject outputs that don’t validate against your schema.

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.