# 基本介绍
# 特点
- 依赖：Python 3.6 及更高版本
- Starlette (opens new window) 是一个轻量级的ASGI框架/工具包，用于构建Python异步网络服务
- Pydantic (opens new window) 负责数据部分(类型提示)
# 快速使用
- 安装
pip install fastapi
- 安装 ASGI 服务器，生产环境可以使用 Uvicorn (opens new window)
pip install uvicorn
- 相关代码
# FastAPI 是一个为你的 API 提供了所有功能的 Python 类
from fastapi import FastAPI
# 这个实例将是创建你所有 API 的主要交互对象。这个 app 同样在如下命令中被 uvicorn 所引用
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello yuan"}
- 通过以下命令运行服务器
uvicorn main:app --port=8090 --reload
- 也可以直接运行
if __name__ == '__main__':
import uvicorn
# main:app" 对应 main.py 文件，app 是 FastAPI 应用实例名
uvicorn.run("main:app", host="127.0.0.1", port=8080, debug=True, reload=True)
- 自动生成的交互式 API 文档 http://127.0.0.1:8080/docs
# 路径操作
# 路径操作装饰器
- fastapi支持各种请求方式
@app.get()
@app.post()
@app.put()
@app.patch()
@app.delete()
@app.options()
@app.head()
@app.trace()
- 路径操作装饰器参数
@app.get("/get/{id}",
status_code=200,
tags=["get方法"],
deprecated=False,
summary="get方法",
description="get方法描述",
response_model=str, # 响应模型
response_description="get方法响应描述",
)
# include_router 的使用
# quick.py
from fastapi import FastAPI
from app1 import app01
from app2 import app02
app = FastAPI()
app.include_router(app01, prefix="/app01", tags=["第一章节：商城接口",])
app.include_router(app02, prefix="/app02", tags=["第二章节：用户中心接口",])
if __name__ == "__main__":
import uvicorn
uvicorn.run("quick:app", host="127.0.0.1", port=8030, reload=True)
# app1.__init__.py
from .app1 import app01
# app2.__init__.py
from .app2 import app02
from fastapi import APIRouter
app01 = APIRouter()
@app01.get("/shop/food")
def shop_food():
return {"shop": "food"}
@app01.get("/shop/bed")
def shop_food():
return {"shop": "bed"}
from fastapi import APIRouter
app02 = APIRouter()
@app02.post("/user/login")
def user_login():
return {"user": "login"}
@app02.post("/user/reg")
def user_reg():
return {"user": "reg"}
# 请求数据
# 路径参数
- 基本用法
@app01.get("/shop/food/{food_id}")
def shop_food(food_id):
return {"shop": food_id}
- 有类型的路径参数
@app01.get("/shop/food/{food_id}")
def shop_food(food_id:int):
return {"shop": f"{type(food_id)}"}
- 注意顺序
# 要确保路径 /user/me 声明在路径 /user/{username}之前
@app.get("/user/me")
async def read_user_me():
return {"username": "the current user"}
@app.get("/user/{username}")
async def read_user(username: str):
return {"username": username}
# 查询参数（请求参数）
路径函数中声明不属于路径参数的其他函数参数时，它们将被自动解释为"查询字符串"参数，就是 url? 之后用&分割的 key-value 键值对
@app01.get("/shop/food/{food_id}")
def shop_food(food_id: int, food_name: Union[str, None] = None):
if food_name and food_price:
return {"food_id": food_id, "food_name": food_name}
else:
return {"food_id": food_id}
# 请求体数据
FastAPI 基于 Pydantic，Pydantic 主要用来做类型强制检查（校验数据）。不符合类型要求就会抛出异常
# 安装 pydantic
pip install pydantic
from datetime import date
from typing import Optional, Union, List
from fastapi import FastAPI
from pydantic import BaseModel, Field, field_validator, ValidationError
class Address(BaseModel):
city: str
state: str
class User(BaseModel):
name: str = "sylone"
# 当一个模型属性具有默认值时，它不是必需的。否则它是一个必需属性
age: int = Field(default=0, ge=18, le=60)
# 表明 birthday 属性的类型可以是 date 对象，也能为 None
birthday: Optional[date] = None
# Mutable default '[]' is not allowed. Use 'default_factory'
friends: List[int] = Field(default_factory=[]) # []
# 将默认值设为 None 可使其成为可选属性
decription: Union[str, None] = None
address: Union[Address, None] = None
@field_validator("name")
def name_must_alpha(cls, v):
assert v.isalpha(), "name must be alpha"
return v
class Item(BaseModel):
users: List[User]
app = FastAPI()
# 测试
# {
# "users": [
# {
# "name": "rain",
# "age": 32,
# "birthday": "2022-09-29",
# "friends": [1],
# "description": "最帅的讲fastapi的老师",
# "address":{
# "city":"111",
# "state":"1111"
# }
# }
# ]
# }
# FastAPI 会自动将定义的模型类转化为JSON
@app.post("/data")
async def create_data(data: Item):
return data
if __name__ == "__main__":
try:
User(name="zhangsan", age=50, friends=[1, 2, 3], decription="123",
address=Address(city="1", state="2"))
except ValidationError as e:
print(e.json())
# form表单数据
FastAPI 可以使用Form组件来接收表单数据，需要先安装 python-multipart
python install python-multipart
@app.post("/reg")
# ... 是 Python 的 Ellipsis 对象
# 它表示一个占位符，用于表示一个参数或变量的值是必须的，但没有具体的值
# 在 FastAPI 里，Ellipsis 对象常被用于 Form、Query、Path 等函数，表明某个参数是必需的
def user_reg(username:str = Form(..., min_length=2, max_length=10, regex="^[a-zA-Z]"),
password:str = Form(..., min_length=6, max_length=10, regex="^[0-9_]")):
print(f"username:{username}, password:{password}")
return {"user": "reg"}
# 验证不通过的返回结果
# {
# "detail": [
# {
# "type": "string_pattern_mismatch",
# "loc": [
# "body",
# "username"
# ],
# "msg": "String should match pattern '^[a-zA-Z]'",
# "input": "1212",
# "ctx": {
# "pattern": "^[a-zA-Z]"
# }
# }
# ]
# }
# 文件上传
- 适合小文件上传
@app.post("/file")
async def file_upload(file: bytes = File()):
print("file", file) # <class 'bytes'>
return {"file_size": len(file)}
@app.post("/multiFile")
async def multi_file_upload(files: list[bytes] = File()):
return {"file_size": [len(file) for file in files]}
- 适合大文件上传
@app.post("/files")
async def files_upload(file: UploadFile = File()):
with open(file.filename, "wb") as f:
f.write(file.file.read())
return {"file_name": file.filename}
@app.post("/multiFiles")
async def multi_files_upload(files: list[UploadFile] = File()):
return {"file_size": [file.filename for file in files]}
# Reqeust对象
在函数中声明Request类型的参数，FastAPI 就会自动传递 Request 对象给这个参数
@app.get("/test")
async def test(request: Request):
return {
"query_params": request.query_params,
"headers": request.headers,
"agent": request.headers.get("user-agent"),
"cookies": request.cookies,
"method": request.method,
"url": request.url, # "http://127.0.0.1:8030/test"
"path": request.url.path, # "/test"
"IP": request.client.host, # "127.0.0.1"
}
# 请求静态文件
请求如 css/js 和图片文件等
from fastapi import FastAPI, Request, File, UploadFile
from starlette.staticfiles import StaticFiles
app = FastAPI()
app.mount("/imgs", StaticFiles(directory="imgs"))
# 响应数据
# response_model
FastAPI将使用response_model进行以下操作：
# 将输出数据转换为response_model中声明的数据类型。
# 验证数据结构和类型
# 将输出数据限制为该model定义的
# 添加到OpenAPI中
# 在自动文档系统中使用
class UserIn(BaseModel):
username: str
password: str
email: EmailStr
full_name: Union[str, None] = None
class UserOut(BaseModel):
username: str
email: EmailStr
full_name: Union[str, None] = None
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
return user
# 数据过滤
- response_model_exclude_unset：仅返回显式设定的值
- response_model_exclude_defaults：不返回是默认值的字段
- response_model_exclude_none：不返回是None的字段
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: list[str] = []
Items = {
"foo": {"name": "Foo", "price": 50.2},
"bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
return Items[item_id]
# {
# "name": "Foo",
# "price": 50.2
# }
@app.get("/items/{item_id}/name", response_model=Item,
response_model_include=["name", "description"])
async def read_item_name(item_id: str):
return Items[item_id]
# {
# "name": "Foo",
# "description": null
# }
@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude=["tax"])
async def read_item_public_data(item_id: str):
return Items[item_id]
# {
# "name": "Foo",
# "price": 50.2,
# "description": null,
# "tags": []
# }
# jinja2模板
jinja2是Flask作者开发的一个模板系统，起初是仿django模板的一个模板引擎，为Flask提供模板支持，由于其灵活，快速和安全等优点被广泛使用。
# jinja2 的变量
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
import uvicorn
app = FastAPI()
# 实例化Jinja2对象，并将文件夹路径设置为以templates命令的文件夹
templates = Jinja2Templates(directory="templates")
@app.get('/')
def hello(request: Request):
return templates.TemplateResponse('index.html',
{
# 注意，返回模板响应时，必须有request键值对，且值为Request请求对象
'request': request,
'user': 'yuan',
"books": ["水浒传", "西游记", "三国演义", "红楼梦"],
"booksDict": {
"水浒传": {"price": 100, "author": "施耐庵"},
"西游记": {"price": 200, "author": "吴承恩"},
}
}
)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>{{ user}}</h1>
<p>{{ books.0 }}</p>
<p>{{ books.1 }}</p>
<p>{{ books.2 }}</p>
<p>{{ books.3 }}</p>
<p>{{ booksDict.水浒传.price }}</p>
</body>
</html>
# jinja2 的过滤器
变量可以通过“过滤器”进⾏修改，过滤器是jinja2里面的内置函数和字符串处理函数。常用的过滤器有：
| 过滤器名称 | 说明 |
|---|---|
| capitialize | 把值的首字母转换成大写，其他字母转换为小写 |
| lower | 把值转换成小写形式 |
| title | 把值中每个单词的首字母都转换成大写 |
| trim | 把值的首尾空格去掉 |
| striptags | 渲染之前把值中所有的HTML标签都删掉 |
| join | 拼接多个值为字符串 |
| round | 默认对数字进⾏四舍五⼊，也可以用参数进⾏控制 |
| safe | 渲染时值不转义 |
需要在变量后面使用管道 | 分割，多个过滤器可以链式调用
{{ 'abc'| captialize }} # Abc
{{ 'abc'| upper }} # ABC
{{ 'hello world'| title }} # Hello World
{{ "hello world"| replace('world','yuan') | upper }} # HELLO YUAN
{{ 18.18 | round | int }} # 18
# jinja2 的控制结构
- 条件语句不需要使用冒号结尾，而结束控制语句，需要使用endif关键字
{% if age > 18 %}
<p>成年区</p>
{% else %}
<p>未成年区</p>
{% endif %}
- for循环用于迭代Python的数据类型，包括列表，元组和字典。在jinja2中不存在while循环
{% for book in books %}
<p>{{ book }}</p>
{% endfor %}
# 中间件的
- "中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应之后工作
- 要创建中间件你可以在函数的顶部使用装饰器 @app.middleware("http")
- 中间件参数接收如下参数：
request
call_next # 它将接收request，作为参数
# 这个函数将 request 传递给相应的 路径操作
# 然后它将返回由相应的路径操作生成的 response
# 然后你可以在返回 response 前进一步修改它
- 函数中间件
import uvicorn
from fastapi import FastAPI
from fastapi import Request
import time
app = FastAPI()
@app.middleware("http")
async def m2(request: Request, call_next):
# 请求代码块
print("m2 request")
response = await call_next(request)
# 响应代码块
response.headers["author"] = "yuan"
print("m2 response")
return response
@app.middleware("http")
async def m1(request: Request, call_next):
# 请求代码块
print("m1 request")
# if request.client.host in ["127.0.0.1", ]: # 黑名单
# return Response(content="visit forbidden")
# if request.url.path in ["/user"]:
# return Response(content="visit forbidden")
start = time.time()
response = await call_next(request)
# 响应代码块
print("m1 response")
end = time.time()
response.headers["ProcessTimer"] = str(end - start)
return response
@app.get("/user")
def get_user():
time.sleep(3)
print("get_user函数执行")
return {
"user": "current user"
}
# 输出
# m1 request
# m2 request
# get_user函数执行
# m2 response
# m1 response
if __name__ == '__main__':
uvicorn.run('main:app', host='127.0.0.1', port=8030, reload=True, workers=1)
- 类中间件
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import JSONResponse, PlainTextResponse
import time
import uvicorn
import os
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
# 记录请求和响应的详细信息
log_message = (
f"请求方法: {request.method}\n"
f"请求路径: {request.url.path}\n"
f"响应状态: {response.status_code}\n"
f"处理时间: {process_time:.4f} 秒"
)
# 在生产环境中，可以将这些信息写入日志文件
print(log_message)
return response
app = FastAPI()
# 添加中间件到应用中
app.add_middleware(LoggingMiddleware)
# 使用 PlainTextResponse 返回简单文本响应
@app.get("/", response_class=PlainTextResponse)
async def read_root():
# 可以直接返回字符串，FastAPI 会自动封装为 PlainTextResponse
return "Hello, World!"
- SessionMiddleware
# 使用 SessionMiddleware 来管理用户会话，这对于需要追踪用户状态或者保持登录状态的应用尤为重要
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from starlette.middleware.sessions import SessionMiddleware
import uvicorn
import os
app = FastAPI()
# 设置 SessionMiddleware，secret_key 应该是一个长随机字符串
# 使用 SessionMiddleware 来管理会话。secret_key 用于签名和/或加密会话 cookie，确保它的安全性
app.add_middleware(
SessionMiddleware, secret_key="!se1cret2-ke3y-sh4ould-b5e-ve8ry-se8cure!"
)
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
# 访问会话中的数据
count = request.session.get("count", 0)
count += 1
request.session["count"] = count # 更新会话数据
return f"<html><body><h1>Visit Count: {count}</h1></body></html>"
@app.get("/reset")
async def reset_count(request: Request):
request.session.pop("count", None) # 重置会话数据
return {"status": "session reset"}
@app.get("/logout")
async def logout(request: Request):
request.session.clear() # 清空所有会话数据
return {"status": "logged out"}
if __name__ == "__main__":
uvicorn.run(
f"{os.path.basename(__file__).split('.')[0]}:app",
host="127.0.0.1",
port=8000,
reload=True,
)
- TrustedHostMiddleware
# 使用 TrustedHostMiddleware 可以提高应用的安全性，限制哪些主机名可以访问应用
from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI()
# 添加 TrustedHostMiddleware
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["example.com", "www.example.com", "*.example.com"]
)
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}
- CORSMiddleware
# 方式一
@app.middleware("http")
async def CORSMiddleware(request: Request, call_next):
response = await call_next(request)
print(response.headers)
return response
# 方式二
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
"http://localhost:63342"
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # *：代表所有客户端
allow_credentials=True,
allow_methods=["GET"],
allow_headers=["*"],
)
@app.get("/")
def main():
return {"message": "Hello World"}
if __name__ == '__main__':
import uvicorn
uvicorn.run("main:app", host="127.0.0.1", port=8080, debug=True, reload=True)
# FastAPI-MCP
FastAPI-MCP是一个零配置工具，旨在将现有的FastAPI端点自动转换为Model Context Protocol(MCP)工具
pip install fastapi-mcp
提供以下核心功能：
零配置集成
# 只需几行代码，就能将 FastAPI 应用转化为 MCP 服务器，自动生成 /mcp 端点，供 AI 模型发现和调用
原生 FastAPI 支持
# 依赖注入: 使用 FastAPI 的 Depends() 机制为 MCP 端点添加认证和授权，确保安全性
ASGI 传输
# 直接利用 FastAPI 的 ASGI 接口，避免额外的 HTTP 调用，提升性能
保留 schema 和文档
# 自动保留 FastAPI 的请求/响应 schema 和 Swagger 文档，AI 模型能够轻松理解端点功能
灵活部署
# 支持将 MCP 服务器挂载到现有 FastAPI 应用，或独立部署，适应不同场景需求
支持流式 HTTP 传输
# 从25年3月的 MCP 规范更新开始，FastAPI-MCP 支持流式HTTP传输（推荐），同时保留对 SSE 的兼容性
认证支持
# 通过FastAPI的依赖注入机制，如get_auth_token，提供OAuth2.1兼容的认证功能，确保授权后才能访问
可扩展性
# 允许开发者添加自定义 MCP 工具，与自动生成的工具并存，满足复杂业务需求
# 基础使用
- 如何将 FastAPI 端点暴露为 MCP 工具
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
# 1. 从你的 FastAPI app 创建一个 MCP 实例
mcp = FastApiMCP(app, name="My API MCP",
description="My API description",
base_url="http://localhost:8000")
# 2. 将 MCP 服务挂载到你的 app 上
mcp.mount()
@app.get("/ping")
async def ping():
return {"pong": True}
# 运行后访问 http://localhost:8000/mcp 将返回 MCP 工具的 JSON 描述，AI模型可据此调用 /ping 端点
- 无缝集成现有认证，让 LLM 在调用这个工具时也带上认证信息
from fastapi import Depends, FastAPI
from fastapi.security import HTTPBearer
app = FastAPI()
token_auth_scheme = HTTPBearer()
# 一个受保护的接口，必须提供有效的 Bearer Token
@app.get("/private")
async def private(token: str = Depends(token_auth_scheme)):
return {"message": f"Your token is: {token.credentials}"}
# 增加MCP的功能
from fastapi_mcp import FastApiMCP, AuthConfig
# ... (前面的 app 和 token_auth_scheme 定义)
# 创建 MCP 服务时，将 FastAPI 中已有的 Depends(token_auth_scheme) 传递给 AuthConfig
# 通过 AuthConfig 告诉 FastApiMCP 需要哪些认证依赖
mcp = FastApiMCP(
app,
name="Protected MCP",
auth_config=AuthConfig(
dependencies=[Depends(token_auth_scheme)], # 把认证依赖传进去
),
)
# 挂载服务
mcp.mount_http()
- 选择性暴露的 API
# 仅暴露 operation_id 为 "get_item" 和 "list_items" 的接口
include_operations_mcp = FastApiMCP(
app,
name="Item API MCP - Included Operations",
include_operations=["get_item", "list_items"],
)
# 暴露除了带有 "search" 标签以外的所有接口
exclude_tags_mcp = FastApiMCP(
app,
name="Item API MCP - Excluded Tags",
exclude_tags=["search"],
)
# 挂载不同的 MCP 服务到不同的路径
include_operations_mcp.mount_http(mount_path="/include-operations-mcp")
exclude_tags_mcp.mount_http(mount_path="/exclude-tags-mcp")
ORM →