概览

personal-finance-mcp 是一个自托管的只读 MCP（Model Context Protocol）服务器，通过 Plaid API 连接用户的银行账户、信用卡、贷款和 brokerage 账户，使 MCP 客户端（如 Claude Code）能够以自然语言查询个人财务数据。 资料来源：README.md

该项目具有以下核心特性：

• 只读访问：所有工具均为只读操作，不会执行任何资金转移或账户修改操作
• 自托管模式：部署在用户自己的基础设施上，无第三方数据聚合商介入
• MCP 协议兼容：符合 Model Context Protocol 标准，可与任何兼容的 MCP 客户端集成
• 单租户设计：每个部署实例仅供一人使用 资料来源：README.md

graph TD
    A[MCP 客户端<br/>Claude Code] --> B[personal-finance-mcp<br/>:8000/mcp]
    B --> C[Plaid API]
    C --> D[银行/信用卡]
    C --> E[贷款账户]
    C --> F[投资账户]

核心工具

项目提供 9 个只读 MCP 工具，覆盖账户余额、交易记录、投资组合和机构健康状态等场景。 资料来源：server.py:1-200

工具使用示例

用户可以用自然语言询问财务问题，MCP 客户端会自动调用相应工具：

用户：上个月我在 groceries 上花了多少钱？
Claude：调用 get_transactions（日期范围：上月1日-末日）
       → 返回 $487.23，14 笔交易，主要商户：
         Whole Foods ($198)、Trader Joe's ($156)、Safeway ($89)

用户：我还在支付哪些订阅？
Claude：调用 get_recurring_transactions
       → 返回所有检测到的周期性支出

资料来源：README.md

系统架构

组件结构

graph LR
    subgraph "部署层"
        A[.env 配置] --> B[server.py<br/>MCP Server]
        C[link_helper.py<br/>Plaid Link] --> A
    end
    
    subgraph "Plaid 集成层"
        B --> D[plaid_client.py]
        D --> E[Plaid API<br/>accounts/balance<br/>transactions<br/>investments<br/>liabilities]
    end
    
    subgraph "MCP 客户端"
        F[Claude Code<br/>或其他 MCP 客户端]
    end
    
    F -->|"streamable-http<br/>:8000/mcp"| B

核心模块

Token 管理机制

系统通过环境变量加载多个 Plaid Access Token，支持同时链接多个银行机构。 资料来源：plaid_client.py:50-60

# 环境变量命名规范: PLAID_TOKEN_{机构名称大写}
PLAID_TOKEN_CHASE=access-prod-xxx...
PLAID_TOKEN_WELLSFARGO=access-prod-yyy...
PLAID_TOKEN_VANGUARD=access-prod-zzz...

Token 加载逻辑使用 SecretStr 包装，确保在日志和错误信息中不会明文泄露。 资料来源：plaid_client.py:25-45

健康检查缓存

每个 Item（银行链接）都有独立的健康状态缓存，缓存有效期为 300 秒。 资料来源：plaid_client.py:140-160

graph TD
    A[请求健康状态] --> B{缓存有效?}
    B -->|是| C[返回缓存的 ItemHealth]
    B -->|否| D[调用 Plaid /item/get]
    D --> E[映射错误码到 HealthStatus]
    E --> F[缓存结果<br/>TTL=300秒]
    F --> C

支持的健康状态包括：healthy、re_auth_required、pending_expiration、item_locked、no_accounts、unknown_error。 资料来源：plaid_client.py:100-115

安全模型

只读保证

所有 MCP 工具均声明 readOnlyHint: True，且实现中仅调用 Plaid 的 GET 类 API：

• /accounts/get、/accounts/balance/get
• /transactions/get、/transactions/recurring/get
• /investments/holdings/get、/investments/transactions/get
• /liabilities/get
• /item/get（仅用于健康检查） 资料来源：server.py:50-180

单租户原则

项目明确声明单租户设计，每个部署实例仅服务于一人。README 强调：不要将部署的 MCP 端点公开暴露，应使用 OAuth 2.1、Cloudflare Access 或私有网络进行访问控制。 资料来源：README.md

Token 安全

• Access Token 仅存储在环境变量中，不写入代码或配置文件
• SecretStr 类防止 Token 在日志输出中泄露
• 链接过程中产生的 Token 需手动添加到 .env，不得提交到版本控制 资料来源：plaid_client.py:25-35

部署方式

本地开发部署

# 1. 克隆仓库并安装依赖
git clone https://github.com/JosueM1109/personal-finance-mcp.git
cd personal-finance-mcp
python3.11 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# 2. 配置环境变量
cp .env.example .env
# 编辑 .env，填入 PLAID_CLIENT_ID 和 PLAID_SECRET

# 3. 链接银行账户
uvicorn link_helper:app --port 8765
# 打开 http://localhost:8765 完成 Plaid Link

# 4. 启动 MCP 服务器
python server.py   # 监听 http://localhost:8000/mcp

# 5. 配置 Claude Code
claude mcp add --transport http personal-finance http://localhost:8000/mcp

资料来源：README.md

Docker 部署

docker build -t personal-finance-mcp .
docker run --rm -p 8000:8000 --env-file .env personal-finance-mcp

Docker 镜像基于 server.json 中的 OCI 配置打包，环境变量包括 PLAID_CLIENT_ID、PLAID_SECRET、PLAID_ENV（默认 production）和 PORT（默认 8000）。 资料来源：server.json:20-45

云平台部署

项目文档推荐了 Prefect Horizon 部署方案（零持续成本），完整教程见 docs/DEPLOYMENT.md。其他支持 Python 的平台（Fly.io、Railway、Raspberry Pi + Tailscale、VPS）也可运行，核心要求是：

• 暴露 /mcp 端点通过 HTTPS 访问
• 使用环境变量配置 Plaid 凭证
• 配置访问认证机制 资料来源：README.md

技术栈

运行时依赖

开发依赖

运行时要求

• Python 版本：3.11+
• Plaid 账户：需要 Plaid 账户（免费 Trial 计划支持 10 个 Items）
• 已启用产品：Transactions、Liabilities、Investments 资料来源：requirements.txt、README.md

错误处理

Plaid 错误映射

Plaid API 错误被统一映射为结构化响应，包含错误码、消息、trace_id 和机构名称。 资料来源：plaid_client.py:60-85

{
    "error": {
        "code": "ITEM_LOGIN_REQUIRED",
        "message": "the login details of this item have changed",
        "trace_id": "uuid-xxx",
        "institution": "Chase"
    }
}

工具响应结构

所有工具返回统一的响应格式，包含主数据字段和 warnings 数组：

{
    "accounts": [...],      # 或 transactions, holdings 等
    "warnings": [
        {
            "institution": "Chase",
            "status": "re_auth_required",
            "reason": "ITEM_LOGIN_REQUIRED"
        }
    ]
}

未健康的 Item（需要重新认证等）会跳过主数据查询，仅在 warnings 中报告状态，避免因单一机构故障导致整个请求失败。 资料来源：server.py:80-120

版本信息

v0.1.0 包含内容：

• 9 个只读工具
• 本地 Plaid Link 助手（一次性 Token 配置）
• Docker 支持
• OCI 包配置（ghcr.io/josuem1109/personal-finance-mcp:1.0.0） 资料来源：server.json:5-10

相关文档

• ARCHITECTURE.md - 架构深入解析（包含为何选择 /transactions/get 而非 /transactions/sync）
• DEPLOYMENT.md - Prefect Horizon 部署教程
• CONTRIBUTING.md - 贡献指南（项目范围限定为只读、单租户、Plaid 后端）

前置要求

在开始安装之前，请确保您的系统满足以下要求。

资料来源：README.md:1

必需的 Python 依赖

项目依赖通过 requirements.txt 管理，主要包含以下核心包：

资料来源：requirements.txt:1-6

系统架构概览

personal-finance-mcp 采用模块化架构，包含三个核心组件。

graph TD
    subgraph Plaid_API["Plaid API"]
        A[银行/券商数据]
    end
    
    subgraph MCP_Server["personal-finance-mcp"]
        B[server.py<br/>MCP 工具入口]
        C[plaid_client.py<br/>API 封装层]
        D[link_helper.py<br/>Token 配置工具]
    end
    
    subgraph MCP_Client["MCP 客户端"]
        E[Claude Code<br/>或其他客户端]
    end
    
    A --> C
    C --> B
    B --> E
    D -.->|"配置 access_token"| C

• server.py：定义 9 个只读 MCP 工具，处理来自客户端的请求
• plaid_client.py：封装 Plaid API 调用，管理 access_token 和健康检查
• link_helper.py：一次性工具，用于生成和配置银行链接 access_token

资料来源：README.md:19-27

第一步：Plaid 账户配置

1.1 注册 Plaid 账户

1. 访问 dashboard.plaid.com/signup 注册账户
2. 选择 Trial 计划（免费，包含 10 个 Items）

1.2 启用产品

在 Team Settings → Products 中启用以下产品：

• Transactions：交易记录
• Liabilities：负债信息（信用卡、学生贷款、房贷等）
• Investments：投资数据

1.3 获取 API 凭证

在 Team Settings → API 页面复制以下凭证：

资料来源：README.md:35-42

第二步：安装与配置

2.1 克隆仓库并安装依赖

git clone https://github.com/JosueM1109/personal-finance-mcp.git
cd personal-finance-mcp
python3.11 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

2.2 配置环境变量

cp .env.example .env

编辑 .env 文件，填入从 Plaid Dashboard 获取的凭证：

PLAID_CLIENT_ID=your_client_id_here
PLAID_SECRET=your_secret_here
PLAID_ENV=production
PORT=8000

资料来源：README.md:44-51

2.3 验证安装

运行测试套件确保依赖正确安装：

pytest -v

资料来源：README.md:51

第三步：绑定银行账户

每个需要连接的银行都需要单独绑定一次。link_helper.py 提供本地 Plaid Link 界面完成此操作。

3.1 启动链接助手

uvicorn link_helper:app --port 8765

服务启动后会显示：

INFO:     Uvicorn running on http://localhost:8765

3.2 完成银行绑定

1. 在浏览器中打开 http://localhost:8765
2. 点击 Link a bank 按钮
3. 在弹出的 Plaid Link 窗口中搜索并选择您的银行
4. 输入网银登录凭据完成认证

3.3 获取 Access Token

绑定成功后，终端会输出如下格式的信息：

============================================================
Institution: Chase Bank
Item ID:     item_xxx_xxx
Add this to your .env (local) and Horizon env (prod):
  PLAID_TOKEN_CHASE=access-prod-xxx...
Do NOT commit this line.
============================================================

将 PLAID_TOKEN_银行名称=access_token 这一行添加到 .env 文件中。

注意：link_helper.py 拒绝在 HORIZON=1 环境变量设置时运行，以确保生产环境安全。

资料来源：link_helper.py:88-97

3.4 支持的产品

首次链接时默认启用 Transactions 产品。可选产品包括：

资料来源：link_helper.py:36-41

第四步：启动 MCP 服务器

4.1 本地运行

python server.py

服务器默认在 http://localhost:8000/mcp 提供服务。

4.2 Docker 部署

docker build -t personal-finance-mcp .
docker run --rm -p 8000:8000 --env-file .env personal-finance-mcp

资料来源：README.md:71-73

4.3 验证服务

检查服务器健康状态：

curl http://localhost:8000/health

第五步：配置 MCP 客户端

以 Claude Code 为例，将 MCP 服务器添加到客户端：

claude mcp add --transport http personal-finance http://localhost:8000/mcp

验证连接是否成功，尝试执行：

list my accounts

资料来源：README.md:75-82

可用的 MCP 工具

服务器提供 9 个只读工具，可在 MCP 客户端中调用：

资料来源：README.md:21-27

常见问题

Q1: 银行连接显示 "re-auth required" 怎么办？

使用 link_helper.py 重新链接该银行，命令如下：

# 设置更新令牌
PLAID_TOKEN_UPDATE=access-prod-xxx... uvicorn link_helper:app --port 8765

然后在浏览器中访问 http://localhost:8765 进行重新认证。

Q2: 交易记录只能查询多久的数据？

Plaid API 限制最多查询约 2 年的历史数据。如果请求范围超出限制，系统会自动裁剪并返回 WINDOW_CLIPPED 警告。

Q3: 如何查看银行连接状态？

调用 get_institutions_status 工具，该工具会检查每个 Item 的健康状态：

• healthy：正常
• re_auth_required：需要重新登录
• pending_expiration：即将过期
• item_locked：账户被锁定

Q4: 可以在生产环境使用吗？

重要安全提醒：MCP 端点暴露在公网会泄露所有链接账户的财务数据。建议使用以下方式保护：

• OAuth 2.1 认证
• Cloudflare Access
• 仅在私有网络暴露

资料来源：README.md:74-78

下一步

• 查阅 ARCHITECTURE.md 了解系统架构设计细节
• 阅读 DEPLOYMENT.md 获取生产环境部署指南
• 参考 CONTRIBUTING.md 参与项目贡献

概述

personal-finance-mcp 是一个自托管的只读 MCP（Model Context Protocol）服务器，通过 Plaid API 将用户的银行账户、信用卡、贷款和经纪账户连接到 MCP 客户端。该项目采用纯本地部署模式，所有数据仅在用户本地环境流转，不依赖任何第三方聚合服务（如 Monarch、Mint 等）。

核心设计原则：

• 只读访问：仅暴露查询类工具，无写入操作
• 单租户架构：数据仅限本地访问，无共享服务端点
• 最小化依赖：仅依赖 Plaid 官方 Python SDK

资料来源：README.md

概述

Plaid 集成机制是 personal-finance-mcp 项目的核心模块，负责将用户的银行、信用卡、贷款和经纪账户通过 Plaid API 连接到 MCP 服务器。该机制完全采用只读模式，确保用户财务数据仅被查询而不被修改。

核心设计目标：

• 本地化密钥管理：所有 access token 存储在本地环境变量中，不经过第三方
• 安全凭证保护：使用 SecretStr 类对敏感凭证进行脱敏处理
• 健康状态缓存：每个 Item 独立的 5 分钟健康检查缓存，减少 API 调用
• 结构化错误映射：将 Plaid API 错误转换为统一格式的 MCP 响应

资料来源：plaid_client.py:1-20

架构总览

┌─────────────────────────────────────────────────────────────────────┐
│                        MCP Client (Claude Code)                      │
└───────────────────────────────┬─────────────────────────────────────┘
                                │ MCP Protocol
                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    server.py (FastMCP)                               │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │
│  │list_accounts│  │get_balances │  │get_transac- │  │get_recurring│ │
│  │             │  │             │  │tions        │  │transactions │ │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘ │
│         │               │               │               │         │
│         └───────────────┴───────┬───────┴───────────────┘         │
│                                 ▼                                  │
│  ┌──────────────────────────────────────────────────────────────┐ │
│  │                    plaid_client.py                           │ │
│  │  - build_api()      构建 Plaid API 客户端                     │ │
│  │  - all_items()      遍历所有 Item 健康状态                     │ │
│  │  - map_plaid_error() 错误映射                                  │ │
│  └──────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
                                │
                                ▼ Plaid API
┌─────────────────────────────────────────────────────────────────────┐
│                      Plaid Cloud API                                 │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │
│  │ Accounts    │  │Transactions │  │ Liabilities │  │ Investments │ │
│  │ Balance     │  │ Recurring   │  │             │  │             │ │
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘ │
└─────────────────────────────────────────────────────────────────────┘

Plaid Link 身份验证流程

流程说明

首次连接银行账户需要使用 Plaid Link 完成身份验证。项目提供了一个本地 FastAPI 应用 link_helper.py 来处理此流程：

1. 用户在浏览器中打开 http://localhost:8765
2. 点击 "Link a bank" 按钮触发 Plaid Link 弹出窗口
3. 用户在 Plaid Link 中完成银行登录
4. Plaid 返回 public_token 到前端页面
5. 后端将 public_token 兑换为长期有效的 access_token
6. 终端输出环境变量行，用户将其添加到 .env 文件

资料来源：link_helper.py:1-50

Link Token 创建

@app.post("/create-link-token")
def create_link_token(req: CreateReq) -> dict:
    if req.update_access_token:
        # 更新模式：用于 re-auth 场景
        body = LinkTokenCreateRequest(
            user=LinkTokenCreateRequestUser(client_user_id="personal-user"),
            client_name="Personal Finance MCP",
            country_codes=[CountryCode("US")],
            language="en",
            access_token=req.update_access_token,
            update=LinkTokenCreateRequestUpdate(account_selection_enabled=False),
        )
    else:
        # 新建模式
        body = LinkTokenCreateRequest(
            user=LinkTokenCreateRequestUser(client_user_id="personal-user"),
            client_name="Personal Finance MCP",
            products=[Products("transactions")],
            optional_products=[
                Products("liabilities"),
                Products("investments"),
            ],
            country_codes=[CountryCode("US")],
            language="en",
        )
    return api.link_token_create(body).to_dict()

资料来源：link_helper.py:60-85

Token 兑换与存储

@app.post("/exchange")
def exchange(req: ExchangeReq) -> dict:
    resp = api.item_public_token_exchange(
        ItemPublicTokenExchangeRequest(public_token=req.public_token)
    ).to_dict()
    access_token = resp["access_token"]
    item_id = resp["item_id"]
    
    # 生成环境变量键名
    env_key = f"PLAID_TOKEN_{env_suffix}"
    print(f"  {env_key}={access_token}", flush=True)

兑换后的 access token 以环境变量形式存储，格式为 PLAID_TOKEN_{机构名称}。

资料来源：link_helper.py:88-120

本地运行保护

link_helper.py 包含防御性检查，禁止在 Horizon 部署环境运行：

if os.environ.get("HORIZON"):
    sys.exit("link_helper.py must not run on Horizon. Run locally only.")

这是为了防止敏感操作暴露在生产环境中。

资料来源：link_helper.py:30-32

API 客户端架构

客户端初始化

build_api() 函数负责创建 Plaid API 实例：

def build_api() -> plaid_api.PlaidApi:
    client_id = os.environ["PLAID_CLIENT_ID"]
    secret = os.environ["PLAID_SECRET"]
    env_name = os.environ.get("PLAID_ENV", "production").lower()
    host = _ENV_MAP.get(env_name, plaid.Environment.Production)
    config = plaid.Configuration(
        host=host,
        api_key={"clientId": client_id, "secret": secret},
    )
    return plaid_api.PlaidApi(plaid.ApiClient(config))

资料来源：plaid_client.py:40-55

环境映射

_ENV_MAP = {
    "production": plaid.Environment.Production,
    "sandbox": plaid.Environment.Sandbox,
}

Token 管理机制

Token 加载

所有银行账户的 access token 通过环境变量加载：

def load_tokens() -> dict[str, SecretStr]:
    out: dict[str, SecretStr] = {}
    prefix = "PLAID_TOKEN_"
    for key, value in os.environ.items():
        if key.startswith(prefix) and value:
            out[key[len(prefix):]] = SecretStr(value)
    return out

资料来源：plaid_client.py:25-35

SecretStr 凭证保护类

class SecretStr:
    __slots__ = ("_value",)

    def __init__(self, value: str) -> None:
        self._value = value

    def reveal(self) -> str:
        return self._value

    def __repr__(self) -> str:
        return "SecretStr('<redacted>')"

    def __str__(self) -> str:
        return "<redacted>"

资料来源：plaid_client.py:10-25

健康状态缓存

缓存数据结构

@dataclass
class ItemHealth:
    env_key: str                          # 环境变量键名
    token: SecretStr                      # 脱敏后的 token
    status: Literal["healthy", "degraded", "error"]  # 健康状态
    reason: str | None                    # 状态原因
    institution_name: str | None          # 机构名称
    checked_at: float                     # 检查时间戳

缓存机制

def all_items(api: PlaidApi) -> Generator[tuple[str, SecretStr, ItemHealth], None, None]:
    # 懒加载缓存，每 5 分钟刷新一次
    if (now - _health_cache.checked_at) < _HEALTH_TTL:
        yield from _health_cache.items
    else:
        _refresh_health_cache(api)
        yield from _health_cache.items

资料来源：plaid_client.py

健康检查流程图

graph TD
    A[all_items 调用] --> B{缓存是否有效?}
    B -->|是| C[返回缓存的 Items]
    B -->|否| D[遍历所有 PLAID_TOKEN_*]
    D --> E[对每个 Item 调用 item_get]
    E --> F{是否有错误?}
    F -->|ITEM_LOGIN_REQUIRED| G[status=error]
    F -->|PRODUCTS_NOT_SUPPORTED| H[status=degraded]
    F -->|无错误| I[status=healthy]
    G --> J[更新缓存]
    H --> J
    I --> J
    J --> K[返回 Items]

错误处理机制

错误映射函数

def map_plaid_error(exc: Exception, institution: str | None) -> dict:
    trace_id = str(uuid.uuid4())
    body: dict = {}
    try:
        parsed = json.loads(getattr(exc, "body", "") or "{}")
        body = parsed if isinstance(parsed, dict) else {}
    except (ValueError, TypeError):
        body = {}
    code = body.get("error_code") or body.get("error_type") or "UNKNOWN"
    message = body.get("error_message") or "Plaid call failed."
    request_id = body.get("request_id")
    
    err: dict = {"code": code, "message": message, "trace_id": trace_id}
    if institution:
        err["institution"] = institution
    return {"error": err}

资料来源：plaid_client.py:55-75

常见错误码

MCP 工具集成

工具注册模式

每个 MCP 工具通过装饰器注册到 FastMCP 服务器：

mcp = FastMCP("personal-finance-mcp")

list_accounts = mcp.tool(
    name="list_accounts",
    annotations={"readOnlyHint": True, "title": "List Accounts"},
)(_list_accounts_impl)

工具列表

资料来源：server.py:1-50

统一调用模式

所有工具遵循相同的调用流程：

def _list_accounts_impl() -> dict:
    api = build_api()
    accounts: list[dict] = []
    warnings: list[dict] = []
    
    for env_key, token, health in all_items(api):
        if health.status != "healthy":
            warnings.append(_warning_from_health(health))
            continue
        try:
            resp = api.accounts_get(
                AccountsGetRequest(access_token=token.reveal())
            ).to_dict()
            for raw in resp.get("accounts", []):
                accounts.append(shape_account(raw, health.institution_name))
        except ApiException as e:
            mapped = map_plaid_error(e, health.institution_name)["error"]
            warnings.append({"institution": health.institution_name, **mapped})
    
    return {"accounts": accounts, "warnings": warnings}

数据响应格式

统一响应结构

所有 MCP 工具返回统一的 JSON 格式：

{
  "[数据字段]": [...],
  "warnings": [
    {
      "code": "ITEM_LOGIN_REQUIRED",
      "message": "...",
      "institution": "CHASE",
      "trace_id": "..."
    }
  ]
}

时间窗口裁剪

交易查询自动裁剪过大的时间范围：

_MAX_LOOKBACK_DAYS = 730  # ~2 years

def _clip_window(start_date: str, end_date: str) -> tuple[str, str, str | None]:
    start = date.fromisoformat(start_date)
    end = date.fromisoformat(end_date)
    earliest = end - timedelta(days=_MAX_LOOKBACK_DAYS)
    if start < earliest:
        return earliest.isoformat(), end.isoformat(), "clipped..."
    return start.isoformat(), end.isoformat(), None

资料来源：server.py:100-115

安全考虑

安全措施

部署安全建议

• 切勿将 .env 文件提交到版本控制
• 生产环境使用 HTTPS 暴露 /mcp 端点
• 使用 OAuth 2.1 或 Cloudflare Access 保护端点
• 或者绑定到私有网络（如 Tailscale）

配置参考

完整环境变量列表

资料来源：server.json

Docker 部署

docker build -t personal-finance-mcp . && \
docker run --rm -p 8000:8000 --env-file .env personal-finance-mcp

故障排查

常见问题

调试方法

1. 检查环境变量是否正确加载：echo $PLAID_TOKEN_*
2. 验证 Plaid 凭证：在 Plaid 仪表板测试 API 密钥
3. 查看服务器日志：python server.py 输出的 stderr 日志
4. 健康状态检查：调用 get_institutions_status 工具

概述

账户与余额工具是 personal-finance-mcp 项目中的核心功能模块，提供对用户所有关联银行账户的只读访问能力。该模块通过集成 Plaid API，实现了两大核心 MCP 工具：list_accounts 和 get_balances。前者负责列举用户所有关联账户的元信息，后者专注于获取账户的实时余额数据。

作为自托管、只读的 MCP 服务器，该模块遵循最小权限原则，仅暴露读取接口，所有数据操作均为查询性质，不涉及任何资金转移或账户修改操作。资料来源：server.py:1-30

工具清单

架构设计

模块依赖关系

graph TD
    A[server.py] --> B[plaid_client.py]
    A --> C[FastMCP]
    B --> D[plaid-python SDK]
    D --> E[Plaid API]
    
    F[link_helper.py] --> B
    G[.env 配置] --> B

数据流架构

sequenceDiagram
    participant MCP as MCP Client
    participant Server as server.py
    participant Client as plaid_client.py
    participant Plaid as Plaid API
    
    MCP->>Server: 调用 list_accounts/get_balances
    Server->>Client: build_api() 构建API客户端
    Server->>Client: all_items() 获取所有关联项
    Client->>Plaid: API 请求
    Plaid-->>Client: 原始响应数据
    Client->>Server: shape_account() 格式化后的数据
    Server-->>MCP: 结构化 JSON 响应

实现详解

1. list_accounts 工具

#### 功能说明

_list_accounts_impl() 函数实现账户列表查询功能。该函数遍历所有通过 Plaid 关联的银行项目（Items），获取每个项目下的所有账户信息，并将原始 Plaid 响应转换为统一格式。资料来源：server.py:70-100

#### 返回数据结构

{
  "accounts": [
    {
      "account_id": "string",
      "name": "string",
      "official_name": "string",
      "type": "depository|credit|loan|investment",
      "subtype": "checking|savings|credit card|mortgage|...",
      "mask": "string",
      "current_balance": "number",
      "available_balance": "number",
      "iso_currency_code": "string",
      "institution": "string"
    }
  ],
  "warnings": [
    {
      "institution": "string",
      "code": "string",
      "message": "string",
      "trace_id": "string"
    }
  ]
}

#### 健康状态检查

在查询账户之前，函数会检查每个关联项的健康状态。对于状态不为 "healthy" 的项目，会生成警告而非直接失败。健康状态包括：

资料来源：plaid_client.py:80-95

2. get_balances 工具

#### 功能说明

_get_balances_impl() 函数专注于获取账户的实时余额数据。该函数支持可选的账户 ID 过滤器，当提供 account_ids 参数时，仅返回指定账户的余额信息；否则返回所有健康关联项下的所有账户余额。资料来源：server.py:105-145

#### 参数说明

#### 过滤逻辑

flowchart TD
    A[开始] --> B{account_ids 是否为空?}
    B -->|是| C[返回所有健康账户余额]
    B -->|否| D{账户是否匹配过滤条件?}
    D -->|是| E[返回该账户余额]
    D -->|否| F{该账户属于其他机构?}
    F -->|是| G[添加 INVALID_ACCOUNT_ID 警告]
    F -->|否| H[静默跳过]

当提供账户ID过滤器但某些账户不匹配时，会根据账户所属机构判断是否发出警告。如果账户属于未关联的机构项目，则返回 INVALID_ACCOUNT_ID 警告；如果属于已关联但未选择该账户的项目，则静默跳过。

资料来源：server.py:130-140

3. 底层 Plaid 客户端

#### SecretStr 安全机制

plaid_client.py 实现了 SecretStr 类用于安全处理敏感凭据。该类确保访问令牌在日志和调试输出中被脱敏处理。资料来源：plaid_client.py:35-55

class SecretStr:
    def __repr__(self) -> str:
        return "SecretStr('<redacted>')"
    
    def __str__(self) -> str:
        return "<redacted>"

#### 令牌加载机制

load_tokens() 函数从环境变量中加载所有以 PLAID_TOKEN_ 前缀开头的配置项，每个关联银行对应一个独立的访问令牌。资料来源：plaid_client.py:60-65

def load_tokens() -> dict[str, SecretStr]:
    out: dict[str, SecretStr] = {}
    prefix = "PLAID_TOKEN_"
    for key, value in os.environ.items():
        if key.startswith(prefix) and value:
            out[key[len(prefix):]] = SecretStr(value)
    return out

#### 错误映射机制

map_plaid_error() 函数将 Plaid API 异常转换为统一的错误结构，包含错误码、错误消息、trace_id 和关联机构信息，便于 MCP 客户端进行错误处理和日志追踪。资料来源：plaid_client.py:90-115

配置要求

环境变量配置

.env 配置示例

PLAID_CLIENT_ID=your_client_id_here
PLAID_SECRET=your_secret_here
PLAID_ENV=production

# 银行令牌（通过 link_helper.py 获取）
PLAID_TOKEN_CHASE=access-prod-xxxxx
PLAID_TOKEN_BANKOFAMERICA=access-prod-yyyyy

使用场景

查询所有账户余额

# MCP 调用示例
result = mcp.call_tool("get_balances", {})

返回结果示例：

{
  "accounts": [
    {
      "account_id": "BxBXxJj...",
      "name": "Plaid Checking",
      "type": "depository",
      "subtype": "checking",
      "current_balance": 110.35,
      "available_balance": 100.35,
      "institution": "Chase"
    }
  ],
  "warnings": []
}

查询特定账户

# 查询指定账户ID的余额
result = mcp.call_tool("get_balances", {
    "account_ids": ["BxBXxJj...", "KxdLLm..."]
})

警告机制

账户与余额工具设计了完善的警告机制，用于在部分数据无法获取时仍能返回可用信息：

性能特性

• 分页处理：内部使用 offset 分页，批量获取数据（每页500条）
• 健康缓存：使用5分钟 TTL 缓存关联项健康状态，减少 API 调用
• 并行处理：按机构分别查询，各机构数据独立获取

相关文档

• 快速入门指南：参见 README.md
• 部署指南：参见 docs/DEPLOYMENT.md
• 其他工具：交易查询工具、投资工具

工具概览

资料来源：server.py:86-145

核心实现架构

graph TD
    A[MCP Client 请求] --> B{选择工具}
    B --> C[get_transactions]
    B --> D[search_transactions]
    
    C --> E[build_api]
    D --> E
    
    E --> F[Plaid API]
    F --> G{健康检查}
    G -->|Item 正常| H[分页获取交易]
    G -->|Item 异常| I[生成警告]
    
    H --> J{数据处理}
    C --> J
    D --> J
    
    J --> K[shape_transaction 格式化]
    K --> L[返回结果 + warnings]

组件职责

资料来源：server.py:1-30

get_transactions 工具

功能描述

get_transactions 是用于获取指定日期范围内所有交易记录的主要工具。它自动处理分页（每页 500 条），遍历所有健康的银行账户（Item），并对超过两年的查询窗口进行自动裁剪。

资料来源：server.py:108-145

函数签名

def _get_transactions_impl(
    start_date: str,
    end_date: str,
    account_ids: list[str] | None = None,
) -> dict:

参数说明

返回值结构

{
  "transactions": [
    {
      "transaction_id": "...",
      "account_id": "...",
      "date": "2024-01-15",
      "name": "STARBUCKS #12345",
      "merchant_name": "Starbucks",
      "amount": 5.75,
      "category": ["Food and Drink", "Coffee Shop"],
      "pending": false,
      "institution": "Chase"
    }
  ],
  "warnings": [
    {
      "code": "WINDOW_CLIPPED",
      "reason": "clipped start from...",
      "message": "..."
    }
  ]
}

资料来源：server.py:126-145

分页处理逻辑

while True:
    options = TransactionsGetRequestOptions(**{**base_options, "offset": offset})
    resp = api.transactions_get(
        TransactionsGetRequest(
            access_token=token.reveal(),
            start_date=date.fromisoformat(clipped_start),
            end_date=date.fromisoformat(clipped_end),
            options=options,
        )
    ).to_dict()
    batch = resp.get("transactions", []) or []
    for raw in batch:
        transactions.append(shape_transaction(raw))
    total = resp.get("total_transactions") or 0
    offset += len(batch)
    if offset >= total or not batch:
        break

系统使用 offset 偏移分页方式，每次请求 500 条记录（count=500），持续遍历直到获取所有交易或遇到空响应。

资料来源：server.py:130-142

search_transactions 工具

功能描述

search_transactions 提供了关键词搜索功能，允许用户通过关键词查找特定商户、交易名称或交易对手方的相关交易。该工具在获取原始 Plaid 数据后，对以下字段进行大小写不敏感的子字符串匹配：

• merchant_name - 商户名称
• name - 交易名称
• counterparties[].name - 交易对手方名称

注意：对手方名称（counterparties）仅在搜索功能中可用，因为 shape_transaction 函数会丢弃该字段。

资料来源：server.py:51-85

与 get_transactions 的区别

函数签名

def _search_transactions_impl(
    query: str,
    start_date: str,
    end_date: str,
) -> dict:

参数说明

搜索实现逻辑

q = query.lower()

for env_key, token, health in all_items(api):
    if health.status != "healthy":
        warnings.append(_warning_from_health(health))
        continue
    offset = 0
    # ... 分页获取交易 ...
    for raw in batch:
        transaction = shape_transaction(raw)
        # 检查搜索条件
        if (q in transaction.get("name", "").lower() or
            q in transaction.get("merchant_name", "").lower() or
            q in str(raw.get("counterparties", [])).lower()):
            transactions.append(transaction)

搜索匹配在原始 Plaid 载荷上执行（raw），而非格式化后的 transaction 对象，以确保对手方名称可被搜索到。

资料来源：server.py:55-75

日期窗口管理

两年纪录限制

Plaid API 对交易历史查询有约两年的回溯限制。系统通过 _clip_window() 函数自动处理这一限制：

_MAX_LOOKBACK_DAYS = 730  # ~2 years

def _clip_window(start_date: str, end_date: str) -> tuple[str, str, str | None]:
    """Return (start, end, warning_reason_or_None) clipped to the 2-year window."""
    start = date.fromisoformat(start_date)
    end = date.fromisoformat(end_date)
    earliest = end - timedelta(days=_MAX_LOOKBACK_DAYS)
    if start < earliest:
        return earliest.isoformat(), end.isoformat(), (
            f"clipped start from {start.isoformat()} to {earliest.isoformat()} "
            "(Plaid max lookback ~2 years)"
        )
    return start.isoformat(), end.isoformat(), None

裁剪规则说明

警告示例

当日期窗口被裁剪时，返回结果中会包含 WINDOW_CLIPPED 警告：

{
  "code": "WINDOW_CLIPPED",
  "reason": "clipped start from 2022-01-01 to 2023-05-15 (Plaid max lookback ~2 years)",
  "message": "clipped start from 2022-01-01 to 2023-05-15 (Plaid max lookback ~2 years)"
}

资料来源：server.py:86-106

交易数据结构

shape_transaction 函数

shape_transaction 函数将 Plaid API 返回的原始交易数据转换为统一格式：

# 来源于 plaid_client.py 的 shape_transaction
def shape_transaction(raw: dict, institution: str | None = None) -> dict:
    return {
        "transaction_id": raw.get("transaction_id"),
        "account_id": raw.get("account_id"),
        "date": str(raw.get("date")) if raw.get("date") else None,
        "name": raw.get("name"),
        "merchant_name": raw.get("merchant_name"),
        "amount": raw.get("amount"),
        "currency": raw.get("iso_currency_code"),
        "category": raw.get("category"),
        "pending": raw.get("pending"),
        "institution": institution,
    }

输出字段说明

金额约定：amount 为正数表示支出，负数表示存款或退款。

错误处理机制

Item 健康状态检查

在查询交易前，系统会检查每个银行账户（Item）的健康状态：

for env_key, token, health in all_items(api):
    if health.status != "healthy":
        warnings.append(_warning_from_health(health))
        continue

健康状态类型

错误映射

Plaid API 错误通过 map_plaid_error() 统一映射：

def map_plaid_error(exc: Exception, institution: str | None) -> dict:
    trace_id = str(uuid.uuid4())
    body: dict = {}
    try:
        parsed = json.loads(getattr(exc, "body", "") or "{}")
        body = parsed if isinstance(parsed, dict) else {}
    except (ValueError, TypeError):
        body = {}
    code = body.get("error_code") or body.get("error_type") or "UNKNOWN"
    message = body.get("error_message") or "Plaid call failed."
    # ... 日志记录 ...
    return {"error": {"code": code, "message": message, "trace_id": trace_id}}

警告结构

{
  "institution": "Chase",
  "code": "ITEM_LOGIN_REQUIRED",
  "message": "the login details of this item have changed",
  "trace_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
}

资料来源：plaid_client.py:50-80

使用示例

示例 1：获取最近 30 天交易

{
  "tool": "get_transactions",
  "arguments": {
    "start_date": "2024-12-01",
    "end_date": "2024-12-31"
  }
}

示例 2：搜索特定商户交易

{
  "tool": "search_transactions",
  "arguments": {
    "query": "starbucks",
    "start_date": "2024-01-01",
    "end_date": "2024-12-31"
  }
}

示例 3：获取特定账户交易

{
  "tool": "get_transactions",
  "arguments": {
    "start_date": "2024-01-01",
    "end_date": "2024-12-31",
    "account_ids": ["acc_xyz789"]
  }
}

技术依赖

资料来源：requirements.txt:1-7

已知限制

1. 两年回溯限制：Plaid API 限制交易查询最多回溯约两年，系统会自动裁剪超长窗口
2. 只读操作：所有交易工具均为只读，不支持创建、修改或删除交易
3. 无搜索过滤：search_transactions 不支持按账户 ID 过滤
4. 对手方数据丢失：格式化后的交易对象不包含对手方信息，仅在搜索时使用原始数据

相关文档

• 架构文档 - 包含为何使用 /transactions/get 而非 /transactions/sync 的详细说明
• 银行链接工具 - 如何使用 Plaid Link 链接新账户

概述

personal-finance-mcp 提供了三个与投资和负债相关的只读 MCP 工具，用于获取用户的投资持仓、交易历史以及各类负债信息。这些工具通过 Plaid API 获取数据，支持信用卡、学生贷款、抵押贷款、投资账户等多种金融产品的查询。

工具列表

架构设计

组件关系

graph TD
    A[MCP Client] -->|调用工具| B[server.py]
    B --> C[plaid_client.py]
    C --> D[Plaid API]
    D -->|持仓数据| E[get_investment_holdings]
    D -->|交易数据| F[get_investment_transactions]
    D -->|负债数据| G[get_liabilities]
    C -->|健康检查| H[ItemHealth Cache]
    H -->|缓存 TTL 300s| I[_health_cache]

数据流处理流程

graph TD
    A[all_items 遍历所有环境变量] --> B{健康状态检查}
    B -->|healthy| C[调用 Plaid API]
    B -->|非 healthy| D[生成健康警告]
    C --> E{API 异常处理}
    E -->|无异常| F[数据整形 shape_*]
    E -->|ApiException| G[map_plaid_error]
    F --> H[返回结果 + warnings]
    G --> I[错误映射 + warnings]
    H --> J[最终响应]
    I --> J

投资持仓工具 (`get_investment_holdings`)

功能说明

get_investment_holdings 用于获取用户所有投资账户的当前持仓信息，包括股票、债券、基金等证券的持有数量、成本基础和当前市值。

实现源码

工具通过 InvestmentsHoldingsGetRequest 调用 Plaid API：

# server.py:144-192
def _get_investment_holdings_impl() -> dict:
    api = build_api()
    holdings: list[dict] = []
    warnings: list[dict] = []

    for env_key, token, health in all_items(api):
        if health.status != "healthy":
            warnings.append(_warning_from_health(health))
            continue
        try:
            resp = api.investments_holdings_get(
                InvestmentsHoldingsGetRequest(access_token=token.reveal())
            ).to_dict()
            secs_by_id = {s["security_id"]: s for s in resp.get("securities", []) or []}
            for h in resp.get("holdings", []) or []:
                holdings.append(shape_holding(h, secs_by_id))
        except ApiException as e:
            mapped = map_plaid_error(e, health.institution_name)["error"]
            warnings.append({"institution": health.institution_name, **mapped})
    return {"holdings": holdings, "warnings": warnings}

返回数据结构

投资交易工具 (`get_investment_transactions`)

功能说明

get_investment_transactions 获取指定日期范围内的投资交易记录，包括买入、卖出和股息收入。工具支持分页处理，默认每页500条记录。

参数定义

分页机制

# server.py:228-264
offset = 0
while True:
    options = InvestmentsTransactionsGetRequestOptions(count=500, offset=offset)
    resp = api.investments_transactions_get(
        InvestmentsTransactionsGetRequest(
            access_token=token.reveal(),
            start_date=date.fromisoformat(start_date),
            end_date=date.fromisoformat(end_date),
            options=options,
        )
    ).to_dict()
    # 处理批次数据...
    offset += len(batch)
    if offset >= total or not batch:
        break

返回数据结构

负债工具 (`get_liabilities`)

功能说明

get_liabilities 获取用户所有负债账户的详细信息，包括信用卡余额、利率、还款计划，以及学生贷款和抵押贷款的摊销信息。

实现源码

# server.py:114-142
def _get_liabilities_impl() -> dict:
    api = build_api()
    liabilities: list[dict] = []
    warnings: list[dict] = []

    for env_key, token, health in all_items(api):
        if health.status != "healthy":
            warnings.append(_warning_from_health(health))
            continue
        try:
            resp = api.liabilities_get(
                LiabilitiesGetRequest(access_token=token.reveal())
            ).to_dict()
            for raw in resp.get("liabilities", {}).get("credit", []) or []:
                liabilities.append(shape_liability(raw, health.institution_name))
            for raw in resp.get("liabilities", {}).get("student_loan", []) or []:
                liabilities.append(shape_liability(raw, health.institution_name))
            for raw in resp.get("liabilities", {}).get("mortgage", []) or []:
                liabilities.append(shape_liability(raw, health.institution_name))
        except ApiException as e:
            mapped = map_plaid_error(e, health.institution_name)["error"]
            warnings.append({"institution": health.institution_name, **mapped})
    return {"liabilities": liabilities, "warnings": warnings}

负债类型

错误处理机制

健康检查与缓存

plaid_client.py 实现了基于 TTL 的健康检查缓存机制：

# plaid_client.py:93-120
_health_cache: dict[str, tuple[ItemHealth, float]] = {}
_CACHE_TTL_SEC = 300  # 5分钟缓存

def get_item_health(api, env_key: str, token: SecretStr) -> ItemHealth:
    now = time.time()
    cached = _health_cache.get(env_key)
    if cached and (now - cached[1]) < _CACHE_TTL_SEC:
        return cached[0]
    # ... 执行健康检查并更新缓存

Plaid 错误映射

# plaid_client.py:63-84
_ERROR_TO_STATUS: dict[str, HealthStatus] = {
    "ITEM_LOGIN_REQUIRED": "re_auth_required",
    "PENDING_EXPIRATION": "pending_expiration",
    "ITEM_LOCKED": "item_locked",
    "NO_ACCOUNTS": "no_accounts",
}

健康状态类型

环境配置

必需环境变量

Token 管理

投资和负债工具依赖 PLAID_TOKEN_* 环境变量存储各银行的访问令牌：

# plaid_client.py:46-52
def load_tokens() -> dict[str, SecretStr]:
    out: dict[str, SecretStr] = {}
    prefix = "PLAID_TOKEN_"
    for key, value in os.environ.items():
        if key.startswith(prefix) and value:
            out[key[len(prefix):]] = SecretStr(value)
    return out

Token 通过 link_helper.py 中的 Plaid Link 流程获取：

# link_helper.py:47-56
@app.post("/exchange")
def exchange(req: ExchangeReq) -> dict:
    resp = api.item_public_token_exchange(...)
    access_token = resp["access_token"]
    env_suffix = "".join(ch for ch in ins_name.upper() if ch.isalnum())
    env_key = f"PLAID_TOKEN_{env_suffix}"
    print(f"  {env_key}={access_token}")

使用示例

查询所有投资持仓

# MCP 工具调用
get_investment_holdings()

响应示例：

{
  "holdings": [
    {
      "account_id": "acc_123",
      "symbol": "AAPL",
      "name": "Apple Inc.",
      "quantity": 50.0,
      "institution_price": 178.50,
      "institution_value": 8925.00,
      "cost_basis": 7500.00,
      "institution": "CHASE"
    }
  ],
  "warnings": []
}

查询投资交易

# MCP 工具调用
get_investment_transactions(
    start_date="2024-01-01",
    end_date="2024-12-31"
)

查询负债

# MCP 工具调用
get_liabilities()

依赖关系

项目依赖以下核心库：

# requirements.txt
plaid-python>=39.1.0,<40.0.0  # Plaid API 客户端
fastmcp>=3.2.4,<4.0.0         # MCP 服务框架

资料来源：requirements.txt:1-7

注意事项

1. 只读限制：所有工具均为只读操作，不支持修改账户或发起交易
2. 数据时效：投资价格数据来自机构，可能存在延迟
3. 访问权限：需在 Plaid Dashboard 启用 Investments 和 Liabilities 产品
4. 账户健康：非 healthy 状态的账户会自动跳过并返回警告
5. 缓存机制：健康检查缓存有效期为 300 秒

相关工具

本项目中完整的 9 个 MCP 工具包括：

• list_accounts - 列出所有账户
• get_balances - 获取账户余额
• get_transactions - 获取交易记录
• search_transactions - 搜索交易
• get_recurring_transactions - 获取定期交易
• get_liabilities - 获取负债信息
• get_investment_holdings - 获取投资持仓
• get_investment_transactions - 获取投资交易
• get_institutions_status - 获取机构状态

部署方案概览

personal-finance-mcp 支持以下部署方式：

环境变量配置

无论选择哪种部署方案，都需要配置以下环境变量：

资料来源：server.json:8-34

获取 Plaid 凭证

1. 访问 https://dashboard.plaid.com/signup 注册 Plaid 账户
2. 选择 Trial 计划（免费，10 个 Items）
3. 进入 Team Settings → Products，启用 Transactions、Liabilities、Investments
4. 进入 Team Settings → API，复制 client_id 和 secret

访问令牌配置

每绑定一个银行账户，会生成一个独立的访问令牌：

PLAID_TOKEN_CHASE=access-prod-xxx...
PLAID_TOKEN_BANKOFAMERICA=access-prod-yyy...
PLAID_TOKEN_UNKNOWN=access-prod-zzz...

资料来源：link_helper.py:35-45

Docker 部署

Docker 是最简单直接的部署方式，项目已包含完整的 Dockerfile。

构建镜像

docker build -t personal-finance-mcp .

运行容器

docker run --rm -p 8000:8000 --env-file .env personal-finance-mcp

访问 MCP 端点

容器运行后，MCP 服务可通过以下地址访问：

http://localhost:8000/mcp

Prefect Horizon 部署

Prefect Horizon 是一种零成本、适合长期运行的部署方案。

架构图

graph LR
    A[Claude Code] -->|HTTP| B[Prefect Horizon]
    B -->|代理| C[personal-finance-mcp 容器]
    C -->|HTTPS| D[Plaid API]
    
    E[link_helper.py] -->|本地| F[Plaid Link]
    F -->|access_token| C

部署步骤

1. 准备环境文件

创建 horizon.env 文件，包含所有必需的环境变量：

PLAID_CLIENT_ID=your_client_id
PLAID_SECRET=your_secret
PLAID_ENV=production
PLAID_TOKEN_CHASE=access-prod-xxx...
PLAID_TOKEN_BANKOFAMERICA=access-prod-yyy...

1. 在 Prefect Horizon 中创建服务

详细步骤请参阅 docs/DEPLOYMENT.md。

1. 验证部署

curl http://localhost:8000/mcp

Fly.io 部署

初始化 Fly.io 应用

fly launch
fly secrets set PLAID_CLIENT_ID=xxx PLAID_SECRET=yyy
fly secrets set PLAID_TOKEN_CHASE=access-prod-xxx
fly deploy

配置健康检查

确保 fly.toml 中配置了正确的健康检查端点。

云服务器通用部署

对于 VPS、Railway 或其他云平台：

1. 安装依赖

git clone https://github.com/JosueM1109/personal-finance-mcp.git
cd personal-finance-mcp
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

2. 配置环境变量

cp .env.example .env
# 编辑 .env 填写实际值

3. 配置反向代理

生产环境建议使用 Nginx 或 Caddy 配置 HTTPS：

server {
    listen 443 ssl;
    server_name your-domain.com;
    
    location /mcp {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

4. 使用 systemd 管理服务

创建 /etc/systemd/system/personal-finance-mcp.service：

[Unit]
Description=Personal Finance MCP Server
After=network.target

[Service]
Type=simple
User=your-user
WorkingDirectory=/path/to/personal-finance-mcp
Environment="PATH=/path/to/personal-finance-mcp/.venv/bin"
ExecStart=/path/to/personal-finance-mcp/.venv/bin/python server.py
Restart=on-failure

[Install]
WantedBy=multi-user.target

启动服务：

sudo systemctl enable personal-finance-mcp
sudo systemctl start personal-finance-mcp

连接 Claude Code

部署完成后，将 MCP 服务添加到 Claude Code：

claude mcp add --transport http personal-finance http://localhost:8000/mcp

验证连接：

"list my accounts"

资料来源：README.md:58-62

安全配置

重要警告

暴露 MCP 端点风险：带有访问令牌的公开 MCP 端点会泄露所有关联账户信息。

资料来源：README.md:46-47

安全建议

link_helper 安全说明

link_helper.py 包含 Plaid Link API 应用，拒绝在 HORIZON=1 环境下运行。

资料来源：README.md:51

银行账户绑定

本地绑定流程

sequenceDiagram
    participant U as 用户
    participant LH as link_helper:8765
    participant P as Plaid Link
    participant API as Plaid API
    
    U->>LH: 访问 localhost:8765
    LH->>API: /create-link-token
    API->>LH: link_token
    LH->>U: 显示"Link a bank"按钮
    U->>P: 点击按钮，完成银行验证
    P->>LH: public_token
    LH->>API: /exchange
    API->>LH: access_token
    LH->>U: 终端显示 PLAID_TOKEN_xxx

绑定步骤

1. 启动 link_helper：

uvicorn link_helper:app --port 8765

1. 打开浏览器访问 http://localhost:8765

1. 点击 Link a bank 按钮，完成 Plaid Link 流程

1. 终端输出访问令牌，例如：

============================================================
Institution: Chase Bank
Item ID:     xxx-xxx-xxx
Add this to your .env (local) and Horizon env (prod):
  PLAID_TOKEN_CHASE=access-prod-xxx...
Do NOT commit this line.
============================================================

1. 将令牌添加到 .env 文件或对应的环境变量

资料来源：link_helper.py:1-75

故障排除

常见问题

健康检查

使用 get_institutions_status 工具检查所有关联银行的状态：

claude: "Any bank that needs re-authentication?"

资料来源：README.md:32

错误代码映射

资料来源：plaid_client.py:108-113

生产环境检查清单

• [ ] 所有环境变量已配置
• [ ] HTTPS 已启用
• [ ] 访问令牌已添加
• [ ] link_helper 已完成银行绑定
• [ ] Claude Code MCP 连接已测试
• [ ] 安全措施已实施（VPN/访问控制）
• [ ] 监控和日志已配置

核心安全原则

单租户架构

personal-finance-mcp 采用单租户（Single-tenant） 架构设计，这意味着每个用户应独立部署自己的服务器实例。系统明确禁止共享部署或多人共用同一服务器实例。

重要警示：暴露的 MCP 端点配合您的令牌会泄露所有链接账户的财务数据，必须严格控制访问权限。

资料来源：README.md:安全部分

最小权限原则

系统仅提供只读（read-only） 访问能力，所有 9 个工具均为只读操作：

资料来源：README.md:What you can ask、server.py:各工具定义

凭证安全管理

SecretStr 封装机制

项目使用自定义 SecretStr 类封装敏感凭证，防止在日志和调试输出中意外泄露：

class SecretStr:
    __slots__ = ("_value",)

    def __init__(self, value: str) -> None:
        self._value = value

    def reveal(self) -> str:
        return self._value

    def __repr__(self) -> str:
        return "SecretStr('<redacted>')"

    def __str__(self) -> str:
        return "<redacted>"

该类的 __repr__ 和 __str__ 方法始终返回 <redacted> 或 SecretStr('<redacted>')，确保即使在异常堆栈或日志输出中也不会暴露真实凭证值。

资料来源：plaid_client.py:SecretStr类定义

环境变量令牌加载

所有 Plaid 访问令牌通过 load_tokens() 函数从环境变量加载，令牌密钥以 PLAID_TOKEN_ 为前缀：

def load_tokens() -> dict[str, SecretStr]:
    out: dict[str, SecretStr] = {}
    prefix = "PLAID_TOKEN_"
    for key, value in os.environ.items():
        if key.startswith(prefix) and value:
            out[key[len(prefix):]] = SecretStr(value)
    return out

每个银行链接后生成的令牌环境变量格式为 PLAID_TOKEN_{机构名称}，例如：

• PLAID_TOKEN_CHASE
• PLAID_TOKEN_WELLSFARGO
• PLAID_TOKEN_FIDELITY

资料来源：plaid_client.py:load_tokens函数

.env 文件保护

项目提供 .env.example 作为模板文件，.gitignore 确保实际 .env 文件不会被提交到版本控制系统：

.env

这防止了包含 PLAID_CLIENT_ID、PLAID_SECRET 和各类 PLAID_TOKEN_* 的敏感文件泄露。

资料来源：.gitignore:敏感文件排除

MCP 端点保护

风险分析

MCP 服务器默认监听 http://localhost:8000/mcp，当暴露到公网时，未授权访问将导致：

graph TD
    A[攻击者] -->|未授权访问| B[MCP 端点]
    B --> C[获取所有账户列表]
    B --> D[获取交易记录]
    B --> E[获取投资持仓]
    B --> F[获取负债信息]
    C --> G[完整财务数据泄露]
    D --> G
    E --> G
    F --> G

资料来源：README.md:部署和安全部分

推荐的端点保护方案

项目文档推荐以下三种端点保护方案：

#### 本地开发环境

对于本地开发，服务器仅绑定到 localhost，无需额外认证：

python server.py   # 仅监听 localhost:8000

资料来源：README.md:Quickstart部分

#### Docker 部署保护

Docker 部署时可通过环境变量文件加载敏感配置：

docker run --rm -p 8000:8000 --env-file .env personal-finance-mcp

确保 .env 文件权限设置为 600（仅所有者可读写）：

chmod 600 .env

资料来源：README.md:Docker部署

私有网络部署架构

对于 Raspberry Pi + Tailscale 或 VPS 部署场景，推荐的网络拓扑：

graph LR
    subgraph 用户网络
        A[MCP客户端]
    end
    
    subgraph 安全边界
        B[Tailscale VPN / Cloudflare Access]
    end
    
    subgraph 部署服务器
        C[personal-finance-mcp]
        D[Plaid API]
    end
    
    A -->|加密隧道| B
    B -->|身份验证| C
    C -->|HTTPS| D

资料来源：README.md:部署部分

Plaid Link 令牌管理

链接助手安全机制

link_helper.py 提供一次性令牌获取功能，该服务拒绝在 HORIZON=1 环境下运行：

@app.get("/health")
def health():
    if os.environ.get("HORIZON") == "1":
        raise HTTPException(status_code=503, detail="Horizon environment")

此机制防止在生产环境意外启动链接助手，确保用户只能在本地或专用环境中添加新账户。

资料来源：link_helper.py:HORIZON检查

令牌获取流程

通过 Plaid Link 获取访问令牌的完整流程：

sequenceDiagram
    participant 用户
    participant 浏览器
    participant link_helper
    participant Plaid
    
    用户->>浏览器: 点击"Link a bank"
    浏览器->>link_helper: POST /create-link-token
    link_helper->>Plaid: link_token_create
    Plaid-->>link_helper: link_token
    link_helper-->>浏览器: link_token
    浏览器->>Plaid: 打开Plaid Link弹窗
    用户->>Plaid: 完成银行验证
    Plaid-->>浏览器: public_token
    浏览器->>link_helper: POST /exchange
    link_helper->>Plaid: item_public_token_exchange
    Plaid-->>link_helper: access_token
    link_helper->>终端: 打印PLAID_TOKEN_xxx=...

资料来源：link_helper.py:API端点实现

令牌输出安全

系统通过终端直接输出访问令牌，而非通过 HTTP 响应返回，这避免了令牌通过日志文件或网络传输泄露：

print("=" * 60, flush=True)
print(f"Institution: {ins_name}", flush=True)
print(f"Item ID:     {item_id}", flush=True)
print("Add this to your .env (local) and Horizon env (prod):", flush=True)
print(f"  {env_key}={access_token}", flush=True)
print("Do NOT commit this line.", flush=True)
print("=" * 60, flush=True)

资料来源：link_helper.py:令牌输出逻辑

Plaid API 凭证保护

凭证配置

Plaid API 需要两类凭证，均标记为必需且敏感：

资料来源：server.json:环境变量定义

API 客户端构建

build_api() 函数从环境变量构建 Plaid API 客户端：

def build_api() -> plaid_api.PlaidApi:
    client_id = os.environ["PLAID_CLIENT_ID"]
    secret = os.environ["PLAID_SECRET"]
    env_name = os.environ.get("PLAID_ENV", "production").lower()
    host = _ENV_MAP.get(env_name, plaid.Environment.Production)
    config = plaid.Configuration(
        host=host,
        api_key={"clientId": client_id, "secret": secret},
    )
    return plaid_api.PlaidApi(plaid.ApiClient(config))

警告：切勿将包含真实 PLAID_SECRET 的代码提交到版本控制系统。

资料来源：plaid_client.py:build_api函数

错误处理与日志安全

敏感信息脱敏

map_plaid_error() 函数处理 Plaid API 错误时，对敏感信息进行脱敏处理：

def map_plaid_error(exc: Exception, institution: str | None) -> dict:
    trace_id = str(uuid.uuid4())
    body: dict = {}
    try:
        parsed = json.loads(getattr(exc, "body", "") or "{}")
        body = parsed if isinstance(parsed, dict) else {}
    except (ValueError, TypeError):
        body = {}
    code = body.get("error_code") or body.get("error_type") or "UNKNOWN"
    message = body.get("error_message") or "Plaid call failed."
    request_id = body.get("request_id")
    _log.warning(
        "plaid_error trace_id=%s request_id=%s code=%s",
        trace_id,
        request_id,
        code,
    )

日志仅记录 trace_id、request_id 和 error_code，不包含敏感的认证信息或账户数据。

资料来源：plaid_client.py:map_plaid_error函数

错误响应结构

API 返回的错误响应包含以下字段：

资料来源：plaid_client.py:错误映射逻辑

部署安全检查清单

部署前检查

链接新银行账户

添加新银行时，确保：

1. 仅在受信任网络：使用本地网络或 VPN 连接
2. 使用 link_helper：通过 uvicorn link_helper:app --port 8765 启动
3. 验证 HORIZON 保护：生产环境 HORIZON=1 会阻止链接助手运行
4. 令牌安全存储：将生成的 PLAID_TOKEN_* 环境变量添加到 .env

资料来源：README.md:链接银行账户流程

第三方依赖安全

项目依赖项（定义于 requirements.txt）需定期更新以获取安全修复：

fastmcp>=3.2.4,<4.0.0
plaid-python>=39.1.0,<40.0.0
python-dotenv>=1.0.0
fastapi>=0.115.0
uvicorn>=0.32.0
pytest>=8.0.0
pytest-mock>=3.12.0

建议使用以下命令检查已知漏洞：

pip-audit

或使用 GitHub Dependabot 自动监控依赖更新。

资料来源：requirements.txt:依赖定义

免责声明

README.md 中明确声明：

Unofficial. This project is not affiliated with, endorsed by, or sponsored by Plaid Inc. "Plaid" is a trademark of Plaid Inc. This is a self-hosted client that talks to Plaid's API using credentials you supply.

用户需自行承担部署和使用本项目的安全责任，确保符合当地法律法规以及 Plaid 的服务条款。

资料来源：README.md:免责声明

概述

personal-finance-mcp 是一个自托管的只读 MCP 服务器，连接银行、投资和贷款账户。所有 9 个工具均设计为只读操作，通过 Plaid API 获取数据。服务器实现了健康检查机制来监控每个已连接账户（Item）的状态，并包含错误映射系统将 Plaid API 错误转换为结构化响应。

资料来源：README.md:1-20

核心诊断工具

获取机构状态

get_institutions_status 工具可诊断所有已链接银行账户的健康状态：

{
  "items": [
    {
      "env_key": "CHASE",
      "institution_id": "ins_3",
      "institution_name": "Chase",
      "item_id": "abc123",
      "status": "healthy",
      "reason": null
    }
  ]
}

资料来源：server.py:150-175

健康状态类型

资料来源：plaid_client.py:95-100

常见错误诊断

Plaid API 错误映射

服务器内置错误处理机制，将 Plaid API 异常转换为结构化错误响应：

{
  "error": {
    "code": "ITEM_LOGIN_REQUIRED",
    "message": "the login details of this item have not been received",
    "trace_id": "uuid-v4-format",
    "institution": "Chase"
  }
}

错误响应包含以下字段：

资料来源：plaid_client.py:52-64

错误代码与状态映射

资料来源：plaid_client.py:101-108

银行链接问题

Plaid Link 配置检查

link_helper.py 提供本地 Plaid Link 界面用于一次性 Token 配置。启动前需确认：

1. 环境变量正确配置：

• PLAID_CLIENT_ID
• PLAID_SECRET
• PLAID_ENV（production 或 sandbox）

1. Plaid 仪表板已启用所需产品：

• Transactions（交易）
• Liabilities（负债，可选）
• Investments（投资，可选）

资料来源：README.md:40-55

访问 Token 问题

首次链接银行后，终端会输出类似内容：

PLAID_TOKEN_CHASE=access-prod-xxx...

将此行添加到 .env 文件（本地）和生产环境变量中。

资料来源：link_helper.py:70-75

HORIZON 环境变量限制

link_helper.py 检测到 HORIZON=1 环境变量时会拒绝运行，这是预设的安全机制：

if os.environ.get("HORIZON") == "1":
    raise RuntimeError("HORIZON=1 is set — link_helper refuses to run in this mode.")

资料来源：link_helper.py:8-10

交易数据问题

日期窗口限制

Plaid API 对交易查询有约 2 年的回溯限制。服务器自动处理此限制：

clipped start from 2022-01-01 to 2023-01-01 (Plaid max lookback ~2 years)

当查询范围超过 2 年时，服务器会：

1. 自动裁剪开始日期
2. 返回 WINDOW_CLIPPED 警告
3. 继续获取被裁剪日期范围内的数据

资料来源：server.py:110-125

分页处理

交易数据使用偏移量分页（每页 500 条）：

while True:
    resp = api.transactions_get(
        TransactionsGetRequest(
            access_token=token.reveal(),
            start_date=date.fromisoformat(clipped_start),
            end_date=date.fromisoformat(clipped_end),
            options=TransactionsGetRequestOptions(count=500, offset=offset),
        )
    ).to_dict()

资料来源：server.py:130-145

搜索交易失败

搜索交易时会过滤以下字段：

• merchant_name
• name
• counterparties[].name

搜索采用不区分大小写的子字符串匹配。Plaid 可能不返回 counterparties 数据，因此建议使用更通用的交易描述进行搜索。

资料来源：server.py:195-210

健康检查缓存

服务器实现了 5 分钟 TTL 的健康检查缓存：

_health_cache: dict[str, tuple[ItemHealth, float]] = {}
_CACHE_TTL_SEC = 300

这意味着状态变更（如重新认证）后，可能需要等待最多 5 分钟才能反映到健康状态中。

资料来源：plaid_client.py:110-115

如需立即刷新缓存，可重启服务器进程。

安全相关问题

端点暴露风险

README.md 明确警告：

Gate the endpoint. An exposed MCP endpoint with your tokens leaks every linked account.

建议的安全措施：

• 使用 OAuth 2.1
• 使用 Cloudflare Access
• 仅在私有网络绑定服务

资料来源：README.md:95-100

Secret 管理

所有 Plaid 凭证通过环境变量注入，不存储在代码或配置文件中：

def build_api() -> plaid_api.PlaidApi:
    client_id = os.environ["PLAID_CLIENT_ID"]
    secret = os.environ["PLAID_SECRET"]
    env_name = os.environ.get("PLAID_ENV", "production").lower()

资料来源：plaid_client.py:45-52

依赖问题

Python 版本要求

需要 Python 3.11+：

python3.11 -m venv .venv && source .venv/bin/activate

资料来源：README.md:45-50

必需依赖

资料来源：requirements.txt:1-8

环境配置问题

常见配置错误

环境变量优先级

1. PLAID_CLIENT_ID - 必须设置
2. PLAID_SECRET - 必须设置
3. PLAID_ENV - 默认为 "production"
4. PLAID_TOKEN_* - 每个链接银行一个

资料来源：plaid_client.py:35-45

诊断流程图

graph TD
    A[开始诊断] --> B{服务器启动成功?}
    B -->|否| C[检查环境变量配置]
    C --> D{PLAID_CLIENT_ID设置?}
    D -->|否| E[配置PLAID_CLIENT_ID]
    D -->|是| F{PLAID_SECRET设置?}
    F -->|否| G[配置PLAID_SECRET]
    F -->|是| H[检查端口占用]
    E --> A
    G --> A
    H --> I[启动成功]
    B -->|是| J{调用工具报错?}
    J -->|是| K[查看错误响应]
    K --> L{error字段存在?}
    L -->|是| M[根据错误代码处理]
    L -->|否| N{返回空数据?}
    N -->|是| O[使用get_institutions_status]
    O --> P{状态为healthy?}
    P -->|否| Q[重新链接账户]
    P -->|是| R[检查Token配置]
    J -->|否| S[功能正常]
    M --> T{code=ITEM_LOGIN_REQUIRED?}
    T -->|是| Q
    T -->|否| U{其他错误?}
    U -->|是| V[查看错误message]
    U -->|否| S

获取帮助

如遇到本文未覆盖的问题：

1. 检查服务器日志中的 trace_id
2. 查看 Plaid 仪表板的 API 日志
3. 使用 get_institutions_status 确认所有账户状态
4. 重启服务器以清除健康检查缓存

Pitfall Log / 踩坑日志

项目：josuem1109/personal-finance-mcp

摘要：发现 8 个潜在踩坑项，其中 1 个为 high/blocking；最高优先级：安全/权限坑 - 涉及密钥、隐私或敏感领域。

1. 安全/权限坑 · 涉及密钥、隐私或敏感领域

• 严重度：high
• 证据强度：source_linked
• 发现：项目文本出现 secret/private key/privacy/trading/finance 等敏感关键词。
• 对用户的影响：金融、交易、隐私和密钥场景必须比普通工具更保守。
• 建议检查：补敏感数据流、密钥存储和权限边界审查。
• 防护动作：敏感领域或密钥场景必须保守推荐并要求人工复核。
• 证据：packet_text.keyword_scan | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | matched secret / private key / privacy / trading / finance keyword

2. 安装坑 · 依赖 Docker 环境

• 严重度：medium
• 证据强度：runtime_trace
• 发现：安装/运行入口包含 Docker 命令：docker run ghcr.io/josuem1109/personal-finance-mcp:1.0.0
• 对用户的影响：非工程用户可能没有 Docker，启动成本明显增加。
• 建议检查：标注 Docker 前置条件，并提供非 Docker 路径或失败提示。
• 复现命令：docker run ghcr.io/josuem1109/personal-finance-mcp:1.0.0
• 防护动作：Docker 前置条件未说明时，不把项目标成普通用户低门槛。
• 证据：identity.distribution | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | docker run ghcr.io/josuem1109/personal-finance-mcp:1.0.0

3. 能力坑 · 能力判断依赖假设

• 严重度：medium
• 证据强度：source_linked
• 发现：README/documentation is current enough for a first validation pass.
• 对用户的影响：假设不成立时，用户拿不到承诺的能力。
• 建议检查：将假设转成下游验证清单。
• 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
• 证据：capability.assumptions | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | README/documentation is current enough for a first validation pass.

4. 维护坑 · 维护活跃度未知

• 严重度：medium
• 证据强度：source_linked
• 发现：未记录 last_activity_observed。
• 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
• 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
• 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
• 证据：evidence.maintainer_signals | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | last_activity_observed missing

5. 安全/权限坑 · 下游验证发现风险项

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：下游已经要求复核，不能在页面中弱化。
• 建议检查：进入安全/权限治理复核队列。
• 防护动作：下游风险存在时必须保持 review/recommendation 降级。
• 证据：downstream_validation.risk_items | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | no_demo; severity=medium

6. 安全/权限坑 · 存在评分风险

• 严重度：medium
• 证据强度：source_linked
• 发现：no_demo
• 对用户的影响：风险会影响是否适合普通用户安装。
• 建议检查：把风险写入边界卡，并确认是否需要人工复核。
• 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
• 证据：risks.scoring_risks | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | no_demo; severity=medium

7. 维护坑 · issue/PR 响应质量未知

• 严重度：low
• 证据强度：source_linked
• 发现：issue_or_pr_quality=unknown。
• 对用户的影响：用户无法判断遇到问题后是否有人维护。
• 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
• 防护动作：issue/PR 响应未知时，必须提示维护风险。
• 证据：evidence.maintainer_signals | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | issue_or_pr_quality=unknown

8. 维护坑 · 发布节奏不明确

• 严重度：low
• 证据强度：source_linked
• 发现：release_recency=unknown。
• 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
• 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
• 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
• 证据：evidence.maintainer_signals | mcp_registry:io.github.JosueM1109/personal-finance-mcp:1.0.0 | https://registry.modelcontextprotocol.io/v0.1/servers/io.github.JosueM1109%2Fpersonal-finance-mcp/versions/1.0.0 | release_recency=unknown