Next Generation Open Finance: Credit Portability

Queryable Encryption for consent and agent privacy

Storing consent records in plaintext exposes sensitive fields to database administrators, backup processes, and potential breaches. To prevent these risks, Open Finance regulations require institutions to protect consumer identity across every consent life cycle event: creation, authorization, data retrieval, and revocation. Encryption at rest also protects AI agent configurations—such as system prompts and tool definitions—to limit exposure of proprietary logic.

MongoDB Queryable Encryption solves this problem by encrypting sensitive fields at the driver level, ensuring the server never sees plaintext. Fields that need filtering can be configured for equality queries. The driver encrypts the query value before sending it, so the server matches on ciphertext without ever seeing the plaintext. Fields that only require read-after-decrypt remain encrypted without query support.

The demo applies Queryable Encryption in two places:

1. Consent collection (encrypted_consents in the Open Finance backend), with four encrypted fields:

   • Consumer.UserName

   • Consumer.UserId

   • Permissions

   • SourceInstitution.InstitutionName

   The Consumer.UserName field supports equality queries so services can list a customer's consents without the database ever seeing the username in plaintext.

2. Agent profile collection (encrypted_agent_profiles in the chatbot backend), with three encrypted fields:

   • agent_name (equality-queryable)

   • system_prompttool_config

Agent prompts are loaded from encrypted MongoDB at runtime. Queryable Encryption generates a separate Data Encryption Key for each field. It supports AWS Key Management Service (KMS), Azure Key Vault, and Google Cloud KMS as key management providers. The following example shows the encrypted connection setup for the Open Finance backend:

from pymongo import MongoClient
from pymongo.encryption_options import AutoEncryptionOpts
class EncryptedMongoDBConnection(MongoDBConnection):
    """Subclasses the standard connection — services that type-hint
    MongoDBConnection accept it without modification."""
    def __init__(self, uri: str, auto_encryption_opts: AutoEncryptionOpts):
        self.uri = uri
        self.client = MongoClient(self.uri, auto_encryption_opts=auto_encryption_opts)

Consent queries work identically to plaintext—the driver handles encryption and decryption transparently:

# Standard query on a plaintext field — works as usual
consent = consents_collection.find_one({"ConsentId": consent_id})
# Equality query on an encrypted field — same syntax, driver encrypts the filter value
consents = list(consents_collection.find({"Consumer.UserName": user_name}))

The encrypted connection extends the standard MongoDBConnection, so every service that type-hints the base class accepts it without modification.

Intelligent transaction classification with MongoDB Atlas Vector Search

When a consumer shares external bank data through a consent, transaction records arrive without standardized merchant categories. Raw merchant strings such as "Nobu Restaurant" or "UBER TRIP"—carry no category metadata. Without classification, spending analysis and credit scoring cannot function.

Atlas Vector Search classifies these transactions by matching merchant descriptions against a reference collection of 49 Merchant Category Codes (MCC). The demo uses Voyage AI's voyage-finance-2 embedding model—purpose-built for financial text—to generate 1024-dimension vectors for both the MCC reference data and incoming transaction descriptions.

The following example shows the vector search classification pipeline:

class MCCClassificationService:
    def __init__(self, connection, db_name, collection_name):
        self.collection = connection.get_collection(db_name, collection_name)
        self.vo = voyageai.Client()
        self.model = "voyage-finance-2"
    def classify_batch(self, transactions):
        # Build query text from merchant name + description
        query_texts = [
            self._build_query_text(txn.get("merchant_name", ""), txn.get("description", ""))
            for txn in transactions
        ]
        # Single batch embedding call — input_type="query" for search queries
        embed_result = self.vo.embed(query_texts, model=self.model, input_type="query")
        # Vector search each embedding against MCC reference codes
        for txn, embedding in zip(transactions, embed_result.embeddings):
            match = self._vector_search(embedding)
            if match:
                txn.update({
                    "MCC": match["MCC"],
                    "CategoryName": match["CategoryName"],
                    "confidence": round(match["score"], 4),
                })
    def _vector_search(self, query_embedding, num_candidates=20, limit=1):
        pipeline = [
            {"$vectorSearch": {
                "index": "mcc_codes_vector_index",
                "path": "embedding",
                "queryVector": query_embedding,
                "numCandidates": num_candidates, "limit": limit,
            }},
            {"$project": {
                "MCC": 1, "MCCDescription": 1, "CategoryId": 1,
                "CategoryName": 1, "score": {"$meta": "vectorSearchScore"}, "_id": 0,
            }},
        ]
        results = list(self.collection.aggregate(pipeline))
        return results[0] if results else None

MCC reference documents are embedded at seed time with input_type="document", while transaction queries use input_type="query", following Voyage AI's asymmetric embedding best practice for retrieval workloads.

Classification is ephemeral—results are returned to the agent but never persisted. External bank data is borrowed through consent, not owned. The portability agent uses classified transactions to compute spending scores, compare them against best-practice benchmarks, and generate deterministic loan portability offers.

Agentic data access with the MongoDB MCP Server

Financial advisors and consumers often need ad hoc answers about account data: "What's my total balance?", "Show my last 10 transactions", or "Which products do I qualify for?". Building custom API endpoints for every possible query is impractical.

The MongoDB MCP Server exposes MongoDB collections as tools that LLM agents can invoke directly. The demo launches the MCP server as a subprocess at application startup, connects it to the leafy_bank database in read-only mode, and passes the resulting tools to a LangGraph agent through a persistent session.

The following example shows the MCP server integration:

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
mcp_client = MultiServerMCPClient({
    "mongodb": {
        "command": "npx",
        "args": ["-y", "mongodb-mcp-server@latest"],
        "transport": "stdio",
        "env": {
            **os.environ,
            "MDB_MCP_CONNECTION_STRING": LEAFY_BANK_MONGODB_URI,
            "MDB_MCP_READ_ONLY": "true",
            "MDB_MCP_DISABLED_TOOLS": disabled_tools,
        },
    }
})
# Persistent session keeps MongoDB connection state across tool calls
async with mcp_client.session("mongodb") as session:
    all_mcp_tools = await load_mcp_tools(session)
    # Pre-connect so the agent never handles connection strings
    connect_tool = next((t for t in all_mcp_tools if t.name == "connect"), None)
    if connect_tool:
        await connect_tool.ainvoke({"connectionString": LEAFY_BANK_MONGODB_URI})
    # Only expose read/query tools to the agent
    allowed_tools = {"find", "aggregate", "count", "list-collections", "collection-schema"}
    mcp_tools = [t for t in all_mcp_tools if t.name in allowed_tools]

The leafy bank agent receives these filtered tools plus a get_current_user_id tool that reads the authenticated customer's identifier from the LangGraph config. It answers natural language questions by generating MongoDB queries autonomously—the agent can perform the following actions:

• find: use for lookups

• aggregate: use for calculations

• Collection-schema: use for discovery.

No custom tool code is needed per collection.

Multi-agent orchestration with LangGraph

Open Finance workflows span distinct domains—consent management, financial analysis, and internal bank data queries. A single monolithic agent handling all three would need a large toolset and a system prompt covering conflicting concerns. Splitting into specialized agents keeps each toolset small and each prompt focused.

A supervisor agent orchestrates three specialists:

• Consent Agent: Manages data-sharing flows

• Portability Agent: Analyzes external data for loan offers

• Leafy Bank Agent: Queries Leafy Bank data through the MCP server.

LangGraph routes each customer message to the appropriate specialist based on intent, and MongoDB Atlas persists conversation state through checkpoint collections.

The following example shows supervisor routing with structured output:

from langgraph.graph import StateGraph, START, END
class RouterDecision(BaseModel):
    next: Literal["consent_agent", "portability_agent", "internal_data_agent", "FINISH"]
    response: str = ""
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("consent_agent", consent_agent)
workflow.add_node("portability_agent", portability_agent)
workflow.add_node("internal_data_agent", internal_data_agent)
workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", route_from_supervisor, {
    "consent_agent": "consent_agent",
    "portability_agent": "portability_agent",
    "internal_data_agent": "internal_data_agent",
    "FINISH": END,
})
workflow.add_edge("consent_agent", "supervisor")
workflow.add_edge("portability_agent", "supervisor")
workflow.add_edge("internal_data_agent", "supervisor")
graph = workflow.compile(checkpointer=MongoDBSaver(client=db.client, db_name=DATABASE_NAME))

Regulated workflows—consent approvals, KYC reviews, payment authorizations—require human checkpoints where an agent must pause and wait for a decision before proceeding. LangGraph's interrupt() mechanism handles this requirement by serializing the full graph state to MongoDB and returning a payload to the caller. The workflow resumes when the external process completes:

from langgraph.types import interrupt, Command
# Agent pauses, returns review payload to the calling application
review = interrupt({
    "type": "APPROVAL_REQUIRED",
    "details": approval_details,
})
# Application resumes the workflow after the human decision
await agent.ainvoke(Command(resume=decision), config)

MongoDB Atlas checkpoint collections persist the full conversation state:

• Message history

• Active consents

• Routing decisions

The workflow survives interrupts that last seconds (a button click) or hours (an overnight compliance review). Each sub-agent runs a ReAct loop (reason → act → observe) until it produces a final response and then returns control to the supervisor agent for the next routing decision.