用 Scalar Agent SDK 打造每日账单催缴后台 AgentStripe 逾期发票 × Resend 邮件提醒实战指南【免费下载链接】scalarScalar is an open-source API platform: Modern REST API Client Beautiful API References ✨ 1st-Class OpenAPI/Swagger Support项目地址: https://gitcode.com/GitHub_Trending/sc/scalar本篇技术指南基于 Scalar 开源仓库中的 Agent 文档documentation/guides/agent/examples/billing-sweep.md撰写讲解如何用 Scalar 的 Agent SDK 构建一个每天自动运行的账单催缴后台任务通过 Scalar 托管的 MCP 服务器读取 Stripe 中已过期的未结发票再调用 Resend 向客户发送付款提醒邮件。读完本文你将掌握从「Scalar 控制台配置 MCP 工具」到「TypeScript / Python 双语言 Agent 编码」再到「crontab / GitHub Actions 定时调度」的完整实战链路并理解其底层的 Installation MCP 与工具执行机制。整体思路一个脚本、两个 API、零人工跟进Billing Sweep 的核心价值在于把「查逾期发票」与「发提醒邮件」这两个本来需要人工或两套脚本分别处理的环节交给一个携带工具tools的 Agent 一次性完成Agent 通过 Scalar 的 MCP 服务器拿到 Stripe 与 Resend 两个 API 的能力模型按照自然语言任务清单自行决定调用顺序与参数脚本可以手动运行、放进 cron 定时执行也可以由 Stripe Webhook 触发。整个过程只有「一个脚本」却横跨「两个 API」最终实现零人工跟进。这一模式与 Scalar 对 Agent 的定位一致——Scalar 位于你的 API 定义与 Agent 运行时之间通过 MCP 把 OpenAPI 文档变成 Agent 可调用的工具无需自写封装参见 快速开始。前置条件在开始前需要准备以下资源资源说明备注运行时Node.js 18TypeScript 方案或 Python 3.10Python 方案二选一即可Scalar Personal Token用于认证对 Scalar MCP 的请求在 Scalar 控制台 Account API Keys 创建Scalar Installation ID标识你的 MCP 安装实例从控制台 SDK 选项卡复制OpenAI API Key驱动大模型示例使用gpt-4oTypeScript 方案需配置OPENAI_API_KEYStripe 账户提供发票数据测试模式test mode即可Resend 账户发送提醒邮件需要有已验证的发信域名关于 Personal Token 的创建方式Agent SDK 文档 中有明确说明在 Dashboard 的Account API Keys下创建个人访问令牌。项目初始化TypeScript 方案mkdir billing-sweep cd billing-sweep npm init -y npm install scalar/agent ai ai-sdk/openai dotenv tsx依赖说明scalar/agentScalar 官方 Agent SDK为 Vercel AI SDK、OpenAI Agents SDK、Anthropic Claude Agent SDK 提供原生集成aiVercel AI SDK提供generateText与stepCountIs等核心能力ai-sdk/openaiOpenAI 模型提供方dotenv加载.env环境变量tsx直接运行 TypeScript 文件无需先编译。Python 方案Python 方案基于 OpenAI Agents SDK配合scalar-agent使用MCP 的接入方式与 TypeScript 完全一致mkdir billing-sweep cd billing-sweep python3 -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate pip install scalar-agent[openai] python-dotenv[openai]这个 extra 会同时安装scalar-agent与 OpenAI Agents SDK 所需的依赖。在 Agent SDK 文档 中还可以看到[anthropic]、[all]等其他 extra 的用法。环境变量把密钥写入.env文件SCALAR_TOKENyour-scalar-personal-token SCALAR_INSTALLATION_IDyour-installation-id OPENAI_API_KEYyour-openai-api-key这里只有三个变量前两个用于连接 Scalar MCPOPENAI_API_KEY用于访问 OpenAI 模型Python 方案下 OpenAI Agents SDK 会自动从环境读取它。注意Stripe 与 Resend 的 API Key不会出现在这里——它们被存放在 Scalar 控制台。在 Scalar 控制台配置 MCP把 Stripe 和 Resend 变成 Agent 工具在 Scalar 控制台中完成以下三步进入MCP → Add tool → Stripe在Authentication处粘贴你的 Stripe 密钥并点击Save然后为GET /v1/invoices启用Execute点击 Add tool → Resend粘贴 Resend API Key 并点击Save为POST /emails启用Execute从SDK选项卡复制你的Installation ID。这样配置之后你的 API 密钥存储在 Scalar 中永远不会出现在你的代码里——这正是 Installation MCP 的核心设计之一。关于工具模式MCP Servers 文档 给出了两种可选模式模式行为Search仅暴露端点供查询/检索不会向你的 API 发送真实请求Execute向你的 API 发出真实、已认证的请求本文的账单催缴场景需要真正读取发票、真正发邮件因此必须为GET /v1/invoices和POST /emails启用Execute模式。需要理解的是这里创建的 MCP 属于Installation MCP它位于独立的端点形如https://mcp.scalar.com/mcp/YOUR_INSTALL_ID默认私有团队成员使用 Personal Access Token 连接外部人员则需要通过你授予的 OAuth 流程详见 MCP Servers 文档。这也解释了为什么SCALAR_TOKEN是连接所必需的。API 认证方面控制台支持Global在安装实例上存一份凭据所有调用共用与Passthrough调用方每次自带凭据、Scalar 透传不存储两种模式本文的 Stripe / Resend 密钥属于 Global 模式。初始化客户端TypeScriptimport dotenv/config import { agentScalar } from scalar/agent import { generateText, stepCountIs } from ai import { openai } from ai-sdk/openai const scalar agentScalar({ token: process.env.SCALAR_TOKEN }) const model openai(gpt-4o)agentScalar是 SDK 的入口函数Agent SDK 文档 给出了它的配置项选项类型说明tokenstringScalar Personal Token用于认证对 MCP 的请求baseUrlstringScalar MCP 服务器的 Base URL默认指向 Scalar 环境Pythonimport os from dotenv import load_dotenv from scalar_agent import agent_scalar load_dotenv() scalar agent_scalar(tokenos.environ[SCALAR_TOKEN])Python 侧对应的参数为token与base_url语义与 TypeScript 完全一致。初始化完成后还需要通过 Installation ID 拿到「安装实例」引用scalar.installation(...)下一步的createVercelAITools()与create_openai_mcp()都挂在它上面。核心催缴逻辑sweep 函数拆解整个催缴逻辑封装在一个函数里先从安装实例获取工具集再把它交给generateText/Runner.run同时给模型一个清晰的任务清单。stopWhen: stepCountIs(20)让 Agent 可以在两个 API 之间按需多次调用工具直到完成或达到步数上限。TypeScript 实现async function sweep() { const installation await scalar.installation(process.env.SCALAR_INSTALLATION_ID) const tools await installation.createVercelAITools() const { text } await generateText({ model, tools, stopWhen: stepCountIs(20), system: You are a billing assistant with access to Stripe and Resend. Use Stripe to read invoice data. Use Resend to send emails. Todays date is ${new Date().toISOString().split(T)[0]}., prompt: Run a billing sweep: 1. Fetch all open Stripe invoices where due_date is before today. 2. For each overdue invoice, send a payment reminder email via Resend. - From: billingyourcompany.com - To: the customer_email on the invoice - Subject: Payment reminder — invoice #[number] - Body: friendly reminder with the amount due and hosted_invoice_url 3. Report how many invoices were found and how many emails were sent., }) console.log(text) } sweep()关键点installation.createVercelAITools()返回的是一套可直接用于generateText或streamText的工具集SDK 原生基于ai-sdk/mcp见 Agent SDK 文档system提示词声明角色与可用能力并把「今天的日期」注入进去——这是让模型正确判断due_date是否逾期所必需的prompt是任务清单三个步骤恰好对应「读 Stripe → 写 Resend → 汇总报告」模型会自主把每一步映射到对应工具调用stepCountIs(20)是 Vercel AI SDK 的终止条件辅助函数限制 Agent 的工具调用步数防止失控循环。Python 实现Python 侧使用MCPServerStreamableHttp建立 MCP 连接再用Runner.run驱动 Agent。max_turns的作用与stepCountIs(20)类似都是限制工具轮次上限import asyncio import os from datetime import date from agents import Agent, Runner from agents.mcp import MCPServerStreamableHttp async def sweep() - None: installation scalar.installation(os.environ[SCALAR_INSTALLATION_ID]) server MCPServerStreamableHttp(**installation.create_openai_mcp()) await server.connect() today date.today().isoformat() agent Agent( namebilling-sweep, instructions( You are a billing assistant with access to Stripe and Resend.\n Use Stripe to read invoice data. Use Resend to send emails.\n fTodays date is {today}. ), mcp_servers[server], ) result await Runner.run( agent, Run a billing sweep: 1. Fetch all open Stripe invoices where due_date is before today. 2. For each overdue invoice, send a payment reminder email via Resend. - From: billingyourcompany.com - To: the customer_email on the invoice - Subject: Payment reminder — invoice #[number] - Body: friendly reminder with the amount due and hosted_invoice_url 3. Report how many invoices were found and how many emails were sent., max_turns20, ) print(result.final_output) await server.cleanup() asyncio.run(sweep())与 TypeScript 的对应关系TypeScriptPython作用installation.createVercelAITools()installation.create_openai_mcp()生成 MCP 连接参数generateTextAgentRunner.run驱动模型执行任务stopWhen: stepCountIs(20)max_turns20限制工具调用轮次system提示词instructions设定 Agent 角色与约束console.log(text)print(result.final_output)输出执行报告—await server.cleanup()显式关闭 MCP 连接TS 侧由 SDK 内部管理installation.create_openai_mcp()返回的是可供MCPServerStreamableHttp直接展开的参数见 Agent SDK 文档Agent 运行时负责工具发现与执行。完整脚本TypeScriptsweep.tsimport dotenv/config import { agentScalar } from scalar/agent import { generateText, stepCountIs } from ai import { openai } from ai-sdk/openai const scalar agentScalar({ token: process.env.SCALAR_TOKEN }) const model openai(gpt-4o) async function sweep() { const installation await scalar.installation(process.env.SCALAR_INSTALLATION_ID) const tools await installation.createVercelAITools() const { text } await generateText({ model, tools, stopWhen: stepCountIs(20), system: You are a billing assistant with access to Stripe and Resend. Use Stripe to read invoice data. Use Resend to send emails. Todays date is ${new Date().toISOString().split(T)[0]}., prompt: Run a billing sweep: 1. Fetch all open Stripe invoices where due_date is before today. 2. For each overdue invoice, send a payment reminder email via Resend. - From: billingyourcompany.com - To: the customer_email on the invoice - Subject: Payment reminder — invoice #[number] - Body: friendly reminder with the amount due and hosted_invoice_url 3. Report how many invoices were found and how many emails were sent., }) console.log(text) } sweep()Pythonsweep.pyimport asyncio import os from datetime import date from agents import Agent, Runner from agents.mcp import MCPServerStreamableHttp from dotenv import load_dotenv from scalar_agent import agent_scalar load_dotenv() async def main() - None: scalar agent_scalar(tokenos.environ[SCALAR_TOKEN]) installation scalar.installation(os.environ[SCALAR_INSTALLATION_ID]) server MCPServerStreamableHttp(**installation.create_openai_mcp()) await server.connect() today date.today().isoformat() agent Agent( namebilling-sweep, instructions( You are a billing assistant with access to Stripe and Resend.\n Use Stripe to read invoice data. Use Resend to send emails.\n fTodays date is {today}. ), mcp_servers[server], ) result await Runner.run( agent, Run a billing sweep: 1. Fetch all open Stripe invoices where due_date is before today. 2. For each overdue invoice, send a payment reminder email via Resend. - From: billingyourcompany.com - To: the customer_email on the invoice - Subject: Payment reminder — invoice #[number] - Body: friendly reminder with the amount due and hosted_invoice_url 3. Report how many invoices were found and how many emails were sent., max_turns20, ) print(result.final_output) await server.cleanup() if __name__ __main__: asyncio.run(main())运行与输出示例npx tsx sweep.ts # or python sweep.py一个典型的输出如下Found 3 overdue invoices in Stripe: - jennyacme.com — $240.00 — 12 days overdue → reminder sent - opsglobex.com — $890.00 — 9 days overdue → reminder sent - financeinitech.com — $120.00 — 3 days overdue → reminder sent 3 invoices found. 3 reminder emails sent via Resend.注意输出内容由模型根据任务清单第 3 步汇总报告生成实际字段会随模型与数据变化但「找到了几张发票、发了几封邮件」的汇总结构是一致的。定时调度让催缴每天自动发生crontab每天上午 9 点crontab -e0 9 * * * cd /path/to/billing-sweep npx tsx sweep.ts sweep.log 21 # Python: 0 9 * * * cd /path/to/billing-sweep .venv/bin/python sweep.py sweep.log 21注意 Python 行使用了 venv 内的解释器路径.venv/bin/python确保定时任务环境中能找到依赖日志统一追加到sweep.log。GitHub ActionsTypeScript 版本.github/workflows/billing-sweep.ymlname: Billing Sweep on: schedule: - cron: 0 9 * * * # 9am UTC daily workflow_dispatch: # manual trigger jobs: sweep: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - uses: actions/setup-nodev4 with: node-version: 22 - run: npm install - run: npx tsx sweep.ts env: SCALAR_TOKEN: ${{ secrets.SCALAR_TOKEN }} SCALAR_INSTALLATION_ID: ${{ secrets.SCALAR_INSTALLATION_ID }} OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}Python 版本.github/workflows/billing-sweep-python.ymlname: Billing Sweep (Python) on: schedule: - cron: 0 9 * * * workflow_dispatch: jobs: sweep: runs-on: ubuntu-latest steps: - uses: actions/checkoutv4 - uses: actions/setup-pythonv5 with: python-version: 3.12 - run: pip install scalar-agent[openai] python-dotenv - run: python sweep.py env: SCALAR_TOKEN: ${{ secrets.SCALAR_TOKEN }} SCALAR_INSTALLATION_ID: ${{ secrets.SCALAR_INSTALLATION_ID }} OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}两个 workflow 都通过schedule的 cron 表达式0 9 * * *UTC 每天 9 点触发同时保留workflow_dispatch用于手动触发调试。三个敏感变量SCALAR_TOKEN、SCALAR_INSTALLATION_ID、OPENAI_API_KEY一律通过 GitHub Secrets 注入绝不写死在仓库里。进一步扩展原文档给出了三条自然的演进方向可以按业务需要组合使用接入 Notion把每次发出的提醒记录到 Notion 账单跟踪表中让财务团队有一个实时可见的视图改用 Stripe Webhook 替代 cron在invoice.payment_failed事件上触发 Agent把「每日批量催缴」升级为「实时单笔提醒」升级机制Escalation对逾期 30 天以上的发票不再重复发邮件而是为 accounts 团队创建一张 Jira 工单形成「提醒 → 升级」的分级处理链路。底层原理小结理解这套方案的关键在于「安装实例Installation」这一抽象你在控制台配置的 Stripe / Resend 工具与认证凭据都绑定在某个 Installation 上端点形如https://mcp.scalar.com/mcp/YOUR_INSTALL_ID而scalar.installation(INSTALLATION_ID)就是代码里对这个实例的引用。SDK 负责把它翻译成各 Agent 框架能消费的形式——Vercel AI SDK 的工具集createVercelAITools、OpenAI Agents SDK 的 MCP 服务器参数create_openai_mcp、Anthropic 的 MCP 配置createAnthropicMCP见 Agent SDK 文档。正因为上游 API 的密钥存储在 Scalar 侧、连接又要求 Personal Token 认证你的代码里才只需要SCALAR_TOKEN与SCALAR_INSTALLATION_ID两个变量即可安全地跨 API 编排任务。本文对应的文档与配套资料均位于仓库documentation/guides/agent/目录下可进一步阅读 Agent SDK、MCP Servers 与 快速开始以及同目录下的 incident-monitor.md、meeting-scheduler.md、revenue-dashboard.md 等其他 Agent 实战示例复用相同的 SDK 接入模式构建更多自动化场景。【免费下载链接】scalarScalar is an open-source API platform: Modern REST API Client Beautiful API References ✨ 1st-Class OpenAPI/Swagger Support项目地址: https://gitcode.com/GitHub_Trending/sc/scalar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考