AWS Lambda调用Xero API报‘Missing access token’错误求助

核心问题 #

你的代码在创建OAuth2Session时，传入的token字典缺少**token_type**字段——requests-oauthlib需要这个字段生成正确的Authorization请求头（格式应为Bearer <access_token>）。当前仅传入access_token会导致请求头未正确设置，Xero API因此返回"Missing access token."错误。

此外，代码还有几处潜在问题需要修正：

• logger未初始化，会触发运行时报错
• 未处理API请求失败场景（如token获取失败、API返回非200状态码）
• 使用refresh_token获取新token时，Xero响应会返回新的refresh_token，未保存更新会导致下次刷新失效

修正后的代码 #

import base64
import requests
import json
import boto3
from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session
import logging

# 初始化日志工具
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def get_xero_token(client_id, client_secret, refresh_token=None):
    token_url = 'https://identity.xero.com/connect/token'

    # 用refresh_token获取新token
    if refresh_token:
        headers = {
            'Authorization': 'Basic ' + base64.b64encode((client_id + ':' + client_secret).encode('utf-8')).decode('utf-8'),
            'Content-Type': 'application/x-www-form-urlencoded'
        }

        data = {
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token
        }

        response = requests.post(token_url, headers=headers, data=data)
        response.raise_for_status()  # 抛出HTTP请求错误
        token_data = response.json()
        logger.info(f"刷新token响应: {token_data}")
        # 返回完整token数据，包含access_token、token_type、refresh_token等字段
        return token_data

    else:
        # 后端应用凭证模式（无需用户授权）
        client = BackendApplicationClient(client_id=client_id)
        oauth = OAuth2Session(client=client)
        token = oauth.fetch_token(
            token_url=token_url,
            client_id=client_id,
            client_secret=client_secret
        )
        logger.info(f"初始token响应: {token}")
        return token


def lambda_handler(event, context):
    # Xero凭证（建议用AWS Secrets Manager存储，禁止硬编码）
    client_id = '{insert}'
    client_secret = '{insert}'
    refresh_token = '{insert}'

    # 获取完整token数据
    token = get_xero_token(client_id, client_secret, refresh_token)
    
    # 校验token有效性
    if not token.get('access_token'):
        raise ValueError("无法获取有效access token")
    
    logger.info(f'使用access token: {token["access_token"]}')

    # 创建OAuth2Session，传入完整token字典
    oauth = OAuth2Session(client_id=client_id, token=token)

    # 初始化S3资源
    s3 = boto3.resource('s3')
    bucket_name = 'xerotos3datatransfer'

    # Xero API端点与对应S3文件名映射
    xero_api_configs = [
        ('https://api.xero.com/api.xro/2.0/Invoices', 'xero_invoices_data.json'),
        ('https://api.xero.com/api.xro/2.0/Accounts', 'xero_accounts_data.json')
    ]

    for url, file_name in xero_api_configs:
        # 调用Xero API
        response = oauth.get(url)
        response.raise_for_status()  # 处理API请求错误
        data = response.json()

        # 上传数据到S3
        s3.Bucket(bucket_name).put_object(Key=file_name, Body=json.dumps(data))
        logger.info(f"已成功上传{file_name}到S3")

    return {
        'statusCode': 200,
        'body': json.dumps('数据已成功从Xero传输到S3！')
    }

关键修正点说明 #

1. 传递完整token字典：get_xero_token现在返回Xero接口返回的完整token数据，OAuth2Session会自动利用这些字段生成正确的Authorization请求头。
2. 初始化日志工具：添加logging模块初始化，解决原代码中logger未定义的问题。
3. 增加错误处理：加入response.raise_for_status()，在HTTP请求失败时直接抛出异常，便于快速排查问题。
4. refresh_token更新处理：使用refresh_token获取新token时，Xero会返回新的refresh_token，建议将其保存到安全存储（如AWS Secrets Manager），避免下次刷新失效。
5. 凭证安全优化：禁止硬编码凭证，推荐使用AWS Secrets Manager存储和读取，提升应用安全性。

额外注意事项 #

• 若Xero演示公司需要用户授权（而非后端应用模式），需改用授权码流程（Authorization Code Flow），调整get_xero_token逻辑处理用户授权回调，获取初始refresh_token。
• 确保Lambda函数具备访问S3存储桶的IAM权限，以及访问Xero API的网络权限（若Lambda在VPC内，需配置NAT网关或VPC端点）。

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