如何将Azure绑定接入FastAPI端点及传递starter字符串

方案思路 #

避免使用trace_context.attributes传递starter（该字段原本用于追踪场景，不适合业务数据传递），改为在Azure HTTP触发器中预先创建DurableOrchestrationClient实例，通过ASGI请求的scope传递给FastAPI，再通过依赖注入供接口使用。同时手动构建符合要求的HTTP 202响应（带Location状态查询头）。

修改后的完整代码 #

1. 触发器/中间件部分（传递Durable客户端）

import azure.functions as func
import azure.durable_functions as df
from fastapi import FastAPI, Request, Depends, responses

fastapi_app = FastAPI()

@df_app.route(route="{*route}", auth_level=func.AuthLevel.ANONYMOUS)
@df_app.generic_input_binding(arg_name="starter", type="durableClient")
async def http_trigger(
    req: func.HttpRequest, context: func.Context, starter: str
) -> func.HttpResponse:
    # 直接创建Durable客户端，并存入ASGI请求的scope中
    durable_client = df.DurableOrchestrationClient(starter)
    req.scope["durable_client"] = durable_client
    
    # 委托FastAPI处理请求
    response: func.HttpResponse = await func.AsgiMiddleware(fastapi_app).handle_async(
        req, context
    )
    return response

2. FastAPI依赖项（获取Durable客户端）

def get_durable_client(request: Request) -> df.DurableOrchestrationClient:
    # 从scope中直接获取预先创建好的客户端实例
    return request.scope["durable_client"]

3. FastAPI接口（启动编排并返回标准响应）

@fastapi_app.get(path="/fast_orchestrator")
async def fast_orchestrator(
    client: df.DurableOrchestrationClient = Depends(get_durable_client)
):
    instance_id = await client.start_new("my_orchestrator")
    # 获取编排实例的状态查询URL
    status_query_url = client.get_status_query_url(instance_id)
    
    # 构建符合Azure Durable要求的HTTP 202响应
    return responses.JSONResponse(
        content={
            "instance_id": instance_id,
            "message": f"Started orchestration with ID = '{instance_id}'."
        },
        status_code=202,
        headers={"Location": status_query_url}
    )

方案优势 #

• 更合理的数据传递：不再滥用追踪上下文传递业务数据，符合Azure Functions和FastAPI的设计规范
• 减少重复代码：在触发器中一次性创建Durable客户端，避免在每个接口中重复实例化
• 标准响应格式：手动构建的202响应与Azure原生create_check_status_response行为一致，包含状态查询的Location头

内容的提问来源于stack exchange，提问作者André Ricardo