You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

aiohttp配置客户端mTLS证书请求接口返回504超时如何解决

修复方案

核心问题定位

你遇到的504超时和SSL配置无关,日志中已经明确显示SSL握手完成(耗时188.5ms),问题出在请求头和请求体的配置错误:

  • 硬编码错误的Content-Length请求头,导致服务端一直等待未发送的请求体数据,直到超时返回504
  • 请求体传参方式和声明的Content-Type不匹配
  • 存在变量引用笔误

具体修改点

  1. 删除硬编码的Content-Length字段
    HTTP客户端会自动根据实际请求体长度计算该值,手动指定错误值会导致服务端等待超时,这是requests和aiohttp表现不一致的核心原因:requests会自动修正用户传入的不合理Content-Length,aiohttp会严格遵循用户配置。
  2. 修正请求体传参方式
    接口要求JSON格式请求体,使用json=payload替代data=payload,aiohttp会自动完成JSON序列化,同时自动设置正确的Content-Type: application/json请求头,无需手动声明。
  3. 修复请求头引用笔误
    创建ClientSession时传入self.headers而非未定义的全局headers变量。
  4. 可选优化:复用SSL上下文
    将SSL上下文初始化移到构造函数中,避免每次请求重复加载证书,提升性能。

修复后代码

import ssl
import os
import asyncio
import aiohttp
import logging

logger = logging.getLogger(__name__)

class Test:
     def __init__(self, url: str):
        self.url = url
        self.api_creds = {"api_key": os.getenv("API_KEY"), "api_secret": os.getenv("API_SECRET")}
        # 删除硬编码的Content-Length,json参数会自动生成正确的Content-Type
        self.headers = {"Content-Type": "application/json;charset=utf-8"}
        # 初始化时一次生成SSL上下文复用
        self.sslcontext = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        self.sslcontext.load_cert_chain(certfile='Path to mtls.crt', keyfile='Path .key')

     async def _post(self, path, payload):
        url = self.url + path
        # 传入正确的实例属性self.headers
        async with aiohttp.ClientSession(headers=self.headers) as session:
            try:
                # 使用json参数传JSON payload,不要用data
                response = await session.post(url, json=payload, ssl=self.sslcontext)
                # 可选:提前读取响应内容,避免连接泄漏
                await response.read()
            except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError, aiohttp.ClientError) as e:
                logger.error(str(e))
                raise
        return response

    async def login(self):
        response = await self._post('/login', self.api_creds)
        print(response)
    

async def main():
  t = Test("https://EP url")
  await t.login()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

内容的提问来源于stack exchange,提问作者Igor Zilberman

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.27 16:57:05