03 · Admin SDK 实战(Node.js + Python)

目标读者:要在 Node.js 或 Python 里发 FCM 推送的开发者 核心问题:怎么初始化?怎么发第一条?多 Firebase 项目怎么一起管?(本节最重要) 前置: 01-FCM核心概念 / 02-HTTP-v1-API与认证 后续: 04-批量发送与广播模式 最后更新:2026-04-11

---

一、为什么用 Admin SDK 而不是裸 HTTP

维度	裸 HTTP	Admin SDK
Token 刷新	自己写	自动
重试 / 退避	自己写	内置(5xx 自动重试 1 次)
类型检查	无	TypeScript / Python typing
多 App	自己管理	initializeApp(cred, name) 原生支持
Topic 订阅 API	自己 POST	subscribeToTopic() 一行
批量发送	循环 + 并发管理	sendEachForMulticast() 一行
行数对比	50+	5

结论:除非环境不支持(Cloudflare Worker / Deno 等),永远用 Admin SDK。

---

二、Node.js 完整实战

2.1 安装

npm install firebase-admin

要求: - Node.js ≥ 18 - firebase-admin 12.x(2026 年当前主流)

2.2 初始化 —— 方式 1:显式传 Service Account 文件

// init-explicit.js
const admin = require('firebase-admin');
const serviceAccount = require('./service-account.json');

admin.initializeApp({
  credential: admin.credential.cert(serviceAccount),
});

2.3 初始化 —— 方式 2:环境变量 (ADC)

export GOOGLE_APPLICATION_CREDENTIALS="/secure/path/service-account.json"

// init-env.js
const admin = require('firebase-admin');
admin.initializeApp(); // 自动读环境变量

两种方式对比:

维度	环境变量 GOOGLE_APPLICATION_CREDENTIALS	显式 cert(path)
多项目	❌ 只能指 1 个	✅ 每个项目独立传
容器 / CI	✅ 简单	一样
Cloud Run / GCE	✅ 自动用 metadata	不必要
本地开发	✅ 一行 export	✅ 写死路径
运营多平台脚本	❌	✅ 唯一选择

2.4 最小发送示例(单 token)

// send-single.js
const admin = require('firebase-admin');
admin.initializeApp({
  credential: admin.credential.cert(require('./service-account.json')),
});

async function main() {
  const message = {
    token: 'eXaMpLe_FCM_ToKeN_HeRe',
    notification: {
      title: '限时优惠码 SUPER88',
      body: '今晚 11 点前充值送 88% 红利',
    },
    data: {
      promo_code: 'SUPER88',
      deeplink: 'myapp://promo/super88',
    },
  };

  const response = await admin.messaging().send(message);
  console.log('成功,消息 ID:', response);
}

main().catch(console.error);

2.5 发送给 Topic(运营批量推送的核心模式)

// send-topic.js
const admin = require('firebase-admin');
admin.initializeApp({
  credential: admin.credential.cert(require('./service-account.json')),
});

const message = {
  topic: 'all_users',
  notification: {
    title: '【晚间福利】优惠码 NIGHT99',
    body: '20:00-23:00 充值即享额外 99 元红包',
  },
  data: {
    promo_code: 'NIGHT99',
    campaign: '2026-04-11-night',
  },
  android: {
    priority: 'HIGH',
    notification: { channel_id: 'promo', sound: 'default' },
  },
  apns: {
    headers: { 'apns-priority': '10' },
    payload: { aps: { sound: 'default', badge: 1 } },
  },
};

admin.messaging().send(message)
  .then(id => console.log('发送成功,消息 ID:', id))
  .catch(err => console.error('发送失败:', err));

这一段代码就是未来自动化脚本的核心 —— 把它包进 cron,每天 5 次定时跑,你就完成了单平台的推送自动化。

2.6 批量发送(明确 token 列表,≤500 个)

// send-multicast.js
const admin = require('firebase-admin');
admin.initializeApp({
  credential: admin.credential.cert(require('./service-account.json')),
});

async function sendBatch(tokens) {
  const message = {
    notification: { title: 'Promo', body: 'NIGHT99' },
    tokens, // ≤500
  };
  const response = await admin.messaging().sendEachForMulticast(message);
  console.log(`成功 ${response.successCount}, 失败 ${response.failureCount}`);

  // 收集失败的 token
  const failedTokens = [];
  response.responses.forEach((resp, idx) => {
    if (!resp.success) {
      console.error(`Token ${tokens[idx]} 失败:`, resp.error.code);
      // 如果是 UNREGISTERED 或 NOT_FOUND,必须从数据库删除
      if (
        resp.error.code === 'messaging/registration-token-not-registered' ||
        resp.error.code === 'messaging/invalid-registration-token'
      ) {
        failedTokens.push(tokens[idx]);
      }
    }
  });
  return failedTokens;
}

(async () => {
  const deadTokens = await sendBatch(['token1', 'token2', /* ... */]);
  console.log('需要从数据库删除的 token:', deadTokens);
})();

2.7 ⭐ 多 Firebase 项目初始化(用户多平台场景的核心)

这是全系列最重要的一段代码。iGaming 运营的场景是"多个平台 = 多个 Firebase 项目",必须用 Admin SDK 的多 app 实例功能。

// multi-project.js
// 作者: Bob · 2026-04-11
// 目标: 一次运行,同时给 N 个平台的 all_users topic 发送相同内容的推送

const admin = require('firebase-admin');

const platforms = [
  { name: 'platform-a', saPath: './secrets/platform-a-sa.json' },
  { name: 'platform-b', saPath: './secrets/platform-b-sa.json' },
  { name: 'platform-c', saPath: './secrets/platform-c-sa.json' },
  // ... 按实际平台数量追加
];

// Step 1: 为每个平台初始化独立的 app 实例
const apps = {};
for (const p of platforms) {
  apps[p.name] = admin.initializeApp(
    {
      credential: admin.credential.cert(require(p.saPath)),
    },
    p.name, // ⚠️ 关键: 第二个参数是 app name, 必须唯一
  );
}

// Step 2: 对每个平台并行推送
async function pushToAllPlatforms(payload) {
  const results = await Promise.all(
    platforms.map(async (p) => {
      try {
        const messaging = admin.messaging(apps[p.name]);
        const id = await messaging.send({
          topic: 'all_users',
          ...payload,
        });
        return { platform: p.name, ok: true, id };
      } catch (err) {
        return { platform: p.name, ok: false, error: err.message };
      }
    }),
  );
  console.table(results);
  return results;
}

// Step 3: 构造 payload 并发送
pushToAllPlatforms({
  notification: {
    title: '【11 点推送】优惠码 LUCKY11',
    body: '点击立即领取,限量 1000 份',
  },
  data: {
    promo_code: 'LUCKY11',
    deeplink: 'myapp://promo/LUCKY11',
  },
  android: {
    priority: 'HIGH',
    notification: { channel_id: 'promo', sound: 'default' },
  },
  apns: {
    headers: { 'apns-priority': '10' },
    payload: { aps: { sound: 'default', badge: 1 } },
  },
}).catch(console.error);

关键点:

1. initializeApp(opts, name) 第二个参数必须每个项目唯一 不传就是 [DEFAULT],第二次调用会抛 app/duplicate-app

2. 从特定 app 拿 service:admin.messaging(apps['platform-a']) 不要写 admin.messaging() —— 那只拿 default app

3. 一个进程内可以同时维护 N 个 Firebase 项目 每个有自己的 token 缓存、HTTP agent

4. 错误隔离 用 Promise.all + try/catch,一个平台失败不会阻塞其它

2.8 Topic 订阅(服务端批量订阅)

// subscribe-all.js
// 每天/每周一次,把所有玩家 token 订阅到 'all_users' topic
async function subscribeAll(messaging, tokens) {
  // 单次最多 1000 个
  for (let i = 0; i < tokens.length; i += 1000) {
    const batch = tokens.slice(i, i + 1000);
    const res = await messaging.subscribeToTopic(batch, 'all_users');
    console.log(`订阅 ${batch.length}: 成功 ${res.successCount}, 失败 ${res.failureCount}`);
  }
}

---

三、Python 完整实战

3.1 安装

pip install firebase-admin

要求: - Python ≥ 3.7 - firebase-admin 6.x(2026 年当前主流)

3.2 初始化 —— 显式 Service Account

import firebase_admin
from firebase_admin import credentials, messaging

cred = credentials.Certificate('./service-account.json')
firebase_admin.initialize_app(cred)

3.3 初始化 —— 环境变量 ADC

export GOOGLE_APPLICATION_CREDENTIALS="/secure/path/service-account.json"

import firebase_admin
firebase_admin.initialize_app()  # 自动 ADC

3.4 最小发送示例

# send_single.py
import firebase_admin
from firebase_admin import credentials, messaging

cred = credentials.Certificate('./service-account.json')
firebase_admin.initialize_app(cred)

message = messaging.Message(
    token='eXaMpLe_FCM_ToKeN_HeRe',
    notification=messaging.Notification(
        title='限时优惠码 SUPER88',
        body='今晚 11 点前充值送 88% 红利',
    ),
    data={
        'promo_code': 'SUPER88',
        'deeplink': 'myapp://promo/super88',
    },
    android=messaging.AndroidConfig(
        priority='high',
        notification=messaging.AndroidNotification(
            channel_id='promo',
            sound='default',
        ),
    ),
    apns=messaging.APNSConfig(
        headers={'apns-priority': '10'},
        payload=messaging.APNSPayload(
            aps=messaging.Aps(sound='default', badge=1),
        ),
    ),
)

response = messaging.send(message)
print('成功,消息 ID:', response)

3.5 发送给 Topic

# send_topic.py
message = messaging.Message(
    topic='all_users',
    notification=messaging.Notification(
        title='【晚间福利】优惠码 NIGHT99',
        body='20:00-23:00 充值即享额外 99 元红包',
    ),
    data={'promo_code': 'NIGHT99'},
)
messaging.send(message)

3.6 批量发送(明确 token 列表)

# send_multicast.py
message = messaging.MulticastMessage(
    tokens=['token1', 'token2'],  # ≤500
    notification=messaging.Notification(title='Promo', body='NIGHT99'),
)
response = messaging.send_each_for_multicast(message)
print(f'成功 {response.success_count}, 失败 {response.failure_count}')

# 收集失败的 token
failed_tokens = []
for idx, resp in enumerate(response.responses):
    if not resp.success:
        if isinstance(resp.exception, messaging.UnregisteredError):
            failed_tokens.append(message.tokens[idx])
print('需要从数据库删除的 token:', failed_tokens)

3.7 ⭐ 多 Firebase 项目初始化(Python 版)

# multi_project.py
# 作者: Bob · 2026-04-11

import firebase_admin
from firebase_admin import credentials, messaging

PLATFORMS = [
    {'name': 'platform-a', 'sa_path': './secrets/platform-a-sa.json'},
    {'name': 'platform-b', 'sa_path': './secrets/platform-b-sa.json'},
    {'name': 'platform-c', 'sa_path': './secrets/platform-c-sa.json'},
]

# Step 1: 为每个平台初始化独立的 app
apps = {}
for p in PLATFORMS:
    cred = credentials.Certificate(p['sa_path'])
    apps[p['name']] = firebase_admin.initialize_app(
        cred,
        name=p['name'],  # ⚠️ 第二个参数: app 名,必须唯一
    )

def push_to_all(title: str, body: str, data: dict | None = None):
    """对所有平台串行推送(并行版本见 Team B 文档)"""
    results = []
    for p in PLATFORMS:
        try:
            msg = messaging.Message(
                topic='all_users',
                notification=messaging.Notification(title=title, body=body),
                data=data or {},
                android=messaging.AndroidConfig(
                    priority='high',
                    notification=messaging.AndroidNotification(
                        channel_id='promo',
                        sound='default',
                    ),
                ),
                apns=messaging.APNSConfig(
                    headers={'apns-priority': '10'},
                    payload=messaging.APNSPayload(
                        aps=messaging.Aps(sound='default', badge=1),
                    ),
                ),
            )
            # ⚠️ 关键: 把 app 实例传给 send()
            msg_id = messaging.send(msg, app=apps[p['name']])
            results.append({'platform': p['name'], 'ok': True, 'id': msg_id})
        except Exception as e:
            results.append({'platform': p['name'], 'ok': False, 'error': str(e)})
    return results

if __name__ == '__main__':
    rs = push_to_all(
        title='【11 点推送】优惠码 LUCKY11',
        body='点击立即领取,限量 1000 份',
        data={'promo_code': 'LUCKY11', 'deeplink': 'myapp://promo/LUCKY11'},
    )
    for r in rs:
        print(r)

关键点(Python 版):

1. firebase_admin.initialize_app(cred, name='xxx') 第二个参数是 app 名,必须唯一

2. 调用 service 时通过 app= 参数指定项目

   messaging.send(msg, app=apps['platform-a'])
   

   不传 app= 会用 default —— 这是最常见的 bug!

3. 取已初始化的 app 实例:

   firebase_admin.get_app('platform-a')
   

3.8 Topic 订阅(Python 版)

def subscribe_all_to_topic(tokens: list[str], topic: str, app):
    """批量订阅,每次最多 1000 个"""
    for i in range(0, len(tokens), 1000):
        batch = tokens[i:i+1000]
        resp = messaging.subscribe_to_topic(batch, topic, app=app)
        print(f'订阅 {len(batch)}: 成功 {resp.success_count}')

---

四、密钥的安全管理

按从弱到强排列:

4.1 .env + dotenv(本地开发)

# .env
FIREBASE_SA_PATH_PLATFORM_A=/secure/path/platform-a-sa.json
FIREBASE_SA_PATH_PLATFORM_B=/secure/path/platform-b-sa.json

务必加 .gitignore:

.env
secrets/
*-sa.json
service-account*.json

4.2 OS Keychain(单机自动化)

macOS Keychain、Windows Credential Manager、Linux secret-tool。

4.3 HashiCorp Vault(自建服务器)

按需拉取,支持动态密钥轮换。

4.4 Google Cloud Secret Manager(推荐) ⭐

把 JSON 内容存为 secret,运行时拉取:

from google.cloud import secretmanager
import json
import firebase_admin
from firebase_admin import credentials

def load_sa(secret_id: str, project: str):
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project}/secrets/{secret_id}/versions/latest"
    payload = client.access_secret_version(
        request={"name": name}
    ).payload.data.decode()
    return credentials.Certificate(json.loads(payload))

# 使用
cred = load_sa('platform-a-firebase-sa', 'my-ops-project')
firebase_admin.initialize_app(cred, name='platform-a')

4.5 AWS Secrets Manager / Azure Key Vault

跨云场景。

4.6 ⛔ 绝对禁止

• ❌ commit JSON 到任何 git 仓库(包括 private repo)
• ❌ 贴到聊天工具(Telegram / Slack / WhatsApp)、邮件
• ❌ 放在前端 / 移动端代码里
• ❌ 写在 Docker 镜像层(用 BuildKit --mount=type=secret)

---

五、陷阱清单(先看再写代码)

1. sendMulticast / sendAll 已过时 firebase-admin Node ≥ 11 推荐用 sendEachForMulticast / sendEach。新代码统一用 sendEachForMulticast。 Python 没有 send_multicast,用 send_each_for_multicast。

2. 多 app 漏 app= 参数(Python) messaging.send(msg) 不传 app= 会用 default,结果发到错的项目。务必传 app=。

3. initializeApp 重复调用 第二次调用同名 app 会抛 app/duplicate-app。脚本启动时初始化一次就好。

4. Service Account 的 project_id 必须和 Firebase Project ID 完全一致 如果重命名过项目,旧 SA 还能用但容易混淆。

5. firebase-admin 默认有重试 网络抖动 5xx 会自动重试 1 次,4xx 不重试(需要自己处理)。

6. 进程退出前清理(长生命周期服务必做,一次性脚本可忽略)

   await Promise.all(Object.values(apps).map(a => a.delete()));
   

7. 限频自处理 Admin SDK 不会自动 rate-limit。如果同时给 N 个项目发巨量消息,自己加并发上限(Promise.all + 限流 / asyncio.Semaphore)。

8. 超过 500 token 用 sendEachForMulticast 会报错 自己分批:

   for (let i = 0; i < tokens.length; i += 500) {
     await messaging.sendEachForMulticast({
       ...msg,
       tokens: tokens.slice(i, i + 500),
     });
   }
   

---

六、最小运营脚本骨架(拼在一起)

// operations-push.js
// 骨架: 定时到点 → 构造 payload → 对所有平台并行推送
const admin = require('firebase-admin');

const platforms = require('./platforms.config.js'); // 你的平台配置

// 初始化所有 app
const apps = {};
for (const p of platforms) {
  apps[p.name] = admin.initializeApp(
    { credential: admin.credential.cert(require(p.saPath)) },
    p.name,
  );
}

// 时间槽 → 文案 + 优惠码
const SLOT_MESSAGES = {
  '11:00': { title: '早安福利', body: '午休前兑换 {{code}}', code: 'MORNING11' },
  '15:00': { title: '下午茶时间', body: '一杯咖啡的功夫,福利到手 {{code}}', code: 'TEA15' },
  '19:00': { title: '晚间首波', body: '开局专享 {{code}}', code: 'EVE19' },
  '21:00': { title: '黄金时段', body: '充多少送多少 {{code}}', code: 'GOLD21' },
  '00:00': { title: '午夜福利', body: '今夜最后一波 {{code}}', code: 'NIGHT00' },
};

async function pushAll(slotKey) {
  const cfg = SLOT_MESSAGES[slotKey];
  if (!cfg) throw new Error(`Unknown slot: ${slotKey}`);

  const payload = {
    notification: {
      title: cfg.title,
      body: cfg.body.replace('{{code}}', cfg.code),
    },
    data: { promo_code: cfg.code, slot: slotKey },
    android: {
      priority: 'HIGH',
      notification: { channel_id: 'promo', sound: 'default' },
    },
    apns: {
      headers: { 'apns-priority': '10' },
      payload: { aps: { sound: 'default', badge: 1 } },
    },
  };

  const results = await Promise.allSettled(
    platforms.map(p =>
      admin.messaging(apps[p.name]).send({ topic: 'all_users', ...payload })
        .then(id => ({ platform: p.name, ok: true, id }))
        .catch(err => ({ platform: p.name, ok: false, error: err.message })),
    ),
  );

  const summary = results.map(r => r.status === 'fulfilled' ? r.value : r.reason);
  console.table(summary);
  return summary;
}

// 入口:从命令行参数拿 slot
const slot = process.argv[2]; // 例: "11:00"
pushAll(slot).catch(console.error);

配合 cron:

# crontab -e (UTC)
# 印尼时间 UTC+7 → UTC 减 7 小时
0 4  * * * node /path/to/operations-push.js 11:00
0 8  * * * node /path/to/operations-push.js 15:00
0 12 * * * node /path/to/operations-push.js 19:00
0 14 * * * node /path/to/operations-push.js 21:00
0 17 * * * node /path/to/operations-push.js 00:00

这就是自动化的核心 MVP。后续的进阶(错误隔离、重试、观测、动态优惠码)见 08-iGaming多平台批量推送实战。

---

七、权威来源

• https://firebase.google.com/docs/admin/setup
• https://firebase.google.com/docs/cloud-messaging/admin/send-messages
• https://firebase.google.com/docs/reference/admin/node/firebase-admin.messaging
• https://firebase.google.com/docs/reference/admin/python/firebase_admin.messaging
• https://github.com/firebase/firebase-admin-node
• https://github.com/firebase/firebase-admin-python