如何在Azure Durable Functions中实现HTTP响应并返回活动函数结果？

错误原因 #

你碰到的TypeError: Object of type AtomicTask is not JSON serializable，核心问题出在编排器函数里：调用context.call_activity()时，直接把返回的任务对象赋值给了result，而非等待任务执行完成拿到实际返回值。call_activity()返回的是AtomicTask类型的任务实例，这个对象没法被JSON序列化，自然就会在返回结果或设置自定义状态时出错。

核心修复：编排器中用yield获取活动函数结果 #

修改编排器函数，调用活动函数时必须通过yield关键字等待任务完成，才能拿到活动函数返回的实际字符串：

@app.orchestration_trigger(context_name="context")
def orchestrator(context: df.DurableOrchestrationContext) -> str:
    parameters = context.get_input()
    process = parameters.get("process")

    result = None
    if process == 0:
        # 关键：用yield等待活动函数执行完毕，获取真实返回值
        result = yield context.call_activity("failed")
    else:
        result = "Executed successfully. Pass a char and string in the query or in the request body."

    context.set_custom_status(result)
    return result

客户端函数优化（可选） #

原客户端函数里，wait_for_completion_or_create_check_status_response()本身就会返回合法的HTTP响应，没必要再手动调用get_status()。可以简化逻辑直接返回该方法的结果：

@app.route(route="orchestrators/client_function")
@app.durable_client_input(client_name="client")
async def client_function(req: func.HttpRequest, client: df.DurableOrchestrationClient) -> func.HttpResponse:
    # 获取请求参数
    process = req.params.get("process")
    if process is None:
        try:
            process = req.get_json().get('process')
        except (ValueError, KeyError):
            process = 0
    process = int(process)

    instance_id = await client.start_new("orchestrator", None, {"process": process,})
    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    
    # 直接返回等待完成后的响应，包含执行结果
    return await client.wait_for_completion_or_create_check_status_response(
        req, instance_id, timeout_in_seconds=30
    )

如果需要自定义返回格式，修复编排器后status.output会变成活动函数返回的字符串，可以这样处理：

@app.route(route="orchestrators/client_function")
@app.durable_client_input(client_name="client")
async def client_function(req: func.HttpRequest, client: df.DurableOrchestrationClient) -> func.HttpResponse:
    # 参数获取逻辑同上
    process = req.params.get("process")
    if process is None:
        try:
            process = req.get_json().get('process')
        except (ValueError, KeyError):
            process = 0
    process = int(process)

    instance_id = await client.start_new("orchestrator", None, {"process": process,})
    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    
    # 等待编排完成
    await client.wait_for_completion_or_create_check_status_response(req, instance_id)
    
    status = await client.get_status(instance_id)
    runtime = status.runtime_status
    output = status.output  

    # 返回格式化的纯文本响应
    return func.HttpResponse(
        f"runtime: {runtime}\n\noutput: {output}",
        status_code=200,
        mimetype="text/plain"
    )

验证修复效果 #

1. 重新部署函数到Azure Free Tier
2. 发送测试请求：
   • GET请求：https://<你的函数应用名>.azurewebsites.net/api/orchestrators/client_function?process=0
   • 预期返回：

     runtime: Completed
     
     output: failed_function executed successfully.
     

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