AI Agents in Practice: Fraud Detection & Prevention Agent (Prompts + Code)

Introduction

In this eighth article of the AI Agents series, we focus on the Fraud Detection & Prevention Agent, a vital agent for any business that handles sensitive data, payments, or user transactions. This agent’s main job is to automatically flag fraudulent activities, detect suspicious patterns, and prevent financial losses. It leverages advanced machine learning techniques, as well as established policies, to monitor transactions in real-time, alerting teams or blocking potentially fraudulent transactions. Just like the previous agents, this system never claims success unless all actions are fully executed and receipts are generated.

The Use Case

In financial sectors, e-commerce, and other data-sensitive industries, fraud detection is a critical aspect of the business. The agent examines transactions based on certain patterns, such as unusual spending behavior, location inconsistencies, or account activity that deviates from the norm. It integrates with payment gateways, banks, and other third-party services to detect irregularities in real-time and alert the relevant teams or automatically block fraudulent transactions. By using machine learning and predefined rules, the agent can prevent significant losses before they occur.

Prompt Contract (agent interface)

# file: contracts/fraud_detection_v1.yaml
role: "FraudDetectionAgent"
scope: >
  Detect fraudulent activities by analyzing transaction data, customer behavior, and account anomalies.
  Ask once for missing fields (transaction_id, customer_id, amount, location, transaction_type).
  Propose tool calls; never assert success without a receipt.
output:
  type: object
  required: [summary, decision, fraud_alert, citations, next_steps, tool_proposals]
  properties:
    summary: {type: string, maxWords: 100}
    decision: {type: string, enum: ["approve","block","need_approval","need_more_info"]}
    fraud_alert:
      type: object
      required: [transaction_id, customer_id, amount, risk_level, flagged_reason]
      properties:
        transaction_id: {type: string}
        customer_id: {type: string}
        amount: {type: integer}
        risk_level: {type: string, enum: ["low", "medium", "high"]}
        flagged_reason: {type: string}
    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: [AnalyzeTransaction, CheckAnomalyPatterns, BlockTransaction, RequestApproval]}
          args: {type: object}
          preconditions: {type: string}
          idempotency_key: {type: string}
policy_id: "fraud_detection_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:fraud:location_mismatch","text":"Transactions from new locations or IP addresses need further scrutiny.",
   "effective_date":"2025-01-01","source_id":"doc:fraud_detection_policy_v4","span":"new locations or IP addresses"},
  {"claim_id":"policy:fraud:high_amount","text":"Transactions exceeding $500 should be flagged for review if the user hasn't made similar payments in the past.",
   "effective_date":"2025-01-01","source_id":"doc:fraud_detection_policy_v4","span":"exceeding $500 should be flagged"},
  {"claim_id":"policy:fraud:unusual_behavior","text":"Transactions with frequent logins from different devices or IP addresses must be flagged.",
   "effective_date":"2025-01-01","source_id":"doc:fraud_detection_policy_v4","span":"frequent logins from different devices"}
]

Tool Interfaces (typed, with receipts)

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

class AnalyzeTransactionArgs(BaseModel):
    transaction_id: str
    customer_id: str
    amount: int
    location: str
    transaction_type: str

class CheckAnomalyPatternsArgs(BaseModel):
    customer_id: str
    transaction_history: List[Dict[str, str]]  # e.g., [{amount: int, location: str, transaction_type: str}]

class BlockTransactionArgs(BaseModel):
    transaction_id: str
    customer_id: str
    amount: int
    flagged_reason: str

class RequestApprovalArgs(BaseModel):
    transaction_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

TRANSACTIONS = {
    "trans-001": {"customer_id": "C001", "amount": 1500, "location": "New York", "transaction_type": "purchase", "approved": False},
    "trans-002": {"customer_id": "C002", "amount": 100, "location": "California", "transaction_type": "transfer", "approved": False}
}

def analyze_transaction(a: AnalyzeTransactionArgs) -> ToolReceipt:
    if a.amount > 1000:
        risk_level = "high"
        flagged_reason = "High transaction amount"
    else:
        risk_level = "low"
        flagged_reason = "Normal transaction"
    return ToolReceipt(tool="AnalyzeTransaction", ok=True, ref=f"trans-{a.transaction_id}",
                       message="Transaction analyzed", data={"risk_level": risk_level, "flagged_reason": flagged_reason})

def check_anomaly_patterns(a: CheckAnomalyPatternsArgs) -> ToolReceipt:
    if len(a.transaction_history) > 5:  # Simple anomaly detection logic
        return ToolReceipt(tool="CheckAnomalyPatterns", ok=False, ref="pattern-mismatch",
                           message="Customer has frequent logins from different locations", data={"flagged": True})
    return ToolReceipt(tool="CheckAnomalyPatterns", ok=True, ref="pattern-ok", message="No anomalies detected")

def block_transaction(a: BlockTransactionArgs) -> ToolReceipt:
    TRANSACTIONS[a.transaction_id]["approved"] = False
    return ToolReceipt(tool="BlockTransaction", ok=True, ref=f"block-{a.transaction_id}",
                       message=f"Transaction {a.transaction_id} blocked", data={"reason": a.flagged_reason})

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

Agent Loop (proposal → verification → execution → receipts)

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

ALLOWED_TOOLS = {"AnalyzeTransaction", "CheckAnomalyPatterns", "BlockTransaction", "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 == "AnalyzeTransaction":      return analyze_transaction(AnalyzeTransactionArgs(**a))
    if n == "CheckAnomalyPatterns":    return check_anomaly_patterns(CheckAnomalyPatternsArgs(**a))
    if n == "BlockTransaction":        return block_transaction(BlockTransactionArgs(**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]], transaction_data: Dict[str,Any]) -> Dict[str,Any]:
    decision = "approve" if transaction_data["amount"] <= 1000 else "block"
    return {
      "summary": f"Transaction for {transaction_data['customer_id']} reviewed for fraud.",
      "decision": decision,
      "fraud_alert": {
        "transaction_id": transaction_data["transaction_id"],
        "customer_id": transaction_data["customer_id"],
        "amount": transaction_data["amount"],
        "risk_level": "high" if decision == "block" else "low",
        "flagged_reason": "High transaction amount" if decision == "block" else "Normal transaction"
      },
      "citations": ["policy:fraud:location_mismatch", "policy:fraud:high_amount", "policy:fraud:unusual_behavior"],
      "next_steps": ["Check transaction patterns", "Request approval for blocked transactions", "Block or approve transaction"],
      "tool_proposals": [
        {"name":"AnalyzeTransaction","args":{"transaction_id":transaction_data["transaction_id"],"customer_id":transaction_data["customer_id"],
                                             "amount":transaction_data["amount"],"location":transaction_data["location"],
                                             "transaction_type":transaction_data["transaction_type"]},
         "preconditions":"Analyze transaction for fraud risk.","idempotency_key": new_idem()},
        {"name":"CheckAnomalyPatterns","args":{"customer_id":transaction_data["customer_id"],
                                               "transaction_history":[{"amount":transaction_data["amount"], "location":transaction_data["location"], "transaction_type":transaction_data["transaction_type"}]},
         "preconditions":"Check for unusual transaction patterns.","idempotency_key": new_idem()},
        {"name":"BlockTransaction","args":{"transaction_id":transaction_data["transaction_id"],"customer_id":transaction_data["customer_id"],
                                           "amount":transaction_data["amount"],"flagged_reason":"High transaction amount"},
         "preconditions":"Block transaction if fraud risk is high.","idempotency_key": new_idem()},
        {"name":"RequestApproval","args":{"transaction_id":transaction_data["transaction_id"],"approver_id":"M002"},
         "preconditions":"Request manager approval if the transaction is blocked.","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"Fraud Alert: {model_json['fraud_alert']}")
    lines.append("")
    lines.append("Next steps:")
    for s in model_json["next_steps"]:
        lines.append(f"• {s}")
    if idx.get("BlockTransaction") and idx["BlockTransaction"].ok:
        lines.append(f"\nTransaction blocked: {idx['BlockTransaction'].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(transaction_data: Dict[str,Any]) -> str:
    contract = open("contracts/fraud_detection_v1.yaml").read()
    claims: List[Dict[str,Any]] = []  # load real claims from policy
    plan = call_model(contract, claims, transaction_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 {"BlockTransaction"}:
            break
    return render_response(plan, receipts)

if __name__ == "__main__":
    example_transaction_data = {
      "transaction_id": "trans-001",
      "customer_id": "C001",
      "amount": 1500,
      "location": "New York",
      "transaction_type": "purchase"
    }
    print(handle(example_transaction_data))

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

System:
You are FraudDetectionAgent. Follow the contract:
- Ask once if transaction_id, customer_id, amount, location, or transaction_type 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, fraud_alert, citations[], next_steps[], tool_proposals[].

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

User:
Review this transaction for fraud: {"transaction_id":"trans-001","customer_id":"C001","amount":1500,"location":"New York","transaction_type":"purchase"}

How to adapt quickly

Integrate with your fraud detection systems, bank transaction gateways, and AML (Anti-Money Laundering) systems to detect anomalies in transactions. Implement idempotency and transactional integrity to ensure no accidental double-blocks. Load claims from fraud prevention policies related to payment types, geolocation checks, and transaction patterns. Add a validation layer to ensure compliance with regulatory guidelines such as PCI-DSS or GDPR. Deploy the contract, policy bundle, and decoder settings behind a feature flag for testing, canary deployments, and rollback support.