API变现教程 — 把你的工具变成印钞机

🎯 课程目标：学习如何将你的代码工具/能力包装成 REST API 服务，上架 API 市场，实现持续被动收入。本课程包含完整的 FastAPI 实战代码，涵盖网关设计、认证、限流、计费和自动文档生成。

第一章：什么是 API 经济

1.1 API 经济概述

API 经济（API Economy）是指通过开放和消费 API 来创造商业价值的经济模式。当今世界，几乎所有互联网服务都通过 API 连接：

Stripe 的 API 处理了全球数千亿美元的支付
OpenAI 的 API 让任何应用都能接入 GPT 能力
Twilio 的 API 让任何应用都能发短信、打电话

API 变现的本质：把你写好的能力（代码），通过标准接口开放出去，让别人付费调用。

1.2 为什么 API 是最好的被动收入

维度	传统软件	API 服务
交付方式	需要安装部署	HTTP 调用即用
计费方式	一次性/年度订阅	按调用次数实时计费
扩展性	需要销售团队	开发者自助注册使用
维护成本	每个客户独立部署	集中维护，所有人共享
变现速度	慢，需要漫长的销售周期	快，开发者看完文档就能付费调用

1.3 你的 82 个工具如何变成 API

假设你已经有了 82 个工具/脚本（文本处理、数据转换、AI 调用、格式转换等），变现路径如下：

工具脚本 → FastAPI 包装 → API 网关（认证+限流+计费） → 上架市场 → 被动收入

每一步本课程都会详细讲解。

第二章：用 FastAPI 包装工具为 REST API

2.1 项目结构

api-monetization/
├── app/
│   ├── __init__.py
│   ├── main.py                  # FastAPI 入口
│   ├── config.py                # 配置
│   ├── database.py              # 数据库
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py              # 用户模型
│   │   └── usage.py             # 使用记录模型
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── api.py               # API 请求/响应模型
│   ├── middleware/
│   │   ├── __init__.py
│   │   ├── auth.py              # API Key 认证
│   │   ├── rate_limit.py        # 限流
│   │   └── billing.py           # 计费中间件
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── auth.py              # 认证路由
│   │   ├── tools.py             # 工具 API 路由
│   │   ├── text.py              # 文本处理工具
│   │   ├── image.py             # 图片处理工具
│   │   ├── data.py              # 数据处理工具
│   │   └── ai.py                # AI 工具
│   └── services/
│       ├── __init__.py
│       ├── tool_registry.py     # 工具注册表
│       └── billing.py           # 计费服务
├── tools/                       # 你的 82 个工具
│   ├── text_tools.py
│   ├── image_tools.py
│   ├── data_tools.py
│   └── ai_tools.py
├── requirements.txt
├── Dockerfile
└── .env.example

2.2 核心配置（config.py）

# app/config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    APP_NAME: str = "ToolAPI"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False

    # 数据库
    DATABASE_URL: str = "postgresql://postgres:***@localhost:5432/toolapi"

    # Redis
    REDIS_URL: str = "redis://localhost:***@app.get("/")
def root():
    return {
        "name": settings.APP_NAME,
        "version": settings.APP_VERSION,
        "endpoints": {
            "docs": "/docs",
            "redoc": "/redoc",
            "openapi": "/openapi.json",
            "tools": "/api/v1/tools",
        },
    }


@app.get("/health")
def health():
    return {"status": "healthy", "version": settings.APP_VERSION}

2.4 完整的工具注册系统

# app/services/tool_registry.py
"""工具注册系统 — 管理所有 API 工具的注册、分类和元数据"""

from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum


class ToolCategory(str, Enum):
    TEXT = "text"
    IMAGE = "image"
    DATA = "data"
    AI = "ai"
    CONVERSION = "conversion"
    VALIDATION = "validation"


class PricingTier(str, Enum):
    FREE = "free"           # 免费
    BASIC = "basic"         # 基础版
    PRO = "pro"             # 专业版
    ENTERPRISE = "enterprise"  # 企业版


@dataclass
class ToolMeta:
    name: str
    description: str
    category: ToolCategory
    pricing_tier: PricingTier
    cost_per_call: int           # 每次调用费用（分）
    input_schema: Dict[str, Any]
    output_schema: Dict[str, Any]
    tags: List[str] = field(default_factory=list)
    rate_limit: int = 100       # 每分钟调用上限
    handler: Optional[Callable] = None

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "category": self.category.value,
            "pricing_tier": self.pricing_tier.value,
            "cost_per_call_cents": self.cost_per_call,
            "rate_limit_per_minute": self.rate_limit,
            "tags": self.tags,
            "input_schema": self.input_schema,
            "output_schema": self.output_schema,
        }


class ToolRegistry:
    """工具注册中心 — 单例模式"""

    _instance = None
    _tools: Dict[str, ToolMeta] = {}

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def register(self, tool_id: str, meta: ToolMeta):
        """注册一个工具"""
        self._tools[tool_id] = meta
        return meta

    def get(self, tool_id: str) -> Optional[ToolMeta]:
        return self._tools.get(tool_id)

    def list_tools(
        self,
        category: Optional[ToolCategory] = None,
        pricing_tier: Optional[PricingTier] = None,
    ) -> List[Dict]:
        tools = list(self._tools.values())
        if category:
            tools = [t for t in tools if t.category == category]
        if pricing_tier:
            tools = [t for t in tools if t.pricing_tier == pricing_tier]
        return [t.to_dict() for t in tools]

    def get_categories(self) -> Dict[str, int]:
        result = {}
        for tool in self._tools.values():
            cat = tool.category.value
            result[cat] = result.get(cat, 0) + 1
        return result


# 全局注册表实例
registry = ToolRegistry()

2.5 工具实现示例

# tools/text_tools.py
"""文本处理工具集 — 每个工具都是独立的函数，通过注册系统暴露为 API"""

import hashlib
import base64
import json
import re
from collections import Counter
from typing import Any, Dict, List

from app.services.tool_registry import (
    registry,
    ToolMeta,
    ToolCategory,
    PricingTier,
)


# ===== 工具1: 文本摘要 =====
def text_summarize(text: str, max_sentences: int = 3) -> Dict[str, Any]:
    """简易文本摘要（基于句子权重）"""
    sentences = re.split(r'[。！？.!?\n]+', text)
    sentences = [s.strip() for s in sentences if len(s.strip()) > 5]

    if len(sentences) <= max_sentences:
        return {"summary": "。".join(sentences), "sentence_count": len(sentences)}

    # 计算词频
    words = re.findall(r'[\w\u4e00-\u9fff]+', text.lower())
    word_freq = Counter(words)

    # 按句子得分排序
    scored = []
    for i, sent in enumerate(sentences):
        sent_words = re.findall(r'[\w\u4e00-\u9fff]+', sent.lower())
        score = sum(word_freq.get(w, 0) for w in sent_words)
        # 位置权重：开头和结尾的句子更重要
        if i < 2:
            score *= 1.5
        if i >= len(sentences) - 2:
            score *= 1.2
        scored.append((score, i, sent))

    scored.sort(reverse=True)
    selected = sorted(scored[:max_sentences], key=lambda x: x[1])
    summary = "。".join(s[2] for s in selected)

    return {"summary": summary, "original_length": len(text), "summary_length": len(summary)}


registry.register(
    "text_summarize",
    ToolMeta(
        name="文本摘要",
        description="对输入文本进行自动摘要提取",
        category=ToolCategory.TEXT,
        pricing_tier=PricingTier.FREE,
        cost_per_call=0,
        rate_limit=30,
        tags=["nlp", "摘要", "文本处理"],
        input_schema={
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "待摘要的文本"},
                "max_sentences": {"type": "integer", "default": 3, "description": "摘要最大句子数"},
            },
            "required": ["text"],
        },
        output_schema={
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "original_length": {"type": "integer"},
                "summary_length": {"type": "integer"},
            },
        },
    ),
)


# ===== 工具2: 关键词提取 =====
def extract_keywords(text: str, top_n: int = 10) -> Dict[str, Any]:
    """基于 TF 的关键词提取"""
    words = re.findall(r'[\w\u4e00-\u9fff]{2,}', text.lower())

    # 停用词（简化版）
    stopwords = set("的了是在我有和人这中大为上个国不以到说时要就出会也年对自其")
    words = [w for w in words if w not in stopwords and len(w) >= 2]

    freq = Counter(words)
    top_keywords = freq.most_common(top_n)

    return {
        "keywords": [{"word": w, "count": c} for w, c in top_keywords],
        "total_words": len(words),
        "unique_words": len(freq),
    }


registry.register(
    "text_keywords",
    ToolMeta(
        name="关键词提取",
        description="从文本中提取最重要的关键词",
        category=ToolCategory.TEXT,
        pricing_tier=PricingTier.FREE,
        cost_per_call=0,
        rate_limit=30,
        tags=["nlp", "关键词", "文本分析"],
        input_schema={
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "待分析的文本"},
                "top_n": {"type": "integer", "default": 10, "description": "返回关键词数量"},
            },
            "required": ["text"],
        },
        output_schema={
            "type": "object",
            "properties": {
                "keywords": {"type": "array", "items": {"type": "object"}},
                "total_words": {"type": "integer"},
                "unique_words": {"type": "integer"},
            },
        },
    ),
)


# ===== 工具3: 文本加密（Base64） =====
def text_encode(text: str, encoding: str = "base64") -> Dict[str, Any]:
    """文本编码/加密"""
    if encoding == "base64":
        result = base64.b64encode(text.encode()).decode()
    elif encoding == "hex":
        result = text.encode().hex()
    elif encoding == "md5":
        result = hashlib.md5(text.encode()).hexdigest()
    elif encoding == "sha256":
        result = hashlib.sha256(text.encode()).hexdigest()
    else:
        return {"error": f"不支持的编码方式: {encoding}"}

    return {"input": text[:50] + "..." if len(text) > 50 else text, "encoding": encoding, "result": result}


registry.register(
    "text_encode",
    ToolMeta(
        name="文本编码",
        description="支持 Base64、Hex、MD5、SHA256 编码",
        category=ToolCategory.CONVERSION,
        pricing_tier=PricingTier.FREE,
        cost_per_call=0,
        rate_limit=60,
        tags=["编码", "加密", "转换"],
        input_schema={
            "type": "object",
            "properties": {
                "text": {"type": "string"},
                "encoding": {"type": "string", "enum": ["base64", "hex", "md5", "sha256"], "default": "base64"},
            },
            "required": ["text"],
        },
        output_schema={
            "type": "object",
            "properties": {
                "input": {"type": "string"},
                "encoding": {"type": "string"},
                "result": {"type": "string"},
            },
        },
    ),
)


# ===== 工具4: JSON 格式化 =====
def json_format(text: str, indent: int = 2) -> Dict[str, Any]:
    """JSON 格式化和验证"""
    try:
        obj = json.loads(text)
        formatted = json.dumps(obj, indent=indent, ensure_ascii=False)
        return {
            "valid": True,
            "formatted": formatted,
            "keys_count": len(obj) if isinstance(obj, dict) else None,
            "type": type(obj).__name__,
        }
    except json.JSONDecodeError as e:
        return {"valid": False, "error": str(e)}


registry.register(
    "json_format",
    ToolMeta(
        name="JSON格式化",
        description="格式化和验证 JSON 数据",
        category=ToolCategory.DATA,
        pricing_tier=PricingTier.FREE,
        cost_per_call=0,
        rate_limit=60,
        tags=["json", "格式化", "验证"],
        input_schema={
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "JSON 字符串"},
                "indent": {"type": "integer", "default": 2},
            },
            "required": ["text"],
        },
        output_schema={
            "type": "object",
            "properties": {
                "valid": {"type": "boolean"},
                "formatted": {"type": "string"},
            },
        },
    ),
)


# ===== 工具5: 正则表达式测试 =====
def regex_test(pattern: str, text: str) -> Dict[str, Any]:
    """正则表达式测试"""
    try:
        matches = list(re.finditer(pattern, text))
        return {
            "valid_pattern": True,
            "match_count": len(matches),
            "matches": [
                {"text": m.group(), "start": m.start(), "end": m.end()}
                for m in matches[:20]
            ],
        }
    except re.error as e:
        return {"valid_pattern": False, "error": str(e)}


registry.register(
    "regex_test",
    ToolMeta(
        name="正则表达式测试",
        description="测试正则表达式并返回匹配结果",
        category=ToolCategory.DEVELOPER,
        pricing_tier=PricingTier.FREE,
        cost_per_call=0,
        rate_limit=60,
        tags=["正则", "开发工具"],
        input_schema={
            "type": "object",
            "properties": {
                "pattern": {"type": "string"},
                "text": {"type": "string"},
            },
            "required": ["pattern", "text"],
        },
        output_schema={
            "type": "object",
            "properties": {
                "valid_pattern": {"type": "boolean"},
                "match_count": {"type": "integer"},
                "matches": {"type": "array"},
            },
        },
    ),
)

2.6 认证中间件

# app/middleware/auth.py
"""API Key 认证中间件"""

from fastapi import Request, HTTPException, Depends
from fastapi.security import APIKeyHeader
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.user import APIUser

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


async def get_api_user(
    request: Request,
    api_key: str = Depends(api_key_header),
    db: Session = Depends(get_db),
) -> APIUser:
    """通过 API Key 认证用户"""
    if not api_key:
        raise HTTPException(
            status_code=401,
            detail={
                "error": "missing_api_key",
                "message": "请在请求头中提供 X-API-Key",
                "docs": "https://yourdomain.com/docs/authentication",
            },
        )

    user = db.query(APIUser).filter(APIUser.api_key == api_key, APIUser.is_active == True).first()
    if not user:
        raise HTTPException(
            status_code=401,
            detail={
                "error": "invalid_api_key",
                "message": "API Key 无效或已禁用",
            },
        )

    # 将用户信息附加到 request.state
    request.state.user = user
    return user


def require_tier(minimum_tier: str):
    """要求用户达到最低套餐等级"""
    tier_order = {"free": 0, "basic": 1, "pro": 2, "enterprise": 3}

    async def check_tier(user: APIUser = Depends(get_api_user)):
        if tier_order.get(user.tier, 0) < tier_order.get(minimum_tier, 0):
            raise HTTPException(
                status_code=403,
                detail={
                    "error": "upgrade_required",
                    "message": f"此功能需要 {minimum_tier} 或更高套餐",
                    "current_tier": user.tier,
                    "required_tier": minimum_tier,
                },
            )
        return user

    return check_tier

2.7 限流中间件

# app/middleware/rate_limit.py
"""基于 Redis 的 API 限流"""

import time
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import redis.asyncio as redis
from app.config import settings

redis_pool = None


async def get_redis():
    global redis_pool
    if redis_pool is None:
        redis_pool = redis.from_url(settings.REDIS_URL, decode_responses=True)
    return redis_pool


class RateLimitMiddleware(BaseHTTPMiddleware):
    """滑动窗口限流"""

    async def dispatch(self, request: Request, call_next):
        # 只限流 API 路径
        if not request.url.path.startswith("/api/v1/"):
            return await call_next(request)

        # 获取用户标识
        api_key = request.headers.get("X-API-Key", "")
        if not api_key:
            return await call_next(request)  # 认证中间件会处理

        r = await get_redis()

        # 获取用户的限流配置
        user_tier = getattr(request.state, "user_tier", "free")
        limits = {
            "free": 60,        # 60次/分钟
            "basic": 300,      # 300次/分钟
            "pro": 1200,       # 1200次/分钟
            "enterprise": 6000, # 6000次/分钟
        }
        max_requests = limits.get(user_tier, 60)

        # 滑动窗口计数
        now = time.time()
        window = 60  # 1分钟窗口
        key = f"rl:{api_key}"

        pipe = r.pipeline()
        pipe.zremrangebyscore(key, 0, now - window)
        pipe.zadd(key, {str(now): now})
        pipe.zcard(key)
        pipe.expire(key, window)
        results = await pipe.execute()

        current_count = results[2]

        if current_count > max_requests:
            return JSONResponse(
                status_code=429,
                content={
                    "error": "rate_limit_exceeded",
                    "message": f"超出速率限制 ({max_requests}/分钟)",
                    "retry_after": 1,
                },
                headers={
                    "X-RateLimit-Limit": str(max_requests),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(int(now + window)),
                    "Retry-After": "1",
                },
            )

        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(max_requests)
        response.headers["X-RateLimit-Remaining"] = str(max(0, max_requests - current_count))
        response.headers["X-RateLimit-Reset"] = str(int(now + window))
        return response

2.8 工具 API 路由（核心）

# app/routers/tools.py
"""统一工具 API 路由 — 所有工具通过一个统一接口调用"""

from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field

from app.database import get_db
from app.middleware.auth import get_api_user
from app.models.user import APIUser
from app.models.usage import UsageRecord
from app.services.tool_registry import registry, ToolCategory, PricingTier
from app.services.billing import BillingService

router = APIRouter()


class ToolCallRequest(BaseModel):
    tool_id: str = Field(..., description="工具 ID")
    params: Dict[str, Any] = Field(..., description="工具参数")


class ToolCallResponse(BaseModel):
    success: bool
    tool_id: str
    result: Any
    usage: Dict[str, Any]


@router.get("/tools", summary="列出所有可用工具")
async def list_tools(
    category: Optional[str] = None,
    pricing_tier: Optional[str] = None,
    user: APIUser = Depends(get_api_user),
):
    """获取所有可用工具列表"""
    cat = ToolCategory(category) if category else None
    tier = PricingTier(pricing_tier) if pricing_tier else None
    tools = registry.list_tools(category=cat, pricing_tier=tier)
    return {
        "total": len(tools),
        "tools": tools,
        "categories": registry.get_categories(),
    }


@router.get("/tools/{tool_id}", summary="获取工具详情")
async def get_tool(tool_id: str, user: APIUser = Depends(get_api_user)):
    """获取单个工具的详细信息"""
    tool = registry.get(tool_id)
    if not tool:
        raise HTTPException(status_code=404, detail=f"工具不存在: {tool_id}")
    return tool.to_dict()


@router.post("/tools/call", response_model=ToolCallResponse, summary="调用工具")
async def call_tool(
    req: ToolCallRequest,
    request: Request,
    user: APIUser = Depends(get_api_user),
    db: Session = Depends(get_db),
):
    """
    统一工具调用接口。所有工具都通过此接口调用。

    示例:
    ```json
    {
        "tool_id": "text_summarize",
        "params": {
            "text": "这是一段很长的文本...",
            "max_sentences": 3
        }
    }
    ```
    """
    # 1. 检查工具是否存在
    tool = registry.get(req.tool_id)
    if not tool:
        raise HTTPException(status_code=404, detail=f"工具不存在: {req.tool_id}")

    # 2. 检查套餐权限
    tier_order = {"free": 0, "basic": 1, "pro": 2, "enterprise": 3}
    if tier_order.get(user.tier, 0) < tier_order.get(tool.pricing_tier.value, 0):
        raise HTTPException(
            status_code=403,
            detail=f"工具 '{req.tool_id}' 需要 {tool.pricing_tier.value} 套餐，当前套餐: {user.tier}",
        )

    # 3. 检查余额（付费工具）
    if tool.cost_per_call > 0:
        if user.balance_cents < tool.cost_per_call:
            raise HTTPException(
                status_code=402,
                detail={
                    "error": "insufficient_balance",
                    "message": f"余额不足，需要 {tool.cost_per_call/100} 元",
                    "current_balance": user.balance_cents / 100,
                },
            )

    # 4. 执行工具
    if not tool.handler:
        raise HTTPException(status_code=500, detail="工具处理函数未注册")

    try:
        result = tool.handler(**req.params)
    except TypeError as e:
        raise HTTPException(status_code=400, detail=f"参数错误: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"工具执行失败: {str(e)}")

    # 5. 记录使用和扣费
    BillingService.record_usage(
        db=db,
        user_id=user.id,
        tool_id=req.tool_id,
        cost_cents=tool.cost_per_call,
    )

    if tool.cost_per_call > 0:
        BillingService.deduct_balance(db, user.id, tool.cost_per_call)

    return ToolCallResponse(
        success=True,
        tool_id=req.tool_id,
        result=result,
        usage={
            "cost_cents": tool.cost_per_call,
            "remaining_balance_cents": user.balance_cents - tool.cost_per_call,
        },
    )

2.9 计费服务

# app/services/billing.py
"""计费服务 — 记录使用量、扣费、余额管理"""

from datetime import datetime, timedelta
from typing import Dict, Optional
from sqlalchemy.orm import Session
from sqlalchemy import func

from app.models.user import APIUser
from app.models.usage import UsageRecord


class BillingService:

    @staticmethod
    def record_usage(db: Session, user_id: int, tool_id: str, cost_cents: int = 0):
        """记录一次 API 使用"""
        record = UsageRecord(
            user_id=user_id,
            tool_id=tool_id,
            cost_cents=cost_cents,
        )
        db.add(record)
        db.commit()

    @staticmethod
    def deduct_balance(db: Session, user_id: int, amount_cents: int):
        """扣减用户余额"""
        user = db.query(APIUser).filter(APIUser.id == user_id).first()
        if user:
            user.balance_cents = max(0, user.balance_cents - amount_cents)
            db.commit()

    @staticmethod
    def get_usage_stats(db: Session, user_id: int, days: int = 30) -> Dict:
        """获取使用统计"""
        since = datetime.utcnow() - timedelta(days=days)

        # 总调用次数和费用
        totals = (
            db.query(
                func.count(UsageRecord.id).label("total_calls"),
                func.coalesce(func.sum(UsageRecord.cost_cents), 0).label("total_cost"),
            )
            .filter(UsageRecord.user_id == user_id, UsageRecord.created_at >= since)
            .first()
        )

        # 按工具分组统计
        by_tool = (
            db.query(
                UsageRecord.tool_id,
                func.count(UsageRecord.id).label("calls"),
                func.sum(UsageRecord.cost_cents).label("cost"),
            )
            .filter(UsageRecord.user_id == user_id, UsageRecord.created_at >= since)
            .group_by(UsageRecord.tool_id)
            .order_by(func.count(UsageRecord.id).desc())
            .limit(20)
            .all()
        )

        # 每日统计
        daily = []
        for i in range(min(days, 30) - 1, -1, -1):
            day = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=i)
            day_end = day + timedelta(days=1)
            count = (
                db.query(func.count(UsageRecord.id))
                .filter(
                    UsageRecord.user_id == user_id,
                    UsageRecord.created_at >= day,
                    UsageRecord.created_at < day_end,
                )
                .scalar()
            )
            daily.append({"date": day.strftime("%Y-%m-%d"), "calls": count or 0})

        return {
            "period_days": days,
            "total_calls": totals.total_calls or 0,
            "total_cost_cents": int(totals.total_cost or 0),
            "total_cost_yuan": round(int(totals.total_cost or 0) / 100, 2),
            "top_tools": [
                {"tool_id": r.tool_id, "calls": r.calls, "cost_cents": int(r.cost or 0)}
                for r in by_tool
            ],
            "daily": daily,
        }

    @staticmethod
    def add_balance(db: Session, user_id: int, amount_cents: int):
        """充值余额"""
        user = db.query(APIUser).filter(APIUser.id == user_id).first()
        if user:
            user.balance_cents += amount_cents
            db.commit()
            return user.balance_cents
        return None

2.10 使用记录模型

# app/models/usage.py
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from app.database import Base


class UsageRecord(Base):
    __tablename__ = "usage_records"

    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey("api_users.id"), nullable=False, index=True)
    tool_id = Column(String(100), nullable=False, index=True)
    cost_cents = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow, index=True)

第三章：API 文档自动生成

3.1 OpenAPI / Swagger

FastAPI 自动基于你的 Pydantic 模型和路由函数生成 OpenAPI 文档。只需访问 /docs 即可看到完整的交互式 API 文档。

3.2 增强文档质量

# 在路由装饰器中添加详细描述
@router.post(
    "/tools/call",
    response_model=ToolCallResponse,
    summary="调用工具",
    description="""
    统一工具调用接口。所有工具都通过此接口调用。

    **使用流程：**
    1. 先调用 `GET /api/v1/tools` 获取可用工具列表
    2. 选择需要的工具，构造参数
    3. 调用 `POST /api/v1/tools/call`，传入 `tool_id` 和 `params`

    **示例请求：**
    ```json
    {
        "tool_id": "text_summarize",
        "params": {
            "text": "你的长文本...",
            "max_sentences": 3
        }
    }
    ```

    **注意事项：**
    - 免费工具无需扣费
    - 付费工具会自动从余额扣除
    - 超出套餐权限会返回 403
    """,
    responses={
        200: {"description": "工具调用成功"},
        400: {"description": "参数错误"},
        402: {"description": "余额不足"},
        403: {"description": "套餐权限不足"},
        404: {"description": "工具不存在"},
        429: {"description": "超出速率限制"},
    },
    tags=["tools"],
)
async def call_tool(...):
    ...

第四章：API 市场上架

4.1 RapidAPI

RapidAPI 是全球最大的 API 市场，月活开发者超过 300 万。

上架步骤：

注册 RapidAPI 开发者账号：https://docs.rapidapi.com/docs/quickstart
创建 API Provider Hub
填写 API 信息（名称、描述、文档 URL）
设置定价方案：
- Basic: $0/月，100 次调用
- Pro: $9.99/月，10000 次调用
- Ultra: $49.99/月，100000 次调用
提交审核（通常 1-3 个工作日）

RapidAPI 分成：平台抽取 20%，你获得 80%。

4.2 APILayer / APIHub

APILayer：https://apilayer.com — 适合数据类和金融类 API
APIHub：https://apihub.io — 综合 API 市场

4.3 自建 API 门户

如果不想被平台抽成，可以自建：

# 简单的 API 门户首页
@app.get("/portal", include_in_schema=False)
def api_portal():
    tools = registry.list_tools()
    return HTMLResponse(f"""
    <html>
    <head><title>ToolAPI - API 市场</title></head>
    <body>
        <h1>🚀 ToolAPI</h1>
        <p>共 {len(tools)} 个 API 工具可用</p>
        <h2>快速开始</h2>
        <pre>
# 1. 获取 API Key
curl -X POST https://api.toolapi.com/auth/register \\
  -d '{{"email": "[email protected]", "password": "yourpass"}}'

# 2. 调用工具
curl -X POST https://api.toolapi.com/api/v1/tools/call \\
  -H "X-API-Key: sk_xxxxx" \\
  -d '{{"tool_id": "text_summarize", "params": {{"text": "..."}}}}'
        </pre>
        <h2>可用工具</h2>
        {"".join(f'<h3>{t["name"]}</h3><p>{t["description"]}</p>' for t in tools)}
    </body>
    </html>
    """)

第五章：定价策略

5.1 按调用次数

最简单直接的定价方式：

费用 = 调用次数 × 单次价格

适合：工具类、数据查询类 API

5.2 按功能分层

不同功能不同价格：

层级	价格	功能
Free	$0	基础工具（文本、编码）
Basic	$9.99/月	数据处理、格式转换
Pro	$29.99/月	AI 工具、高级分析
Enterprise	$99.99/月	全部功能 + SLA

5.3 包月订阅 + 超量计费

月费包含 N 次调用，超出部分按次计费

这是最推荐的方式，既有稳定的 MRR，又能捕获大客户的增量收入。

第六章：完整部署方案

6.1 Dockerfile

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

6.2 docker-compose.yml

version: "3.8"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://postgres:***@db:5432/toolapi
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - db
      - redis

  db:
    image: postgres:15
    environment:
      POSTGRES_DB: toolapi
      POSTGRES_PASSWORD: postgres
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - api

volumes:
  pgdata:

6.3 Nginx 配置

# nginx.conf
events {
    worker_connections 1024;
}

http {
    # 限流配置
    limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;

    server {
        listen 80;
        server_name api.toolapi.com;

        # API 限流
        location /api/ {
            limit_req zone=api burst=20 nodelay;
            proxy_pass http://api:8000;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
        }

        # 文档页面（无限流）
        location /docs {
            proxy_pass http://api:8000;
            proxy_set_header Host $host;
        }

        location / {
            proxy_pass http://api:8000;
        }
    }
}

6.4 requirements.txt

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
pydantic[email]==2.5.2
pydantic-settings==2.1.0
redis==5.0.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6

总结

通过本课程，你学到了：

API 经济：理解了为什么 API 是最好的被动收入方式
工具包装：用 FastAPI + 工具注册系统，将任意 Python 函数包装为标准 API
网关设计：实现了 API Key 认证、滑动窗口限流、计费扣费的完整网关
自动文档：利用 FastAPI 的 OpenAPI 自动生成交互式 API 文档
市场上架：了解了 RapidAPI、APILayer 等市场的上架流程和分成模式
定价策略：按调用次数、按功能分层、包月+超量三种定价方式
完整部署：Docker + Nginx 的生产级部署方案

下一步行动：

把你的 82 个工具逐个包装为 API（每个只需 5 分钟）
注册 RapidAPI 开发者账号并上架
在社交媒体和开发者社区推广你的 API
监控使用数据，优化定价和功能

记住：好的 API 不仅要好用，还要好发现。文档质量、开发者体验、定价透明度是决定你的 API 能否变现的关键。

API变现实战教程2026：用FastAPI把代码工具变成被动收入印钞机，含完整Python代码

API变现教程 — 把你的工具变成印钞机

第一章：什么是 API 经济

1.1 API 经济概述

1.2 为什么 API 是最好的被动收入

1.3 你的 82 个工具如何变成 API

第二章：用 FastAPI 包装工具为 REST API

2.1 项目结构

2.2 核心配置（config.py）

2.4 完整的工具注册系统

2.5 工具实现示例

2.6 认证中间件

2.7 限流中间件

2.8 工具 API 路由（核心）

2.9 计费服务

2.10 使用记录模型

第三章：API 文档自动生成

3.1 OpenAPI / Swagger

3.2 增强文档质量

第四章：API 市场上架

4.1 RapidAPI

4.2 APILayer / APIHub

4.3 自建 API 门户

第五章：定价策略

5.1 按调用次数

5.2 按功能分层

5.3 包月订阅 + 超量计费

第六章：完整部署方案

6.1 Dockerfile

6.2 docker-compose.yml

6.3 Nginx 配置

6.4 requirements.txt

总结

评论

API变现教程 — 把你的工具变成印钞机

第一章：什么是 API 经济

1.1 API 经济概述

1.2 为什么 API 是最好的被动收入

1.3 你的 82 个工具如何变成 API

第二章：用 FastAPI 包装工具为 REST API

2.1 项目结构

2.2 核心配置（config.py）

2.4 完整的工具注册系统

2.5 工具实现示例

2.6 认证中间件

2.7 限流中间件

2.8 工具 API 路由（核心）

2.9 计费服务

2.10 使用记录模型

第三章：API 文档自动生成

3.1 OpenAPI / Swagger

3.2 增强文档质量

第四章：API 市场上架

4.1 RapidAPI

4.2 APILayer / APIHub

4.3 自建 API 门户

第五章：定价策略

5.1 按调用次数

5.2 按功能分层

5.3 包月订阅 + 超量计费

第六章：完整部署方案

6.1 Dockerfile

6.2 docker-compose.yml

6.3 Nginx 配置

6.4 requirements.txt

总结

评论

相关推荐

LiteLLM API聚合实战教程2026：一个接口统一管理100+AI模型，含完整Python代码

RAG知识库实战教程2026：从零搭建企业级AI知识库，含完整Python代码

AI安全实战教程2026：Prompt注入攻击全解析——从6大攻击手法到完整Python防御系统，含可运行代码

Prompt Engineering变现实战教程2026：5大核心技巧+商业模板库，含完整代码示例

2026年网络安全变现实战教程：渗透测试+漏洞赏金从零到月入5万，含完整Python代码