Open Finance de próxima generación: portabilidad del crédito

Queryable Encryption para la Privacidad del Consentimiento y del Agente

Almacenar registros de consentimiento en texto plano expone campos sensibles a los administradores de bases de datos, a los procesos de copia de seguridad y a posibles incumplimientos. Para prevenir estos riesgos, las regulaciones de Open Finance exigen que las instituciones protejan la identidad del consumidor en cada evento del ciclo de vida del consentimiento: creación, autorización, recuperación de datos y revocación. El cifrado en reposo también protege las configuraciones de los agentes de IA—como los indicaciones del sistema y las definiciones de herramientas—para limitar la exposición de la lógica propietaria.

MongoDB Queryable Encryption resuelve este problema al cifrar los campos sensibles a nivel del driver, garantizando que el servidor nunca vea el texto en claro. Los campos que necesitan filtro se pueden configurar para queries de igualdad. El driver cifra el valor de la query antes de enviarlo, por lo que el servidor coincide con el texto cifrado sin ver nunca el texto en claro. Los campos que solo requieren lectura después de descifrar permanecen cifrados sin soporte de query.

La demostración aplica Queryable Encryption en dos lugares:

1. Colección de consentimientos (encrypted_consents en el backend de Open Finance), con cuatro campos cifrados:

   • Consumer.UserName

   • Consumer.UserId

   • Permissions

   • SourceInstitution.InstitutionName

   El campo Consumer.UserName admite consultas de igualdad para que los servicios puedan listar los consentimientos de un cliente sin que la base de datos vea nunca el nombre de usuario en texto sin formato.

2. Colección de perfiles de agentes (encrypted_agent_profiles en el backend del chatbot), con tres campos cifrados:

   • agent_name (consultable por igualdad)

   • system_prompttool_config

Los mensajes del agente se cargan desde MongoDB encriptada durante la ejecución. Queryable Encryption genera una llave de cifrado de datos diferente para cada campo. Admite AWS Key Management Service (KMS), Azure Key Vault y Google Cloud KMS como proveedores de gestión de claves. El siguiente ejemplo muestra la configuración de la conexión cifrada para el backend de Open Finance:

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)

Las consultas de consentimiento funcionan de manera idéntica a texto sin formato: el controlador gestiona el cifrado y descifrado de manera transparente:

# 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}))

La conexión cifrada amplía el MongoDBConnection estándar, por lo que todos los servicios que insinúan el tipo de la clase base la aceptan sin modificaciones.

Clasificación inteligente de transacciones con MongoDB Atlas búsqueda vectorial

Cuando un consumidor comparte datos externos del banco mediante un consentimiento, los registros de transacciones llegan sin categorías estandarizadas de comerciantes. Las cadenas sin procesar de comerciantes como "Nobu Restaurant" o "UBER TRIP" no llevan metadatos de categoría. Sin clasificación, no se puede realizar el análisis de gastos ni la puntuación de crédito.

La Búsqueda Vectorial Atlas clasifica estas transacciones comparando las descripciones de los comercios con una colección de referencia de 49 Códigos de Categoría de Comerciantes (MCC). La demo utiliza el modelo embedding voyage-finance-2 de Voyage IA—diseñado específicamente para texto financiero—para generar vectores somer 1024- dimensiones tanto para los datos de referencia MCC como para las descripciones de transacciones entrantes.

El siguiente ejemplo muestra el pipeline de clasificación de búsqueda vectorial:

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

Los documentos de referencia de MCC se incrustan en el momento de la creación con input_type="document", mientras que las queries de transacciones usan input_type="query", siguiendo la mejor práctica de incrustación asimétrica de Voyage IA para cargas de trabajo de recuperación.

La clasificación es efímera: los resultados se devuelven al agente pero nunca se almacenan de forma permanente. Los datos bancarios externos se toman prestados mediante consentimiento, no se poseen. El agente de portabilidad utiliza transacciones clasificadas para calcular puntuaciones de gastos, compararlas con puntos de referencia de mejores prácticas y generar ofertas deterministas de portabilidad de préstamos.

Acceso a los datos agente con el servidor MongoDB MCP

Los asesores financieros y los consumidores suelen necesitar respuestas puntuales sobre los datos de las cuentas: "¿Cuál es mi saldo total?", "Mostrar mis últimas 10 transacciones", o "¿Para cuáles productos califico?". Crear endpoints API personalizados para cada posible query no es práctico.

El servidor MongoDB MCP expone las colecciones de MongoDB como herramientas que los agentes LLM pueden invocar directamente. La demo lanza el servidor MCP como un subproceso al iniciar la aplicación, lo conecta a la base de datos leafy_bank en modo de solo lectura y pasa las herramientas resultantes a un agente LangGraph a través de una sesión persistente.

El siguiente ejemplo muestra la integración del servidor MCP:

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]

El agente bancario de Leafy recibe estas herramientas filtradas más una herramienta get_current_user_id que lee el identificador del cliente autenticado a partir de la configuración de LangGraph. Responde a preguntas en lenguaje natural generando consultas MongoDB de manera autónoma—el agente puede realizar las siguientes acciones:

• findpara buscar

• aggregatepara realizar cálculos

• Collection-schema: utilizar para el descubrimiento.

No es necesario contar con un código de herramienta personalizado por colección.

Orquestación multiagente con LangGraph

Los flujos de trabajo de Open Finance abarcan dominios distintos: gestión del consentimiento, análisis financiero y consultas de datos internas del banco. Un solo agente monolítico que maneje los tres necesitaría un amplio conjunto de herramientas y una solicitud para el sistema que aborde preocupaciones conflictivas. Dividir en agentes especializados mantiene cada conjunto de herramientas pequeño y cada prompt enfocado.

Un agente supervisor orquesta a tres especialistas:

• Agente de consentimiento: gestiona los flujos de uso compartido de datos

• Agente de portabilidad: analiza datos externos para ofertas de préstamos

• Leafy Bank Agente: Consulta los datos de Leafy Bank a través del servidor MCP.

LangGraph dirige cada mensaje del cliente al especialista adecuado según la intención, y MongoDB Atlas conserva el estado de la conversación a través de colecciones de puntos de control.

El siguiente ejemplo muestra el enrutamiento de supervisores con salida estructurada:

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))

Los flujos de trabajo regulados —aprobaciones de consentimientos, revisiones de KYC, autorizaciones de pago— requieren puntos de control humanos donde el agente debe hacer una pausa y esperar una decisión antes de continuar. El mecanismo interrupt() de LangGraph maneja este requerimiento serializando el estado completo del grafo en MongoDB y devolviendo una carga útil al usuario que llama. El flujo de trabajo se reanuda cuando se completa el proceso externo:

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)

Las colecciones de puntos de control de MongoDB Atlas conservan todo el estado de la conversación:

• Historial de mensajes

• Consentimientos activos

• Decisiones de enrutamiento

El flujo de trabajo resiste interrupciones que duran segundos (un clic de botón) o incluso horas (una revisión nocturna de cumplimiento). Cada subagente ejecuta un bucle ReAct (razonar → actuar → observar) hasta que genera una respuesta final y luego devuelve el control al agente supervisor para la próxima decisión de enrutamiento.