Agentic AI-Powered Payments Orchestration

Data-Driven Message Translation

Legacy systems often hard-code translation logic (for example, "field 4 maps to field 32A"). This solution stores translation rules and metadata in MongoDB alongside the data, making the translation logic data-driven.

Instead of recompiling code to change a mapping, you update a configuration document in MongoDB. The converter service reads these rules dynamically—extract patterns, field mappings, and output paths—to normalize incoming ISO 8583 or SWIFT MT messages into the canonical JSON format.

Config-Driven Transformation:

# Load config from MongoDB
config = db.conversion_configs.find_one({"_id": "MT103_to_JSON"})
# Extract fields using patterns from config["extract"]
for field_id, pattern in config["extract"].items():
    extracted[field_id] = re.search(pattern, message).group(1)
# Transform using rules from config["map"]
for mapping in config["map"]:
    value = extracted[mapping["from"]]
    if "ai" not in mapping:                  # RULES lane
        output[mapping["to"]] = value
    else:                                    # AI lane
        ai_fields.append({"value": value, "field_type": mapping["ai"]})

Some fields resist regex patterns such as SWIFT Field 70 (remittance information) can contain free-text payment purposes, invoice references, and instructions across multiple lines. The configuration marks these fields for AI processing ("ai": "remittance"), and the LLM extracts structured data from the unstructured text. This approach keeps deterministic rules for most fields but handles the complex remainder through inference.

New message formats are enabled through document inserts—the converter code stays unchanged, keeping translation logic out of the deployment cycle.

Gen AI-Powered Adaptive Conversion

With rules and historical metadata stored in the database, you can use Generative AI (LLMs) to accelerate onboarding of new payment formats.

To add support for a new message type, provide a sample message. The Adaptive Learning Service loads all existing conversion configurations from MongoDB and extracts fields using format-specific parsing. It matches detected field IDs against mappings from existing configurations. For example, if field "32A" exists in MT103_to_JSON, that exact mapping is reused.

For fields not found in any existing config, an LLM suggests mappings that you review and refine. This generates a draft configuration ready for human review and deployment. Payment rails that once took months to integrate can now be onboarded faster.

Adaptive Learning with Aggregation and LLM:

# Aggregation pipeline: extract field mappings from ALL existing configs
pipeline = [
    {"$addFields": {"extract_array": {"$objectToArray": "$extract"}}},
    {"$unwind": "$map"},
    {"$project": {"field_id": "$map.from", "mapping": "$map", "source_config": "$_id"}},
    {"$group": {"_id": "$field_id", "mapping": {"$first": "$mapping"}}}
]
field_lookup = {doc["_id"]: doc for doc in db.conversion_configs.aggregate(pipeline)}
# Match detected fields against learned patterns
for field_id in detected_fields:
    if field_id in field_lookup:
        matched[field_id] = field_lookup[field_id]["mapping"]
    else:
        unknown.append(field_id)
# For unknown fields: LLM suggests mappings constrained by format spec
suggestions = bedrock.invoke_claude(f"Map {unknown} to {target_format} fields")

The aggregation pipeline builds a field-to-mapping lookup from all configs in one query. Known fields get instant matches. Unknown fields get LLM suggestions constrained by the target format's supported fields.

Agentic Payment Repair

Cross-border payments frequently fail because of local clearing requirements, such as missing codes, incorrect scripts, or incomplete institution data. Traditional systems queue these exceptions for manual review, often taking days to resolve.

The Resolution Agent handles these exceptions by using tools that the LLM autonomously selects based on docstrings and system prompts. The pattern guides the LLM toward cheaper, faster options first.

Resolution Agent Tool Selection:

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
@tool
def atlas_search(collection: str, query: str, search_fields: list, fuzzy: bool = True) -> dict:
    """Search MongoDB using Atlas Search. Use fuzzy=False for exact match first."""
    pipeline = [{"$search": {
        "index": f"{collection}_search",
        "text": {"query": query, "path": search_fields, "fuzzy": {"maxEdits": 2} if fuzzy else {}}
    }}]
    results = list(db[collection].aggregate(pipeline))
    return {"found": bool(results), "top_result": results[0] if results else None}
@tool
def vector_search(collection: str, query: str) -> dict:
    """Semantic search using Atlas Vector Search - matches by meaning, not keywords."""
    embedding = voyage_client.embed([query], model="voyage-3").embeddings[0]
    pipeline = [{"$vectorSearch": {
        "index": f"{collection}_vector", "path": "embedding",
        "queryVector": embedding, "numCandidates": 100, "limit": 3
    }}]
    results = list(db[collection].aggregate(pipeline))
    return {"found": bool(results), "top_result": results[0] if results else None}
agent = create_react_agent(model=llm, tools=[atlas_search, vector_search])

The LLM reads tool docstrings and system prompts to determine which tool to call. Atlas Search handles exact lookups and typo-tolerant matching. Vector Search handles semantic classification where keywords don't match but meaning does. To illustrate these capabilities in practice, consider the following payment scenarios:

• Japan (Katakana Transliteration): A payment to Tokyo requires the beneficiary name in Katakana script, but the incoming message contains Latin characters (for example, Heinrich Mueller). The agent first checks the bank_details collection for a known translation. If not found, it uses Atlas Search for fuzzy matching. As a fallback, it calls an LLM to transliterate the name into Katakana (" ハインリッヒ・ミュラー").

• India (IFSC Code Lookup): A payment to India requires an IFSC code to route to the correct branch, but the sender provides only a bank name and city. The agent queries the ifsc_codes collection for an exact match. If the input contains typos (for example, "HDFC Connaught" instead of "Connaught Place"), Atlas Search handles the fuzzy matching to find the correct code.

from langgraph.types import interrupt
def human_review_node(state: dict) -> dict:
    """Pause workflow for human approval."""
    # interrupt() halts execution, returns data to caller, waits for resume
    review_decision = interrupt({
        "original_value": state.get("original_value"),
        "proposed_value": state.get("solution", {}).get("proposed_value"),
        "confidence": state.get("solution", {}).get("confidence")
    })
    # When resumed: {"approved": True/False, "modified_value": "..."}
    return {"approved": review_decision.get("approved"), "human_review": review_decision}
# Workflow: resolution → human_review → execution
workflow.add_edge("resolution", "human_review")  # Always review before execution
app = workflow.compile(checkpointer=MemorySaver())  # Checkpointer persists state

Multi-Rail Extensibility: Blockchain Settlement Example

The canonical JSON model provides a stable internal abstraction for integrating multiple payment rails by normalizing core payment intent—such as parties, assets, amounts, and instructions—into a single, consistent structure. Using the MongoDB polymorphic document model, rail-specific execution details and metadata are stored alongside canonical fields without requiring separate schemas or data models. This keeps protocol complexity out of upstream and downstream systems while enabling different rails to consume the same payment representation.

The demo shows this pattern in action with a Solana integration: payments containing crypto settlement fields are routed to a Solana service that executes real transfers on devnet.

Solana Service Integration:

# Step 1: Convert incoming message to canonical JSON
hop1_result = await converter.convert(
    source_format="pacs.008",  # or MT103, ISO8583, etc.
    target_format="JSON",
    message=raw_message
)
# Step 2: Parse the canonical JSON from conversion result
canonical_json = json.loads(hop1_result['converted_message'])
# Step 3: Check canonical JSON for crypto settlement fields
if canonical_json.get('crypto_blockchain') and canonical_json.get('crypto_receiver_wallet'):
    # Route to Solana
    solana_service.transfer(
        to_pubkey=canonical_json['crypto_receiver_wallet'],
        amount_sol=50000
    )
# step 4: Solana Service Class
class SolanaService:
    def transfer(self, to_pubkey: str, amount_sol: float) -> dict:
        response = self.client.send_raw_transaction(bytes(tx))
        self.client.confirm_transaction(response.value)
        return {
            "signature": str(response.value),
            "explorer_url": f"https://explorer.solana.com/tx/{response.value}?cluster=devnet"
        }

Unlike traditional payment rails, blockchain provides built-in proof of settlement. Every transfer returns a Solana Explorer URL where users verify the transaction on-chain sender, receiver, amount, timestamp, and confirmation status. This transparency isn't available with SWIFT or card networks, where settlement proof requires back-office reconciliation.

The demo runs on Solana devnet, which uses the same protocol as mainnet: Transactions are cryptographically signed, propagated to validators, and permanently recorded.