能否使用单个AWS Lambda函数调用多个API？如何传参指定处理流程？

一、当然可以用单个Lambda调用多个API #

Lambda本质是无服务器的计算运行环境，只要你的代码里引入HTTP客户端（比如Python的requests、Node.js的axios），就能调用任意数量的外部API——不管是AWS自家的服务API（比如DynamoDB、S3）还是第三方公共API。

举个简单的Python示例，在一个Lambda里调用两个不同的外部API：

import requests

def lambda_handler(event, context):
    # 调用第一个API
    api1_response = requests.get("https://api.example.com/service-a")
    # 调用第二个API
    api2_response = requests.post("https://api.example.com/service-b", json={"key": "value"})
    
    return {
        "statusCode": 200,
        "body": {
            "api1_result": api1_response.json(),
            "api2_result": api2_response.json()
        }
    }

二、整合多端点处理流程，通过参数指定执行逻辑 #

要把多个业务流程整合到一个Lambda里，核心思路是通过传入的参数做分支逻辑判断，让Lambda知道该执行哪段代码。下面分常见场景说明：

1. 确定参数传递方式 #

根据Lambda的触发方式，参数可以通过不同途径传入：

• API Gateway触发：可以用路径参数（比如/api/{action}）、查询参数（比如?action=create_order），或者请求体里的字段（比如{"action": "update_user"}）。
• 直接调用Lambda：在调用时把标识参数放在event对象里（比如client.invoke(FunctionName='my-lambda', Payload=json.dumps({"action": "process_refund"}))）。

2. 编写分支逻辑代码 #

把每个业务流程封装成独立函数，然后在Lambda入口函数里根据参数分发任务。以下是Python的完整示例：

import requests
import json

def lambda_handler(event, context):
    # 从不同来源获取action参数
    action = None
    # 优先从请求体获取（如果是POST请求）
    if event.get('body'):
        body = json.loads(event['body'])
        action = body.get('action')
    # 其次从查询参数获取（比如GET请求）
    if not action and event.get('queryStringParameters'):
        action = event['queryStringParameters'].get('action')
    # 最后从路径参数获取（如果API Gateway配置了路径变量）
    if not action and event.get('pathParameters'):
        action = event['pathParameters'].get('action')

    # 根据action执行对应流程
    if action == 'fetch_user':
        return fetch_user(event)
    elif action == 'create_order':
        return create_order(event)
    elif action == 'process_refund':
        return process_refund(event)
    else:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": "Invalid action specified"})
        }

# 流程1：获取用户信息（调用用户API）
def fetch_user(event):
    user_id = event['queryStringParameters']['user_id']
    response = requests.get(f"https://api.example.com/users/{user_id}")
    return {
        "statusCode": response.status_code,
        "body": json.dumps(response.json())
    }

# 流程2：创建订单（先调用库存API，再调用订单API）
def create_order(event):
    order_data = json.loads(event['body'])
    # 检查库存
    inventory_res = requests.get(f"https://api.example.com/inventory/{order_data['product_id']}")
    if inventory_res.json()['stock'] <= 0:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": "Out of stock"})
        }
    # 创建订单
    order_res = requests.post("https://api.example.com/orders", json=order_data)
    return {
        "statusCode": order_res.status_code,
        "body": json.dumps(order_res.json())
    }

# 流程3：处理退款（调用退款API）
def process_refund(event):
    refund_id = event['pathParameters']['refund_id']
    response = requests.post(f"https://api.example.com/refunds/{refund_id}")
    return {
        "statusCode": response.status_code,
        "body": json.dumps(response.json())
    }

3. 最佳实践提醒 #

• 代码模块化：每个流程封装成独立函数，便于单独测试和维护。
• 错误处理：给每个API调用添加异常捕获（比如try-except），避免单个API调用失败导致整个Lambda崩溃。
• 权限配置：如果调用AWS内部服务，要给Lambda对应的IAM权限（比如调用DynamoDB需要dynamodb:PutItem权限）；如果调用外部API，确保Lambda所在VPC能访问公网（配置NAT网关或使用Lambda的公网访问权限）。
• 冷启动考量：如果整合的流程资源需求差异很大（比如一个是轻量查询，一个是大数据处理），建议拆分Lambda；如果都是轻量级流程，整合在一起可以减少维护成本。

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