导航菜单

  • 1.什么是MCP
  • 2.MCP架构
  • 3.MCP服务器
  • 4.MCP客户端
  • 5.版本控制
  • 6.连接MCP服务器
  • 7.SDKs
  • 8.Inspector
  • 9.规范
  • 10.架构
  • 11.协议
  • 12.生命周期
  • 13.工具
  • 14.资源
  • 15.提示
  • 16.日志
  • 17.进度
  • 18.传输
  • 19.补全
  • 20.引导
  • 21.采样
  • 22.任务
  • 23.取消
  • 24.Ping
  • 25.根
  • 26.分页
  • 27.授权
  • 28.初始化
  • 29.工具
  • 30.资源
  • 31.结构化输出
  • 32.提示词
  • 33.上下文
  • 34.StreamableHTTP
  • 35.参数补全
  • 36.引导
  • 37.采样
  • 38.LowLevel
  • 39.任务
  • 40.取消
  • 41.ping
  • 42.根
  • 43.分页
  • 44.授权
  • 45.FunctionCalling
  • starlette
  • FastAPI
  • Keycloak
  • asyncio
  • contextlib
  • httpx
  • pathlib
  • pydantic
  • queue
  • subprocess
  • threading
  • uvicorn
  • JSON-RPC
  • 1. 什么是 FastAPI?
  • 2. 前置知识:Web API 与 HTTP 方法
    • 2.1 什么是 Web API?
    • 2.2 常用 HTTP 方法
  • 3. 安装
  • 4. 第一个 FastAPI 应用
  • 5. 路径参数与查询参数
    • 5.1 路径参数
    • 5.2 查询参数
  • 6. 请求体与 Pydantic 模型
    • 6.1 Pydantic 是什么?
    • 6.2 定义模型并接收请求体
  • 7. 增删改查(CRUD)
  • 8. 依赖注入(Depends)
    • 8.1 什么是依赖注入?
    • 8.2 基础用法
    • 8.3 依赖链:依赖可以依赖其他依赖
    • 8.4 常见场景:模拟「当前用户校验」
    • 8.6 常见场景:数据库连接
    • 8.7 小结
  • 9. CORS 跨域配置
  • 10. 自动文档
  • 11. 运行方式
  • 12. 总结

1. 什么是 FastAPI? #

FastAPI 是一个用于构建 Web API 的 Python 框架,特点是快、简单、自动生成文档。

你写几行 Python 代码,就能得到一个可以通过浏览器或程序访问的接口。比如你写一个「获取用户信息」的函数,FastAPI 会把它变成 GET /users/1 这样的网址,别人访问就能拿到数据。

打个比方:就像开一家「自助取餐窗口」——你定义好菜单(接口),顾客按编号点餐(发请求),窗口自动把菜端出来(返回数据)。FastAPI 帮你把「菜单」和「取餐流程」都标准化好了。

2. 前置知识:Web API 与 HTTP 方法 #

在学 FastAPI 之前,需要了解两个基础概念。

2.1 什么是 Web API? #

API(Application Programming Interface)是程序之间通信的接口。Web API 就是通过网址(URL)来调用的接口。

例如:访问 https://api.example.com/users/1,服务器返回用户 1 的信息(通常是 JSON 格式)。你的程序发请求,服务器返回数据,这就是 Web API 的典型用法。

2.2 常用 HTTP 方法 #

方法 含义 常见用途
GET 获取 查询数据,如获取用户列表
POST 提交 创建数据,如新建用户
PUT 更新 整体替换,如修改用户信息
DELETE 删除 删除数据

FastAPI 用装饰器区分这些方法:@app.get()和@app.post()等。

3. 安装 #

FastAPI 需要配合 ASGI 服务器 才能运行,推荐使用 Uvicorn。

pip install fastapi uvicorn[standard]

若使用 uv 管理依赖:uv add fastapi "uvicorn[standard]"

4. 第一个 FastAPI 应用 #

# 导入 FastAPI 类
from fastapi import FastAPI
# 导入 uvicorn 用于启动服务器
import uvicorn
# 创建 FastAPI 应用实例,title 会显示在自动文档中
app = FastAPI(title="我的第一个 API")

# 使用 @app.get 定义 GET 请求的路由,"/" 表示根路径
@app.get("/")
def read_root():
    # 返回一个字典,FastAPI 会自动转为 JSON
    return {"message": "Hello, FastAPI!"}


# 定义 GET /items/{item_id},item_id 是路径参数
@app.get("/items/{item_id}")
def read_item(item_id: int):
    # 路径参数会自动解析并做类型校验
    return {"item_id": item_id, "name": f"商品{item_id}"}


# 程序入口:直接运行此文件时执行
if __name__ == "__main__":

    # 启动服务器,host 和 port 为监听地址和端口
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行方式:保存为 main.py,执行 python main.py,然后访问:

  • http://127.0.0.1:8000 → 返回 {"message": "Hello, FastAPI!"}
  • http://127.0.0.1:8000/items/1 → 返回 {"item_id": 1, "name": "商品1"}
  • http://127.0.0.1:8000/docs → 自动生成的交互式文档

5. 路径参数与查询参数 #

5.1 路径参数 #

路径参数写在 URL 路径中,如 /items/123 里的 123。

# 路径参数:写在花括号 {param} 中,与函数参数同名
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

5.2 查询参数 #

查询参数写在 URL 的 ? 后面,如 /items?skip=0&limit=10。

# 查询参数:有默认值的函数参数,未在路径中出现的即为查询参数
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

访问 http://127.0.0.1:8000/items?skip=5&limit=20 会得到 {"skip": 5, "limit": 20}。

6. 请求体与 Pydantic 模型 #

6.1 Pydantic 是什么? #

Pydantic 是 FastAPI 用来做数据校验和序列化的库。你定义一个「模型」(类),指定每个字段的类型,Pydantic 会自动:

  • 校验传入数据是否符合类型
  • 把 Python 对象转成 JSON
  • 生成 API 文档中的参数说明

6.2 定义模型并接收请求体 #

当客户端用 POST 发送 JSON 数据时,可以用 Pydantic 模型来接收和校验。

# 导入 FastAPI 和 Pydantic 的 BaseModel
from fastapi import FastAPI
from pydantic import BaseModel

# 创建应用
app = FastAPI()


# 定义请求体模型:继承 BaseModel,声明字段及类型
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False


# POST 请求:item 会自动从请求体中解析,并做类型校验
@app.post("/items/")
def create_item(item: Item):
    return item


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)
curl -i -X POST "http://127.0.0.1:8000/items/"  -H "Content-Type: application/json"   -d '{"name":"apple","price":12.5}'

7. 增删改查(CRUD) #

# 导入 FastAPI 和 Pydantic
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# 创建应用
app = FastAPI(title="商品 API")

# 内存中模拟数据库(实际项目会用真实数据库)
items_db = {}
next_id = 1


# 定义商品模型
class Item(BaseModel):
    name: str
    price: float
    description: str = ""


# 获取所有商品
@app.get("/items")
def list_items():
    return {"items": [{"id": k, **v.model_dump()} for k, v in items_db.items()]}


# 根据 id 获取单个商品
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    return {"id": item_id, **items_db[item_id].model_dump()}


# 创建商品(id 自动递增)
@app.post("/items")
def create_item(item: Item):
    global next_id
    items_db[next_id] = item
    result_id = next_id
    next_id += 1
    return {"message": "创建成功", "id": result_id, "item": item}


# 更新商品
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    items_db[item_id] = item
    return {"message": "更新成功", "id": item_id, "item": item}


# 删除商品
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    del items_db[item_id]
    return {"message": "删除成功"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行方式:保存为 main.py,执行 python main.py,通过 /docs 或 curl 测试各接口。

# 1) GET /items  获取所有商品
curl -i "http://127.0.0.1:8000/items"
# 2) GET /items/{item_id}  获取单个商品(示例 id=1)
curl -i "http://127.0.0.1:8000/items/1"
# 3) POST /items  创建商品
curl -i -X POST "http://127.0.0.1:8000/items" \
  -H "Content-Type: application/json" \
  -d '{"name":"apple","price":12.5,"description":"fresh"}'
# 4) PUT /items/{item_id}  更新商品(示例 id=1)
curl -i -X PUT "http://127.0.0.1:8000/items/1" \
  -H "Content-Type: application/json" \
  -d '{"name":"apple-pro","price":19.9,"description":"updated"}'
# 5) DELETE /items/{item_id}  删除商品(示例 id=1)
curl -i -X DELETE "http://127.0.0.1:8000/items/1"

补充:这组命令里 GET/PUT/DELETE 用到 id=1,所以建议先执行一次 POST 创建数据,再测其余 3 条。

8. 依赖注入(Depends) #

8.1 什么是依赖注入? #

依赖注入 是把「获取某样东西」的逻辑单独写成函数,由 FastAPI 在每次请求时自动调用并传入路由。这样路由函数只关心业务逻辑,不必重复写「取参数、校验、获取数据库连接」等代码。

通俗理解:就像点外卖——你只说「我要一份盖浇饭」,平台自动帮你联系餐厅、安排骑手。你不需要自己打电话、自己取餐。Depends 就是 FastAPI 里的「自动安排」机制。

8.2 基础用法 #

用 Depends(函数名) 声明依赖,FastAPI 会在处理请求前先调用该函数,把返回值传给路由参数。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends

# 创建应用
app = FastAPI()


# 定义依赖函数:返回公共的查询参数
def common_params(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}


# 使用 Depends 注入:commons 会自动调用 common_params 并传入结果
@app.get("/items")
def read_items(commons: dict = Depends(common_params)):
    return commons


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

访问 http://127.0.0.1:8000/items?skip=5&limit=20 会得到 {"skip": 5, "limit": 20}。

8.3 依赖链:依赖可以依赖其他依赖 #

依赖函数本身也可以依赖其他依赖,形成依赖链。FastAPI 会按依赖关系自动解析并调用。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends

# 创建应用
app = FastAPI()


# 第一层依赖:分页参数
def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}


# 第二层依赖:依赖 pagination,在其基础上增加 keyword
def search_params(pg: dict = Depends(pagination), keyword: str = None):
    return {**pg, "keyword": keyword or ""}


# 路由依赖 search_params(内部会先调用 pagination)
@app.get("/items")
def list_items(params: dict = Depends(search_params)):
    return params


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

访问 http://127.0.0.1:8000/items?skip=2&limit=5&keyword=手机 会得到 {"skip": 2, "limit": 5, "keyword": "手机"}。

8.4 常见场景:模拟「当前用户校验」 #

依赖注入常用于校验请求头、Token 等,通过后再把用户信息传给路由。从请求头读取参数需使用 Header。

# 导入 FastAPI、Depends、Header 和 HTTPException
from fastapi import FastAPI, Depends, Header, HTTPException

# 创建应用
app = FastAPI()


# 从请求头 X-User-Id 读取用户 id,Header(None) 表示可选,缺省为 None
def get_current_user_id(x_user_id: str = Header(None)):
    if not x_user_id:
        raise HTTPException(status_code=401, detail="请提供 X-User-Id 请求头")
    return {"user_id": x_user_id}


# 需要「登录」的接口:依赖 get_current_user_id
@app.get("/me")
def read_me(user: dict = Depends(get_current_user_id)):
    return {"message": f"当前用户: {user['user_id']}"}


# 公开接口:不声明依赖
@app.get("/public")
def read_public():
    return {"message": "公开接口"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

测试方式:

# 不带请求头,返回 401
curl -i "http://127.0.0.1:8000/me"

# 带 X-User-Id,返回 200
curl -i -H "X-User-Id: alice" "http://127.0.0.1:8000/me"

8.6 常见场景:数据库连接 #

通过依赖注入管理数据库连接,每次请求获取连接、用完后自动关闭,避免连接泄漏。使用 yield 时,yield 之前的代码在请求开始时执行,之后的代码在请求结束后执行(类似 finally)。

# 导入 asynccontextmanager 上下文管理器用于 lifespan 生命周期管理
from contextlib import asynccontextmanager
# 从 FastAPI 导入 FastAPI 应用类和 Depends 依赖注入工具
from fastapi import FastAPI, Depends
# 导入 sqlite3 模块,用于操作 SQLite 数据库
import sqlite3

# 定义数据库文件路径
DB_PATH = "demo.db"


# 依赖函数:获取数据库连接,使用 yield 实现请求结束后自动关闭连接
def get_db():
    # 连接到 SQLite 数据库
    conn = sqlite3.connect(DB_PATH)
    # 设置查询返回的每一行以 dict 形式表现,支持通过列名获取数据
    conn.row_factory = sqlite3.Row
    try:
        # 通过 yield 暴露连接对象
        yield conn
    finally:
        # 请求结束后关闭连接,防止资源泄漏
        conn.close()


# 初始化数据库,创建 users 表并插入初始用户数据
def init_db():
    # 建立数据库连接
    conn = sqlite3.connect(DB_PATH)
    # 创建表(如果不存在),包含 id 和 name 字段
    conn.execute("CREATE TABLE IF NOT EXISTS users (id INT, name TEXT)")
    # 插入或更新样例数据:id=1 的张三、id=2 的李四
    conn.execute("INSERT OR REPLACE INTO users VALUES (1, '张三'), (2, '李四')")
    # 提交事务
    conn.commit()
    # 关闭连接
    conn.close()


# 声明异步 lifespan 生命周期上下文,在应用启动时初始化数据库
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 在应用启动阶段调用数据库初始化函数
    init_db()
    # 进入主生命周期,后续事件继续执行
    yield


# 创建 FastAPI 应用实例,并指定 lifespan 生命周期管理
app = FastAPI(lifespan=lifespan)


# 定义获取单个用户信息的 GET 路由接口,依赖数据库连接
@app.get("/users/{user_id}")
def get_user(user_id: int, conn=Depends(get_db)):
    # 执行 SQL 查询,根据用户 id 查找用户
    cur = conn.execute("SELECT id, name FROM users WHERE id = ?", (user_id,))
    # 获取一条查询结果
    row = cur.fetchone()
    # 如果没有找到对应用户,返回错误信息
    if not row:
        return {"error": "用户不存在"}
    # 找到对应用户,返回 id 和 name 信息
    return {"id": row["id"], "name": row["name"]}


# 判断当前脚本是否为主程序入口
if __name__ == "__main__":
    # 导入 uvicorn 用于启动 FastAPI 应用
    import uvicorn
    # 启动 uvicorn 服务,监听本地 127.0.0.1:8000 端口
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行方式:保存为 main.py,执行 python main.py,访问 http://127.0.0.1:8000/users/1 可得到 {"id": 1, "name": "张三"}。

8.7 小结 #

要点 说明
声明方式 参数 = Depends(依赖函数)
执行时机 每次请求前自动调用依赖函数
依赖链 依赖函数可以再依赖其他依赖
常见用途 分页参数、用户校验、数据库连接等

9. CORS 跨域配置 #

当前端(如网页)从不同域名访问你的 API 时,浏览器会做跨域检查。若未配置 CORS,请求可能被拦截。

# 导入 FastAPI 类
from fastapi import FastAPI
# 导入用于支持跨域请求的 CORS 中间件
from fastapi.middleware.cors import CORSMiddleware

# 创建 FastAPI 应用实例
app = FastAPI()

# 配置并添加 CORS 中间件,允许所有来源的跨域请求
app.add_middleware(
    CORSMiddleware,
    # 允许所有来源访问,生产环境推荐填写具体域名
    allow_origins=["*"],
    # 允许客户端发送 cookies 等凭证信息
    allow_credentials=True,
    # 允许所有 HTTP 方法
    allow_methods=["*"],
    # 允许所有请求头
    allow_headers=["*"],
)

# 定义根路径的 GET 请求接口
@app.get("/")
def read_root():
    # 返回一条消息
    return {"message": "Hello"}

# 判断是否以脚本方式运行
if __name__ == "__main__":
    # 导入 uvicorn,用于运行 ASGI 服务器
    import uvicorn
    # 启动 FastAPI 应用,监听本地 127.0.0.1:8000
    uvicorn.run(app, host="127.0.0.1", port=8000)

10. 自动文档 #

FastAPI 会自动根据你的路由和模型生成文档,无需额外配置。

地址 说明
http://127.0.0.1:8000/docs Swagger UI,可在线调试
http://127.0.0.1:8000/redoc ReDoc,阅读式文档

11. 运行方式 #

方式一:在代码中调用 uvicorn(如前文示例)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

方式二:命令行启动(推荐开发时使用 --reload 热重载)

uvicorn main:app --reload --host 127.0.0.1 --port 8000

其中 main 是文件名,app 是 FastAPI 实例的变量名。

12. 总结 #

  • FastAPI 用装饰器定义路由,用 Pydantic 做请求体验证,用 Depends 做依赖注入。
  • 路径参数 写在 {param} 中,查询参数 用带默认值的函数参数。
  • 请求体 用 Pydantic 模型接收,自动校验并生成文档。
  • 启动后访问 /docs 可查看和调试所有接口。
← 上一节 contextlib 下一节 httpx →

访问验证

请输入访问令牌

Token不正确,请重新输入