List Accounts

Instructions

Implementation Reference

• server.py:40-63 (handler)

  Core implementation function _list_accounts_impl that iterates all linked Plaid Items, fetches accounts via Plaid API, shapes them, and returns a dict with accounts and warnings.

  def _list_accounts_impl() -> dict:
      """List every account across all linked Items, with balances.
  
      Returns:
          {"accounts": [...], "warnings": [...]}. Warnings describe Items that
          are unhealthy (re-auth required, etc.) or hit API errors on this call.
      """
      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}

• server.py:66-69 (registration)

  Registration of list_accounts as a FastMCP tool with read-only hint and title annotation, wrapping _list_accounts_impl.

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

• server.py:32-37 (helper)

  Helper _warning_from_health used by _list_accounts_impl to convert ItemHealth to a warning dict.

  def _warning_from_health(h: ItemHealth) -> dict:
      return {
          "institution": h.institution_name or h.env_key,
          "status": h.status,
          "reason": h.reason,
      }

• tests/test_server.py:8-45 (schema)

  Test verifying list_accounts aggregates accounts from healthy items and warns on unhealthy items.

  def test_list_accounts_aggregates_across_items(fake_env_tokens):
      fake_api = MagicMock()
      fake_api.accounts_get.return_value.to_dict.return_value = {
          "accounts": [
              {
                  "account_id": "a1",
                  "name": "Checking",
                  "mask": "0001",
                  "type": "depository",
                  "subtype": "checking",
                  "balances": {"current": 100, "available": 100, "iso_currency_code": "USD"},
              },
          ],
      }
      items = [
          ("CHASE", SecretStr("t"), ItemHealth("CHASE", "healthy", "ins_3", "Chase")),
          ("FIDELITY", SecretStr("t2"), ItemHealth("FIDELITY", "re_auth_required", "ins_9", "Fidelity", reason="ITEM_LOGIN_REQUIRED")),
      ]
      with patch.object(srv, "build_api", return_value=fake_api), \
           patch.object(srv, "all_items", return_value=items):
          out = srv._list_accounts_impl()
      assert len(out["accounts"]) == 1
      assert out["accounts"][0]["institution"] == "Chase"
      assert out["accounts"][0]["handle"] == "chase_checking_0001"
      assert len(out["warnings"]) == 1
      assert out["warnings"][0]["institution"] == "Fidelity"
      assert out["warnings"][0]["status"] == "re_auth_required"
  
  
  def test_list_accounts_surfaces_api_exception_as_warning(fake_env_tokens):
      from plaid.exceptions import ApiException
      fake_api = MagicMock()
      exc = ApiException(status=500, reason="boom")
      exc.body = '{"error_code":"INTERNAL_SERVER_ERROR","error_message":"plaid down"}'
      fake_api.accounts_get.side_effect = exc
      items = [("CHASE", SecretStr("t"), ItemHealth("CHASE", "healthy", "ins_3", "Chase"))]
      with patch.object(srv, "build_api", return_value=fake_api), \
           patch.object(srv, "all_items", return_value=items):

• tests/test_server.py:37-52 (schema)

  Test verifying list_accounts surfaces Plaid API exceptions as warnings.

  def test_list_accounts_surfaces_api_exception_as_warning(fake_env_tokens):
      from plaid.exceptions import ApiException
      fake_api = MagicMock()
      exc = ApiException(status=500, reason="boom")
      exc.body = '{"error_code":"INTERNAL_SERVER_ERROR","error_message":"plaid down"}'
      fake_api.accounts_get.side_effect = exc
      items = [("CHASE", SecretStr("t"), ItemHealth("CHASE", "healthy", "ins_3", "Chase"))]
      with patch.object(srv, "build_api", return_value=fake_api), \
           patch.object(srv, "all_items", return_value=items):
          out = srv._list_accounts_impl()
      assert out["accounts"] == []
      assert len(out["warnings"]) == 1
      w = out["warnings"][0]
      assert w["institution"] == "Chase"
      assert w["code"] == "INTERNAL_SERVER_ERROR"
      assert w["trace_id"]