基于MCP协议构建银行数据AI助手：bank-mcp项目架构与实现详解

1. 项目概述与核心价值最近在折腾一个挺有意思的项目叫bank-mcp是xProgrammerAlvz大佬开源在 GitHub 上的。简单来说这是一个模型上下文协议Model Context Protocol, MCP的服务器实现专门用来把银行账户数据安全、结构化地喂给像 Claude、Cursor 这类 AI 助手。想象一下你正在用 Claude 分析月度开支或者让 Cursor 帮你写一段基于交易记录的预算分析代码你不再需要手动导出 CSV、复制粘贴一堆数字而是直接问 AI“我这个月在餐饮上花了多少钱”或者“帮我找出所有可疑的重复扣款”。bank-mcp干的就是这个“桥梁”的活儿它让 AI 助手能“看见”并理解你的财务数据。这个项目吸引我的点在于它精准地踩在了两个趋势的交汇处一是个人财务管理的自动化和智能化需求日益增长二是 AI 智能体Agent需要更丰富、更可靠的工具来扩展其能力边界。它不是一个完整的金融 App而是一个基础设施组件。通过标准化的 MCP 协议它将复杂的银行 API比如欧洲的 PSD2、澳洲的 Up Bank API、或通用的 Teller API封装成一套简单的、AI 友好的工具函数比如list_accounts列出账户、get_transactions获取交易。这对于开发者、数据分析师或者任何想用 AI 来增强自己财务工作流的人来说都是一个非常实用的工具。2. 核心架构与设计思路拆解2.1 为什么是 MCP要理解bank-mcp首先得搞懂 MCP 是什么。你可以把它想象成 AI 世界的“USB 标准”。在没有 MCP 之前每个 AI 应用如 Claude Desktop、Cursor想连接外部数据源如数据库、银行、文件系统都需要开发专属的、硬编码的插件或集成过程繁琐且不通用。MCP 由 Anthropic 提出旨在定义一套标准协议让任何 MCP 服务器提供数据或工具都能被任何兼容 MCP 的客户端AI 应用即插即用。bank-mcp选择基于 MCP 构建是极具远见的设计。这意味着一次开发多处使用只要 Claude、Cursor、未来其他支持 MCP 的 AI 应用都能直接接入无需为每个应用单独适配。关注点分离bank-mcp只需专注于最难的部分——安全、稳定、合规地与各类银行 API 打交道并将数据格式化为 AI 易理解的 JSON 结构。AI 应用则专注于如何利用这些数据生成洞察或执行操作。安全性提升MCP 协议本身设计了资源数据和工具操作的权限模型。bank-mcp可以声明它提供“只读”的交易数据而不会暴露转账等高风险操作这为财务数据这种敏感信息增加了一层安全抽象。2.2 项目架构与核心组件浏览项目代码和文档可以梳理出其核心架构分为三层协议层MCP Server这是项目的基石。它使用官方modelcontextprotocol/sdk或其他 MCP SDK 实现了一个标准的 MCP 服务器。这个服务器的核心工作是向客户端AI宣告“我这里有哪些工具Tools和资源Resources可用”。对于bank-mcp工具可能就是query_transactions资源可能就是account://savings/balance。适配器层Bank Adapters这是项目的心脏也是复杂度所在。银行 API 千差万别有 RESTful 的有 OAuth 2.0 认证的有数据格式迥异的。bank-mcp需要为每一个支持的银行如 Up Bank、某支持 PSD2 的欧洲银行编写一个适配器Adapter。每个适配器都负责认证处理安全地存储和管理访问令牌Access Token、刷新令牌Refresh Token。API 调用封装将 MCP 工具调用的通用参数转换为目标银行 API 的具体 HTTP 请求。数据标准化将银行返回的原始数据可能是 XML、各种嵌套的 JSON清洗、转换为一套内部统一的、结构化的数据模型例如一个标准的Transaction对象包含id,date,amount,description,category等字段。配置与运行层用户需要通过配置文件如config.yaml或环境变量来指定使用哪个银行适配器并提供必要的认证信息如 API Key、Client ID/Secret。项目通常提供一个主入口文件来启动这个集成了具体适配器的 MCP 服务器。注意从关键词看项目可能涉及cloudflare-workers。这暗示了一种非常巧妙的部署方式将bank-mcp服务器部署为 Cloudflare Worker。这样做的好处是极致轻量、全球低延迟、以及天然的边缘计算特性。AI 客户端如 Claude Desktop可以通过 HTTP 或 SSE 与这个 Worker 通信无需在本地运行一个 Python 进程更加便捷和安全。2.3 技术栈选型解析关键词透露了项目的技术倾向python很可能是实现 MCP 服务器和核心逻辑的主力语言因其在数据处理、API 集成和快速开发方面的丰富生态。cloudflare-workers如前所述作为无服务器部署平台可能是首选。它支持 JavaScript/TypeScript 或 Python通过 Pyodide项目可能需要考虑跨语言实现或选择其一。psd2,teller,up-bank这些是具体的银行 API 或标准代表了项目计划支持或已支持的数据源。PSD2 是欧洲的开放银行法规Teller 是一个聚合多家银行 API 的服务Up Bank 是澳洲一家提供优秀 API 的数字银行。chembl-web-client,bioinformatics,biology这几个关键词非常有趣它们似乎超出了“银行”的范畴。这强烈暗示了bank-mcp项目的雄心不限于金融领域。它可能是一个通用 MCP 服务器框架或示例bank只是其中一个应用实例。作者可能用银行数据这个高价值、结构化的场景来验证框架而其架构完全可以复用于接入生物信息学数据库如 ChEMBL、科研数据等。这体现了“关注点分离”架构的优势——核心的 MCP 协议层和适配器模式是通用的。3. 核心细节解析与实操要点3.1 MCP 工具与资源的设计在 MCP 中AI 主要通过两种方式与服务器交互调用工具Tools类似函数调用。例如AI 可以调用get_transactions工具传入account_id和start_date参数服务器执行后返回交易列表。读取资源Resources类似读取文件或 URL。每个资源有一个唯一的 URI如bank://up/accounts/checking/balance。AI 可以请求这个 URI 的内容服务器返回最新的余额数据。对于bank-mcp工具和资源的设计至关重要工具设计示例list_accounts(): 返回所有账户的概要信息。get_transactions(account_id, start_date, end_date, limit): 获取指定账户、时间范围内的交易记录。categorize_transaction(transaction_id, category): 如果银行 API 支持为某笔交易打标签。analyze_spending(timeframe, category):这是一个高阶工具。它可能不是直接调用银行 API而是在服务器端聚合计算交易数据后返回 AI 可以直接引用的分析结果如“餐饮类目本月总额为 $XXX”这比把原始交易记录全扔给 AI 再让它分析更高效、成本更低。资源设计示例bank://{adapter_name}/accounts 列出所有账户的静态资源。bank://{adapter_name}/accounts/{account_id}/transactions?start2024-01-01 一个动态资源URI 本身包含了查询参数。实操心得在设计工具时要平衡“灵活性”和“效率”。给 AI 过于原始的工具如“获取所有交易”会导致上下文令牌Token的浪费。提供一些聚合计算后的工具如“获取本月支出摘要”能显著提升 AI 响应的速度和准确性。这需要服务器端实现一定的业务逻辑。3.2 安全与认证处理这是金融数据集成中最关键、最容易踩坑的部分。bank-mcp必须妥善处理凭证存储绝不能将银行 API 的密钥、令牌硬编码在代码中。必须通过环境变量或加密的配置文件来管理。在 Cloudflare Worker 环境下可以使用 Worker 的 Secrets 或环境变量功能。OAuth 2.0 流许多银行 API如 PSD2使用 OAuth 2.0。bank-mcp需要实现一个“授权回调端点”。流程是用户首次配置时服务器提供一个授权 URL 给用户。用户访问该 URL 并在银行页面登录授权。银行回调到bank-mcp预设的端点如https://your-worker.your-subdomain.workers.dev/auth/callback并携带授权码。bank-mcp用授权码换取长期的访问令牌和刷新令牌并安全存储。后续请求都使用访问令牌。令牌刷新访问令牌会过期。适配器必须实现自动刷新令牌的逻辑在令牌失效前或收到 401 错误时使用刷新令牌获取新的访问令牌确保服务不间断。数据脱敏与权限在返回给 AI 的数据中应考虑对极端敏感信息如银行卡号全文进行脱敏。同时在 MCP 服务器声明中明确只提供“只读”工具从协议层面杜绝误操作风险。3.3 错误处理与健壮性银行 API 可能不稳定网络可能波动用户数据可能异常。一个健壮的bank-mcp服务器必须包含重试机制对于网络超时或银行 API 的瞬时错误5xx实现带指数退避的智能重试。优雅降级如果某个高级工具如analyze_spending依赖的第三方分析服务失败应能回退到返回原始数据并告知 AI“分析功能暂不可用以下是原始交易数据”。清晰的错误信息将银行 API 返回的晦涩错误码转换为对 AI和最终用户友好的自然语言描述。例如将INSUFFICIENT_PERMISSIONS转换为“当前凭证权限不足无法读取交易历史请检查授权范围”。速率限制遵守银行 API 的调用频率限制并在服务器内部实现队列或限流避免触发风控。4. 实操过程与核心环节实现假设我们要为Up Bank API实现一个适配器并部署到 Cloudflare Workers。以下是核心步骤和代码要点。4.1 环境准备与项目初始化首先确保你有一个 Up Bank 账户并已在其开发者门户创建应用获取CLIENT_ID和CLIENT_SECRET。我们将使用 Python 的mcp库和httpx库。# 创建一个新项目目录 mkdir bank-mcp-up-adapter cd bank-mcp-up-adapter python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install mcp httpx python-dotenv创建关键文件. ├── main.py # MCP 服务器主入口 ├── up_adapter.py # Up Bank 适配器实现 ├── config.py # 配置管理 ├── .env # 存储敏感环境变量不要提交到Git └── requirements.txt4.2 实现 Up Bank 适配器核心up_adapter.py的核心是定义一个类封装所有与 Up Bank API 的交互。import os import httpx from typing import List, Dict, Any, Optional from datetime import datetime, date import json from config import settings class UpBankAdapter: BASE_URL https://api.up.com.au/api/v1 def __init__(self, access_token: str): self.client httpx.AsyncClient( headers{ Authorization: fBearer {access_token}, Content-Type: application/json }, timeout30.0 ) async def list_accounts(self) - List[Dict[str, Any]]: 获取所有账户列表 try: response await self.client.get(f{self.BASE_URL}/accounts) response.raise_for_status() data response.json() # 标准化账户信息 accounts [] for acc in data.get(data, []): attrs acc[attributes] accounts.append({ id: acc[id], name: attrs[displayName], type: attrs[accountType], # e.g., SAVER, TRANSACTIONAL balance: float(attrs[balance][value]), currency: attrs[balance][currencyCode] }) return accounts except httpx.HTTPStatusError as e: # 处理特定错误如令牌过期 if e.response.status_code 401: raise Exception(认证失败访问令牌可能已过期请重新授权。) raise Exception(f获取账户列表失败: {e}) async def get_transactions(self, account_id: Optional[str] None, since: Optional[date] None, limit: int 50) - List[Dict[str, Any]]: 获取交易记录。如果不指定 account_id则获取所有账户的交易。 params {page[size]: limit} if since: params[filter[since]] since.isoformat() endpoint f{self.BASE_URL}/transactions if account_id: endpoint f{self.BASE_URL}/accounts/{account_id}/transactions try: response await self.client.get(endpoint, paramsparams) response.raise_for_status() data response.json() transactions [] for tx in data.get(data, []): attrs tx[attributes] # 标准化交易信息这是给AI理解的关键 std_tx { id: tx[id], date: attrs[createdAt], # ISO 格式时间 description: attrs[description], raw_text: attrs.get(rawText, ), # 银行原始描述有时更有用 amount: float(attrs[amount][value]), currency: attrs[amount][currencyCode], category: attrs.get(category, 未分类), status: attrs[status] # SETTLED 或 HELD } # 处理对方信息如果存在 if holdInfo in attrs and attrs[holdInfo] is not None: std_tx[counterparty] attrs[holdInfo].get(counterPartyName) transactions.append(std_tx) return transactions except httpx.HTTPStatusError as e: raise Exception(f获取交易记录失败: {e}) async def close(self): await self.client.aclose()关键点解析适配器的核心工作是数据标准化。这里我们将 Up Bank 返回的复杂嵌套 JSON转换为了一个扁平、字段名清晰的字典列表。raw_text字段的保留很重要因为银行的原始描述有时包含了商户代码而description是处理过的两者结合 AI 更能准确理解交易性质。4.3 构建 MCP 服务器在main.py中我们将适配器与 MCP 协议绑定。import asyncio from mcp import Server, Tool import json from typing import Any from up_adapter import UpBankAdapter from config import settings # 初始化全局适配器实例实际生产环境需考虑多用户和生命周期管理 _adapter: UpBankAdapter None async def initialize_adapter(): 初始化银行适配器。在实际应用中这里需要根据用户配置动态创建。 global _adapter # 从安全存储如环境变量、数据库中读取 access_token access_token settings.UP_ACCESS_TOKEN if not access_token: raise ValueError(UP_ACCESS_TOKEN 未配置。请完成 OAuth 授权流程。) _adapter UpBankAdapter(access_token) async def handle_list_accounts() - str: MCP 工具列出账户 accounts await _adapter.list_accounts() # 返回给 AI 的应该是清晰、自然的文本描述而非纯JSON summary f您共有 {len(accounts)} 个账户\n for acc in accounts: balance_sign if acc[balance] 0 else summary f- **{acc[name]}** ({acc[type]}): {balance_sign}{acc[balance]:.2f} {acc[currency]}\n return summary async def handle_get_transactions(account_name: str None, days: int 30) - str: MCP 工具获取近期交易 from datetime import datetime, timedelta since_date datetime.now().date() - timedelta(daysdays) # 根据账户名找到账户ID这里简化处理实际可能需要更精确的匹配 accounts await _adapter.list_accounts() account_id None if account_name: for acc in accounts: if account_name.lower() in acc[name].lower(): account_id acc[id] break if not account_id: return f未找到名称为 {account_name} 的账户。 transactions await _adapter.get_transactions(account_idaccount_id, sincesince_date) if not transactions: return f在过去 {days} 天内未找到交易记录。 # 格式化输出便于AI阅读和分析 output f共找到 {len(transactions)} 笔交易\n\n for tx in transactions: amount tx[amount] sign 收入 if amount 0 else 支出 output f- **{tx[date][:10]}** | {tx[description]} | **{abs(amount):.2f}** {tx[currency]} ({sign}, 类别: {tx[category]})\n # 附加一个简单的统计这对AI后续分析非常有帮助 total_spent sum(tx[amount] for tx in transactions if tx[amount] 0) total_income sum(tx[amount] for tx in transactions if tx[amount] 0) output f\n**简要统计**: 总支出 {abs(total_spent):.2f} {tx[currency]}, 总收入 {total_income:.2f} {tx[currency]}。 return output async def main(): await initialize_adapter() # 创建 MCP 服务器 server Server(bank-mcp-server) # 注册工具Tools server.list_tools() async def list_tools(): return [ Tool( namelist_accounts, description列出用户的所有银行账户及其当前余额。, inputSchema{ type: object, properties: {} } ), Tool( nameget_recent_transactions, description获取近期交易记录。可以指定账户名称可选和时间范围默认最近30天。, inputSchema{ type: object, properties: { account_name: { type: string, description: 账户名称如 Savings。如果不提供则返回所有账户的交易。 }, days: { type: integer, description: 查询过去多少天的交易默认30天。, default: 30 } } } ) ] # 绑定工具处理函数 server.call_tool() async def call_tool(name: str, arguments: dict) - list[dict[str, Any]]: if name list_accounts: result await handle_list_accounts() return [{type: text, text: result}] elif name get_recent_transactions: account_name arguments.get(account_name) days arguments.get(days, 30) result await handle_get_transactions(account_name, days) return [{type: text, text: result}] else: raise ValueError(f未知工具: {name}) # 运行服务器标准输入/输出 await server.run() if __name__ __main__: asyncio.run(main())4.4 部署到 Cloudflare Workers为了能让 Claude Desktop 远程连接我们需要将服务器部署到云端。Cloudflare Workers 是一个极佳选择。安装 Wrangler CLInpm install -g wrangler初始化 Workerwrangler init bank-mcp-worker选择“Python”模板。调整代码结构Workers 使用index.py作为入口点并需要兼容其无服务器环境。我们需要将异步 HTTP 服务器改为 Workers 的fetch事件处理模式。幸运的是mcp库通常支持通过StdioServer或HTTPServer运行。我们可以创建一个HTTPServer。创建index.py:from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import asyncio from .up_adapter import UpBankAdapter import os # 注意Workers 环境下的全局变量和持久化需要特别处理 # 这里简化处理实际中需要利用 Workers 的 KV 或 D1 来存储用户令牌 adapter None async def handle_request(request): # 这里需要实现 MCP over HTTP 的协议处理 # 这是一个简化示例实际需按照 MCP HTTP 规范解析请求体 global adapter if adapter is None: adapter UpBankAdapter(os.environ[UP_ACCESS_TOKEN]) # 解析 request调用对应的工具函数... # 返回 JSON 响应... pass async def on_fetch(request, env, ctx): return await handle_request(request)配置wrangler.toml并设置环境变量UP_ACCESS_TOKEN。部署wrangler deploy。部署成功后你会获得一个https://your-worker.your-subdomain.workers.dev的地址。在 Claude Desktop 的 MCP 设置中就可以添加这个 HTTP 服务器地址实现远程连接。5. 常见问题与排查技巧实录在实际搭建和使用bank-mcp的过程中我遇到了不少坑。这里记录下最典型的几个问题和解决方法。5.1 认证失败与令牌管理问题运行后AI 调用工具总是返回“认证失败”或“401 Unauthorized”。排查思路检查环境变量首先确认UP_ACCESS_TOKEN等环境变量是否正确设置且已加载。在本地可以用print(os.environ.get(UP_ACCESS_TOKEN))调试在 Cloudflare Workers使用wrangler secret命令设置。令牌是否过期访问令牌通常有较短的有效期几小时到几天。如果使用的是长期有效的 Personal Access Token部分银行提供请确认其权限范围Scopes是否包含你需要的操作如transactions:read。如果使用的是 OAuth 临时令牌需要实现刷新逻辑。实现令牌自动刷新这是生产环境必备。在适配器中加入检查令牌过期的逻辑并在接近过期时自动使用刷新令牌获取新令牌。代码框架如下class UpBankAdapterWithRefresh(UpBankAdapter): def __init__(self, access_token, refresh_token, client_id, client_secret, token_expiry): super().__init__(access_token) self.refresh_token refresh_token self.client_id client_id self.client_secret client_secret self.token_expiry token_expiry self.token_url https://api.up.com.au/oauth/token async def _ensure_valid_token(self): 检查并刷新令牌 if datetime.now() self.token_expiry - timedelta(minutes5): await self._refresh_access_token() async def _refresh_access_token(self): async with httpx.AsyncClient() as client: resp await client.post(self.token_url, data{ grant_type: refresh_token, refresh_token: self.refresh_token, client_id: self.client_id, client_secret: self.client_secret }) resp.raise_for_status() new_tokens resp.json() # 更新内存中的令牌 self.client.headers[Authorization] fBearer {new_tokens[access_token]} # **重要将新令牌持久化到安全存储如数据库** # await save_tokens_to_db(new_tokens[access_token], new_tokens[refresh_token]) self.token_expiry datetime.now() timedelta(secondsnew_tokens[expires_in])5.2 AI 理解数据格式不佳问题AI 拿到了交易数据但做出的分析很肤浅或错误比如无法正确分类“美团”属于餐饮。解决方案数据增强在适配器返回数据前可以做一层简单的本地增强。例如维护一个关键词到类别的映射字典{美团: 餐饮, 星巴克: 餐饮, 中石化: 交通燃油}在description或raw_text匹配时自动补充或修正category字段。提供更丰富的上下文不要只返回干巴巴的交易列表。在handle_get_transactions函数中像我们之前做的那样附带一个简要的统计摘要总支出、总收入、按类别粗略汇总。这相当于给了 AI 一个“分析起点”它能基于此做更深入的推理。设计更智能的工具与其让 AI 分析原始数据不如在服务器端实现一些分析工具。例如增加一个get_spending_trend(category, period)工具服务器端用pandas计算好环比、同比数据再以自然语言形式返回给 AI。这样 AI 的负担小回答更精准。5.3 性能与速率限制问题查询数月交易时响应慢或频繁被银行 API 限流。优化技巧分页与缓存银行 API 通常支持分页。首次查询时可以按需获取更多页面。对于不常变动的数据如账户列表可以在内存或 Workers KV 中设置短期缓存如 5 分钟。增量同步记录最后一次同步交易的时间戳。下次查询时只拉取该时间戳之后的交易大幅减少数据量。这需要在服务器端为每个用户维护一个同步状态。遵守速率限制仔细阅读银行 API 文档的速率限制部分。在适配器中实现一个简单的令牌桶Token Bucket算法控制请求频率避免触发 HTTP 429 错误。5.4 多用户支持与数据隔离问题项目初始设计可能是单用户的。如何支持多个 AI 用户连接各自的银行账户架构演进无状态服务器 外部存储MCP 服务器本身不存储任何用户令牌。当 AI 客户端连接时通过某种安全方式例如连接时传递一个用户 ID 或会话令牌来标识用户。令牌管理服务需要一个独立的、安全的服务或数据库如 Cloudflare D1、Supabase来存储用户与其银行凭证的映射关系。当 MCP 工具被调用时根据当前会话的用户 ID去查询对应的访问令牌再初始化对应的适配器。连接池与生命周期为每个活跃用户会话维护一个适配器实例并在会话结束时清理。注意在无服务器环境下需要妥善处理冷启动带来的延迟。这个从单用户到多用户的改造是bank-mcp从一个个人脚本演变为一个真正可服务化产品的关键一步复杂度会显著上升但架构清晰分离后是完全可以实现的。最后我想说的是bank-mcp这类项目代表了 AI 应用开发的一个新范式将专业能力封装成标准的、可组合的“工具”。它不仅仅是一个银行查询工具其适配器模式和 MCP 协议的应用可以扩展到任何需要让 AI 与专业系统交互的场景比如你提到的生物信息学ChEMBL、物联网、内部业务系统等。动手实现一个这样的服务器是理解 AI Agent 底层工作逻辑的绝佳方式。