FastAPI携带Bearer Token调用外部API遇401未授权及JSON解码错误求助
看起来你遇到了两个关联的问题——调用外部API时的401未授权,以及后续触发的JSON解码错误。我来帮你一步步拆解分析,给出具体的排查和解决方向:
一、先解决JSONDecodeError的根源
你看到的json.decoder.JSONDecodeError其实是401未授权后的连锁问题。当外部API返回401状态时,它的响应内容大概率不是合法的JSON(可能是HTML错误页面、纯文本提示,甚至是空内容),这时候你直接调用response_api.json()就会触发解码失败。
你可以验证这一点:看打印的response_api.text,如果是401状态,输出的内容肯定不是标准JSON格式——这就是解码错误的真正原因,和Token本身是否合法无关。
二、401未授权的核心排查点
你用Postman/Insomnia能成功,但代码里不行,说明代码中的请求和手动测试的请求存在关键差异。以下是几个高概率的问题点和解决办法:
1. 去掉GET请求中多余的Content-Type头部
GET请求没有请求体,完全不需要Content-Type: application/json头部。有些外部API会严格校验请求头部,多余的Content-Type可能会被判定为非法请求,直接拒绝授权。
2. 对Token做去空格处理
虽然你打印的Token看起来正常,但有时候JSON返回的字符串可能带有不可见的空格(比如换行符、制表符),导致Bearer {access_token}格式无效。可以给Token加个去空格处理:
access_token = response.json().get("access_token", "").strip()
3. 避免同一个AsyncClient的Cookie干扰
你在同一个AsyncClient实例中先请求Token API,再请求外部API,httpx会自动保存Token API返回的Set-Cookie头部,并在后续请求中自动携带。如果外部API和Token API是不同服务,这些多余的Cookie可能会导致授权失败。
解决办法:用两个独立的AsyncClient实例分别处理Token请求和外部API请求:
# 单独获取Token async with httpx.AsyncClient(verify=False) as token_client: token_response = await token_client.get('https://myTokenApi.com/token') access_token = token_response.json().get("access_token", "").strip() # 单独调用外部API async with httpx.AsyncClient(verify=False) as api_client: response_api = await api_client.get( f"https://externalApi.com/email/{email}", headers={"Authorization": f"Bearer {access_token}"} )
4. 对比请求头部,和Postman完全对齐
把代码中要发送的请求头部打印出来,和Postman里的请求头部做对比,看是否有差异:
api_headers = {"Authorization": f"Bearer {access_token}"} print(f"发送的请求头部: {api_headers}")
如果Postman里有默认的User-Agent等头部,也可以加到代码的请求头部中,确保和手动请求完全一致。
5. 硬编码Token做测试
把你在Postman里用的有效Token硬编码到代码中,比如:
access_token = "你在Postman里复制的有效Token字符串"
如果这样能成功,说明你获取Token的过程有隐藏问题(比如Token API返回的内容有特殊字符、解析逻辑有疏漏);如果还是401,说明是请求头部或客户端配置的问题。
三、修改后的完整代码示例
我把上述优化点整合到代码中,同时添加了更完善的错误处理和调试信息:
from fastapi import FastAPI import httpx app = FastAPI() @app.get('/query') async def getToken(): email = "myEmail@provider.com" access_token = "" # 第一步:独立获取Token try: async with httpx.AsyncClient(verify=False) as token_client: token_response = await token_client.get('https://myTokenApi.com/token') token_response.raise_for_status() # 确保Token请求成功 token_data = token_response.json() access_token = token_data.get("access_token", "").strip() print(f"获取到的Token(前10位): {access_token[:10]}...") except httpx.HTTPStatusError as e: print(f"获取Token失败: {e}") return {"error": "获取Token失败", "details": str(e)} except ValueError as e: print(f"Token响应不是合法JSON: {e}, 响应内容: {token_response.text}") return {"error": "Token响应解析失败", "details": token_response.text} # 第二步:独立调用外部API try: async with httpx.AsyncClient(verify=False) as api_client: api_headers = {"Authorization": f"Bearer {access_token}"} print(f"发送的请求头部: {api_headers}") response_api = await api_client.get( f"https://externalApi.com/email/{email}", headers=api_headers ) print(f"外部API响应状态码: {response_api.status_code}") print(f"外部API响应内容(前200位): {response_api.text[:200]}...") # 根据状态码处理响应 if response_api.status_code in (200, 201): return response_api.json() else: return { "error": "外部API请求失败", "status_code": response_api.status_code, "details": response_api.text } except httpx.HTTPStatusError as e: print(f"外部API请求错误: {e}") return {"error": "外部API请求失败", "details": str(e)} except ValueError as e: print(f"外部API响应不是合法JSON: {e}") return {"error": "外部API响应解析失败", "details": response_api.text} if __name__ == '__main__': import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=3000, log_level="info", reload=True)
按照上面的步骤排查,应该能快速定位到问题。核心思路是:先确保Token获取和格式完全正确,再让代码中的请求和手动测试的请求完全一致(包括头部、客户端状态等)。
备注:内容来源于stack exchange,提问作者c0d3rbox

