下一代开放金融：信用可转移性

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

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

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

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)