别让AI随手就给你开通云服务

现在很多独立开发者和单人团队用AI写代码，效率确实高。但有一个隐形大坑：AI在生成代码时，动不动就会帮你接入Postgres、Redis、AWS S3或者云端邮件API。这些服务虽然可能有免费额度，但一旦数据量上去或触发某些条件，账单会悄咪咪地涨起来。更重要的是，云SDK会让你的应用变得复杂，调试和回退的成本极高。

如果你只是想在本周末快速跑通一个最小可行产品（MVP），根本不需要那些重型云服务。真正的开发效率不是靠AI打字快，而是靠严格限制AI能用的资源范围。

这篇文章分享一种“零账单发货法”：通过一个配置文件（能力票证）、一个严格拦截的工厂模式，以及只允许使用本地组件的设计原则，确保你写的每一行代码都不会偷偷连上付费云服务。适合想今天就把东西跑起来、且能接受初期功能限制的独立开发者。这不是生产环境扩容指南，而是早期冷启动的实战打法。

第一步：建立“能力票证”文件

所谓“票证”，其实就是给系统能力发许可证。每个许可证对应一个具体需求，绝不许隐藏任何付费客户端。在项目根目录创建一个 capabilities.json 文件，只列出允许使用的本地组件：

{
  "version": 1,
  "tickets": {
    "storage": "sqlite_file",
    "queue": "memory_list",
    "mail": "maildir",
    "blobs": "local_dir",
    "jobs": "inline",
    "auth": "hmac_cookie"
  },
  "data_dir": "./data",
  "listen": "127.0.0.1:8080"
}

这里面的取值故意做得很小、很死板，目的就是为了防作弊：
– sqlite_file：只用项目文件夹下的一个SQLite数据库文件
– memory_list：只使用进程内内存队列，重启即清空
– maildir：直接把邮件写成 .eml 文本文件保存在本地
– local_dir：把上传的文件哈希后存在本地磁盘
– inline：任务直接在当前请求线程里跑完，不单独起进程
– hmac_cookie：用本地密钥签名的Cookie做身份验证

任何超出上述范围的请求，都会被系统直接拒绝。拒绝发生在程序启动阶段，而不是等用户点击后才报错。这能保证你的应用从一开始就是干净、可控的。

第二周前的资源对照清单（原配置表转为列表）

在开发的第一周，建议把这张清单作为审核基准。任何未经人工修改就擅自引入新依赖的AI补丁，都应该被退回。

• 数据存储：允许 sqlite_file。禁止使用 Postgres 或托管MySQL。理由：只需一个文件，无需额外服务器。
• 消息队列：允许 memory_list。禁止使用 Redis、SQS 或 Celery。理由：随进程销毁，对于早期产品完全可以接受。
• 邮件服务：允许 maildir。禁止使用 SendGrid、SES 或 Postmark。理由：创始人可以直接打开本地的 .eml 文件查看内容。
• 文件存储：允许 local_dir。禁止使用 S3、GCS 或 R2。理由：哈希后的文件直接放在SQLite旁边。
• 后台任务：允许 inline。禁止使用云端调度器或Worker池。理由：不需要额外计费进程。
• 身份验证：允许 hmac_cookie。禁止使用 Auth0、Cognito 或 Clerk。理由：仅依赖本地密钥文件。
• 监听地址：允许 localhost 回路地址。禁止绑定 0.0.0.0 公网发布。理由：真实用户出现前，不需要暴露端口。

这份清单就是你的产品契约。下面的代码只负责严格执行它。

第三：构建“闭源式”拦截工厂

应用里永远不要临时去创建数据库连接或队列。统一交给一个“工厂”来创建。这个工厂一次性读取配置，如果发现不符合白名单的配置，直接抛出异常并终止启动。

# budget.py
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

# 定义允许的能力白名单，每种能力只能从固定集合中选择
ALLOWED = {
    "storage": {"sqlite_file"},
    "queue": {"memory_list"},
    "mail": {"maildir"},
    "blobs": {"local_dir"},
    "jobs": {"inline"},
    "auth": {"hmac_cookie"},
}

# 自定义预算超限异常类，用于拦截非法配置
class BudgetError(RuntimeError):
    pass

@dataclass(frozen=True)
class Tickets:
    storage: str
    queue: str
    mail: str
    blobs: str
    jobs: str
    auth: str
    data_dir: Path
    listen: str

def load_tickets(root: Path) -> Tickets:
    # 读取JSON配置文件
    raw = json.loads((root / "capabilities.json").read_text(encoding="utf-8"))
    tickets = raw["tickets"]

    # 逐一校验配置项是否在白名单内，不在则直接报错
    for key, allowed in ALLOWED.items():
        got = tickets.get(key)
        if got not in allowed:
            raise BudgetError(f"ticket {key}={got!r} is not in {sorted(allowed)}")

    # 解析数据存储路径，防止越权访问上级目录
    data_dir = (root / raw["data_dir"]).resolve()
    if not str(data_dir).startswith(str(root.resolve())):
        raise BudgetError("data_dir escapes the repo")

    # 确保目录存在
    data_dir.mkdir(parents=True, exist_ok=True)

    # 返回冻结的配置对象，后续代码无法篡改
    return Tickets(
        storage=tickets["storage"],
        queue=tickets["queue"],
        mail=tickets["mail"],
        blobs=tickets["blobs"],
        jobs=tickets["jobs"],
        auth=tickets["auth"],
        data_dir=data_dir,
        listen=raw["listen"],
    )

路径安全检查非常重要。如果AI试图把数据目录指向 /var/lib/postgresql 或远程挂载点，程序应该直接拒绝启动。这个工厂就是你的“财务审批台”，业务代码没有资格跟它讨价还价。

第四：全部使用“枯燥”的本地适配器

保持适配器的实现越枯燥越好。枯燥意味着稳定、易读、无外部依赖。

# adapters.py
from __future__ import annotations

import hashlib
import sqlite3
import time
from collections import deque
from pathlib import Path
from typing import Any, Callable, Deque, Dict, Tuple

from budget import BudgetError, Tickets

class Store:
    def __init__(self, tickets: Tickets) -> None:
        # 强制校验存储方式
        if tickets.storage != "sqlite_file":
            raise BudgetError("storage adapter refused")
        self.path = tickets.data_dir / "app.sqlite"
        self._db = sqlite3.connect(self.path, check_same_thread=False)
        self._db.execute(
            "CREATE TABLE IF NOT EXISTS notes "
            "(id INTEGER PRIMARY KEY, body TEXT NOT NULL, created_at REAL NOT NULL)"
        )
        self._db.commit()

    def add_note(self, body: str) -> int:
        cur = self._db.execute(
            "INSERT INTO notes(body, created_at) VALUES (?, ?)",
            (body, time.time()),
        )
        self._db.commit()
        return int(cur.lastrowid)

    def list_notes(self) -> list[tuple[int, str, float]]:
        rows = self._db.execute(
            "SELECT id, body, created_at FROM notes ORDER BY id DESC LIMIT 50"
        )
        return list(rows)

class MailDir:
    def __init__(self, tickets: Tickets) -> None:
        if tickets.mail != "maildir":
            raise BudgetError("mail adapter refused")
        self.dir = tickets.data_dir / "maildir"
        self.dir.mkdir(exist_ok=True)

    def send(self, to: str, subject: str, body: str) -> Path:
        name = f"{int(time.time() * 1000)}_{hashlib.sha256(to.encode()).hexdigest()[:8]}.eml"
        path = self.dir / name
        path.write_text(
            f"To: {to}\nSubject: {subject}\n\n{body}\n",
            encoding="utf-8",
        )
        return path

class MemoryQueue:
    def __init__(self, tickets: Tickets) -> None:
        if tickets.queue != "memory_list":
            raise BudgetError("queue adapter refused")
        self._q: Deque[Tuple[str, Dict[str, Any]]] = deque()

    def push(self, kind: str, payload: Dict[str, Any]) -> None:
        self._q.append((kind, payload))

    def drain(self, handler: Callable[[str, Dict[str, Any]], None]) -> int:
        n = 0
        while self._q:
            kind, payload = self._q.popleft()
            handler(kind, payload)
            n += 1
        return n

class BlobDir:
    def __init__(self, tickets: Tickets) -> None:
        if tickets.blobs != "local_dir":
            raise BudgetError("blob adapter refused")
        self.dir = tickets.data_dir / "blobs"
        self.dir.mkdir(exist_ok=True)

    def put(self, raw: bytes) -> str:
        digest = hashlib.sha256(raw).hexdigest()
        (self.dir / digest).write_bytes(raw)
        return digest

任务处理采用“同步阻塞”模式：HTTP处理器将任务放入内存队列，然后在返回响应前立刻消费掉。这在并发量大时会更慢，但对于早期几乎不存在流量的产品来说，它是免费的、可审计的，完全够用。

第五：用最基础的HTTP服务展示功能

第一版接口根本不需要复杂的ASGI框架。Python标准库的HTTP服务就足以证明你的架构能跑通。

# app.py
from __future__ import annotations

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

from adapters import MailDir, MemoryQueue, Store
from budget import load_tickets

ROOT = Path(__file__).resolve().parent
# 全局加载一次配置和组件
TICKETS = load_tickets(ROOT)
STORE = Store(TICKETS)
MAIL = MailDir(TICKETS)
QUEUE = MemoryQueue(TICKETS)

def run_jobs(kind: str, payload: dict) -> None:
    if kind == "note_mail":
        MAIL.send(payload["to"], "Note saved", payload["body"])

class Handler(BaseHTTPRequestHandler):
    def _json(self, code: int, payload: dict) -> None:
        raw = json.dumps(payload).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)

    def do_GET(self) -> None:
        if urlparse(self.path).path != "/notes":
            self._json(404, {"error": "not_found"})
            return
        rows = [
            {"id": i, "body": body, "created_at": ts}
            for i, body, ts in STORE.list_notes()
        ]
        self._json(200, {"listen": TICKETS.listen, "notes": rows})

    def do_POST(self) -> None:
        if urlparse(self.path).path != "/notes":
            self._json(404, {"error": "not_found"})
            return
        length = int(self.headers.get("Content-Length", "0"))
        data = json.loads(self.rfile.read(length) or b"{}")
        body = str(data.get("body", "")).strip()
        to = str(data.get("notify", "")).strip()
        if not body:
            self._json(400, {"error": "body_required"})
            return
        note_id = STORE.add_note(body)
        if to:
            QUEUE.push("note_mail", {"to": to, "body": body})
            QUEUE.drain(run_jobs)
        self._json(201, {"id": note_id})

    def log_message(self, fmt: str, *args) -> None:
        return  # 关闭默认日志输出，保持终端整洁

def main() -> None:
    host, port_s = TICKETS.listen.split(":")
    # 强制要求只能监听本地回环地址
    if host not in {"127.0.0.1", "localhost"}:
        raise SystemExit("listen host is not loopback; ticket refused")
    httpd = ThreadingHTTPServer((host, int(port_s)), Handler)
    httpd.serve_forever()

if __name__ == "__main__":
    main()

监听地址也是票证的一部分。绑定 0.0.0.0 属于后期决策，必须在真实用户出现后，由人决定开放，绝不能因为AI建议开启Docker端口就直接放行。

第六：编写“发货证明”脚本

能不能发货不能靠感觉，得看文件。验证脚本会启动服务，写入一条记录，然后检查磁盘上是否成功生成了SQLite文件和邮件文本。

# prove_ship.py
from __future__ import annotations

import json
import socket
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent

def wait_port(host: str, port: int, timeout: float = 5.0) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with socket.create_connection((host, port), timeout=0.2):
                return
        except OSError:
            time.sleep(0.05)
    raise SystemExit("server did not bind loopback in time")

def main() -> None:
    tickets = json.loads((ROOT / "capabilities.json").read_text(encoding="utf-8"))
    host, port_s = tickets["listen"].split(":")
    port = int(port_s)
    proc = subprocess.Popen([sys.executable, str(ROOT / "app.py")], cwd=ROOT)
    try:
        wait_port(host, port)
        # 发送POST请求创建笔记
        req = urllib.request.Request(
            f"http://{host}:{port}/notes",
            data=json.dumps(
                {"body": "ship today", "notify": "founder@localhost"}
            ).encode(),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=2) as resp:
            created = json.loads(resp.read().decode())
        # 发送GET请求验证列表返回正常
        with urllib.request.urlopen(f"http://{host}:{port}/notes", timeout=2) as resp:
            listed = json.loads(resp.read().decode())
    finally:
        proc.terminate()
        proc.wait(timeout=3)

    sqlite_path = ROOT / "data" / "app.sqlite"
    mails = list((ROOT / "data" / "maildir").glob("*.eml"))
    # 生成验证报告
    proof = {
        "ok": True,
        "note_id": created["id"],
        "listed": len(listed["notes"]),
        "sqlite_bytes": sqlite_path.stat().st_size,
        "mail_files": len(mails),
        "listen": tickets["listen"],
        "tickets": tickets["tickets"],
    }
    # 核心断言：必须有数据落盘才算成功
    if proof["listed"] < 1 or proof["mail_files"] < 1:
        raise SystemExit("ship proof failed")
    out = ROOT / "SHIP_PROOF.json"
    out.write_text(json.dumps(proof, indent=2), encoding="utf-8")
    print(out.read_text(encoding="utf-8"))

if __name__ == "__main__":
    main()

在项目根目录运行此脚本：

python prove_ship.py

运行成功后会生成 SHIP_PROOF.json。把这个文件连同票证文件一起提交到版本控制。下一次使用AI辅助时，要把这两个文件当作“必须遵守的输入约束”，而不是可有可无的建议。

第七：用单元测试守卫配置红线

为了防止AI在改代码时悄悄扩大白名单，需要加入自动化检查，否则CI流水线会失去把关作用。

# test_budget.py
from pathlib import Path
import json
import tempfile
import unittest

from budget import BudgetError, load_tickets

class TicketTests(unittest.TestCase):
    # 辅助方法：动态生成测试用的配置文件目录
    def write(self, tickets, data_dir="./data"):
        td = Path(tempfile.mkdtemp())
        payload = {
            "version": 1,
            "tickets": tickets,
            "data_dir": data_dir,
            "listen": "127.0.0.1:8080",
        }
        (td / "capabilities.json").write_text(json.dumps(payload), encoding="utf-8")
        return td

    def test_accepts_local_set(self):
        root = self.write({
            "storage": "sqlite_file",
            "queue": "memory_list",
            "mail": "maildir",
            "blobs": "local_dir",
            "jobs": "inline",
            "auth": "hmac_cookie",
        })
        loaded = load_tickets(root)
        self.assertEqual(loaded.storage, "sqlite_file")

    def test_rejects_postgres(self):
        root = self.write({
            "storage": "postgres",  # 模拟AI错误引入云数据库
            "queue": "memory_list",
            "mail": "maildir",
            "blobs": "local_dir",
            "jobs": "inline",
            "auth": "hmac_cookie",
        })
        with self.assertRaises(BudgetError):
            load_tickets(root)

    def test_rejects_escaped_data_dir(self):
        root = self.write({
            "storage": "sqlite_file",
            "queue": "memory_list",
            "mail": "maildir",
            "blobs": "local_dir",
            "jobs": "inline",
            "auth": "hmac_cookie",
        }, data_dir="../outside")  # 模拟尝试访问父级敏感目录
        with self.assertRaises(BudgetError):
            load_tickets(root)

if __name__ == "__main__":
    unittest.main()

执行测试：

python -m unittest test_budget.py

把这个测试文件紧挨着票证文件存放。只改业务逻辑的代码容易审查；凡是动到 budget.py 或配置表的修改，都应视为“升级预算申请”，必须经过人工确认。

关于适用场景与边界提醒

这套方案的核心价值在于“把选择权收回到开发者手中”。AI生成的代码再聪明，也需要人来守住底线。白名单制度不仅靠技术拦截，更靠团队协作纪律。票证文件就是预算申请书，证明文件就是付款回执。

但你要清楚它的局限性：
– 内存队列在重启后会丢失，不适合需要持久化任务状态的场景。
– 邮件保存为本地文件不具备真实投递能力，仅作流程验证。
– SQLite在多机环境下不支持并发写入。
– 绑定本地回环地址不等于公网上线，HMAC签名也不等同于专业的身份认证平台。

如果你的产品已经拥有付费用户，需要多实例故障转移、全球备份或合规审计，请果断放弃此方案。那是另一套规则，应该在产生收入后再引入。

对于只想在本周快速验证想法、跑通闭环的独立开发者而言，先写好本地适配器，封死配置白名单，再用基础脚本完成发货验证。当你的第一个真实用户出现时，再从容规划下一阶段的架构演进。早一点克制，晚一点花钱，这才是独立产品最稳健的成长节奏。