FastAPI中间件报错：TypeError: object str can't be used in 'await' expression

核心错误原因 #

你直接将gr.Interface实例作为ASGI应用传给Mount，但gr.Interface本身不是标准ASGI应用，中间件调用它时返回了字符串类型，触发TypeError: object str can't be used in 'await' expression。此外代码还存在导入缺失、同步阻塞调用、异常处理不规范的问题。

分步修复 #

1. 正确转换Gradio应用为ASGI兼容格式 #

将Gradio Interface转换为ASGI应用，使用asgi_app()方法：

io = gr.Interface(lambda x: "Hello, " + x + "!", "textbox", "textbox").asgi_app()

2. 补全缺失的导入 #

添加Request和RedirectResponse的导入：

from starlette.requests import Request
from starlette.responses import RedirectResponse

3. 处理Stripe同步调用（避免阻塞事件循环） #

Stripe的同步API调用会阻塞FastAPI的异步事件循环，需用FastAPI提供的run_in_threadpool包装：

from fastapi.concurrency import run_in_threadpool

async def dispatch(self, request: Request, call_next):
    # 用线程池包装同步Stripe调用
    customer = await run_in_threadpool(
        stripe.Customer.retrieve,
        request.session.get("user")
    )
    r = await run_in_threadpool(
        stripe.Subscription.list,
        customer=customer.id,
        limit=3
    )

4. 替换中间件中的HTTPException为响应对象 #

中间件中直接抛出HTTPException无法被FastAPI的异常处理器正确捕获，应返回RedirectResponse或标准响应对象：

if r.data[0].status == "incomplete":
    return RedirectResponse(url="/gradio")

完整修复后的代码 #

from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.routing import Mount
from starlette.requests import Request
from starlette.responses import RedirectResponse
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
import stripe
import gradio as gr

# 初始化Stripe API密钥
stripe.api_key = "你的Stripe密钥"

class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        user_id = request.session.get("user")
        if not user_id:
            return RedirectResponse(url="/gradio")
        
        # 用线程池处理同步Stripe调用
        customer = await run_in_threadpool(
            stripe.Customer.retrieve,
            user_id
        )
        subscriptions = await run_in_threadpool(
            stripe.Subscription.list,
            customer=customer.id,
            limit=3
        )
        
        if subscriptions.data and subscriptions.data[0].status == "incomplete":
            return RedirectResponse(url="/gradio")
        
        response = await call_next(request)
        return response  

# 将Gradio Interface转换为ASGI应用
io = gr.Interface(lambda x: "Hello, " + x + "!", "textbox", "textbox").asgi_app()

middleware = [
    Middleware(CustomHeaderMiddleware)
]

routes = [
    Mount('/gradio', app=io, middleware=middleware),
]
app = FastAPI(routes=routes)

额外注意事项 #

• 生产环境中建议添加Stripe调用的异常捕获，避免API请求失败导致服务崩溃
• 若不需要自定义路由结构，可直接用app.mount("/gradio", io, middleware=middleware)简化代码

内容的提问来源于stack exchange，提问作者shivam jha