오픈 뱅킹: MongoDB를 이용한 안전한 데이터 공유

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)

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

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]

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.mongodb import MongoDBSaver
# Each specialist is built from a factory that loads its system prompt
# from a Queryable-Encryption-backed MongoDB collection at startup.
consent_agent = create_consent_agent(prompts["consent_agent"])
financial_advice_agent = create_financial_advice_agent(
    prompts["financial_advice_agent"], mcp_tools
)
supervisor = create_supervisor_node(prompts["supervisor"])
def route_from_supervisor(state: AgentState) -> str:
    # The supervisor writes its routing decision to state["next"]
    return state.get("next", "FINISH")
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("consent_agent", consent_agent)
workflow.add_node("financial_advice_agent", financial_advice_agent)
workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", route_from_supervisor, {
    "consent_agent": "consent_agent",
    "financial_advice_agent": "financial_advice_agent",
    "FINISH": END,
})
workflow.add_edge("consent_agent", "supervisor")
workflow.add_edge("financial_advice_agent", "supervisor")
checkpointer = MongoDBSaver(
    client=db.client,
    db_name=DATABASE_NAME,
    checkpoint_collection_name=CHECKPOINTS_AIO_COLLECTION,
    writes_collection_name=CHECKPOINTS_WRITES_AIO_COLLECTION,
)
graph = workflow.compile(checkpointer=checkpointer)

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)