树莓派Pico W调用Azure ML REST端点推理无响应问题排查

API 探路者

2026-7-11

树莓派Pico W调用Azure ML REST端点无响应问题排查与解决

问题背景

在Python 3.10环境下可正常调用Azure ML模型的REST端点完成推理，但在Raspberry Pi Pico W的MicroPython环境中，使用urllib.urequests.urlopen()返回空结果，也无法像标准Python那样结合Request对象使用。试过urequests、urllib.urequests、uurequests三个库均无效，调用response.content时返回空值或超时，其内部的raw.read()执行失败。

可正常运行的Python代码

import urllib.request
import json
import os
import ssl

def allowSelfSignedHttps(allowed):
    # 绕过客户端的服务器证书验证
    if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None):
        ssl._create_default_https_context = ssl._create_unverified_context

allowSelfSignedHttps(True) # 如果评分服务使用自签名证书，需要此行代码

# 请求数据
data =  {
  "input_data": {
    "columns": [
      "PLAT_PH",
      "PLAT_ORP",
      "PSU1",
      "PSU2",
      "PSU3"
    ],
    "index": [0],
    "data": [[3.0, 600, 4.0, 5.0, 0]]
  }
}

body = str.encode(json.dumps(data))

url = 'my-azure-endpoint-url'
# 替换为端点的主/次密钥或AMLToken
api_key = 'my-api-key'
if not api_key:
    raise Exception("调用端点需要提供密钥")

# azureml-model-deployment头部会强制请求指向特定部署
# 删除此头部将遵循端点流量规则
headers = {'Content-Type':'application/json', 'Authorization':('Bearer '+ api_key), 'azureml-model-deployment': 'platypus-ph-orp-ps-yn-02-1' }

req = urllib.request.Request(url, body, headers)

try:
    response = urllib.request.urlopen(req)
    result = response.read()
    print(result)
except urllib.error.HTTPError as error:
    print("请求失败，状态码: " + str(error.code))
    # 打印头部（包含请求ID和时间戳，有助于调试失败原因）
    print(error.info())
    print(error.read().decode("utf8", 'ignore'))

尝试的MicroPython代码

body = str.encode(json.dumps(data))
headers = {'Content-Type':'application/json', 'Authorization':('Bearer '+ api_key), 'azureml-model-deployment': 'my-azure-deployment' }

response = urequests.request('POST', url, data=body, headers=headers)
print("response: ", response)
result = response.content
print("result: ", result)

使用的urequests库源码

import usocket

class Response:

    def __init__(self, f):
        self.raw = f
        self.encoding = "utf-8"
        self._cached = None

    def close(self):
        if self.raw:
            self.raw.close()
            self.raw = None
        self._cached = None

    @property
    def content(self):
        if self._cached is None:
            try:
                self._cached = self.raw.read()
            finally:
                self.raw.close()
                self.raw = None
        return self._cached

    @property
    def text(self):
        return str(self.content, self.encoding)

    def json(self):
        import ujson
        return ujson.loads(self.content)


def request(method, url, data=None, json=None, headers={}, stream=None):
    try:
        proto, dummy, host, path = url.split("/", 3)
    except ValueError:
        proto, dummy, host = url.split("/", 2)
        path = ""
    if proto == "http:":
        port = 80
    elif proto == "https:":
        import ussl
        port = 443
    else:
        raise ValueError("Unsupported protocol: " + proto)

    if ":" in host:
        host, port = host.split(":", 1)
        port = int(port)

    ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
    ai = ai[0]

    s = usocket.socket(ai[0], ai[1], ai[2])
    try:
        s.connect(ai[-1])
        if proto == "https:":
            s = ussl.wrap_socket(s, server_hostname=host)
        s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
        if not "Host" in headers:
            s.write(b"Host: %s\r\n" % host)
        # Iterate over keys to avoid tuple alloc
        for k in headers:
            s.write(k)
            s.write(b": ")
            s.write(headers[k])
            s.write(b"\r\n")
        if json is not None:
            assert data is None
            import ujson
            data = ujson.dumps(json)
            s.write(b"Content-Type: application/json\r\n")
        if data:
            s.write(b"Content-Length: %d\r\n" % len(data))
        s.write(b"\r\n")
        if data:
            s.write(data)

        l = s.readline()
        #print(l)
        l = l.split(None, 2)
        status = int(l[1])
        reason = ""
        if len(l) > 2:
            reason = l[2].rstrip()
        while True:
            l = s.readline()
            if not l or l == b"\r\n":
                break
            #print(l)
            if l.startswith(b"Transfer-Encoding:"):
                if b"chunked" in l:
                    raise ValueError("Unsupported " + l)
            elif l.startswith(b"Location:") and not 200 <= status <= 299:
                raise NotImplementedError("Redirects not yet supported")
    except OSError:
        s.close()
        raise

    resp = Response(s)
    resp.status_code = status
    resp.reason = reason
    return resp

def head(url, **kw):
    return request("HEAD", url, **kw)

def get(url, **kw):
    return request("GET", url, **kw)

def post(url, **kw):
    return request("POST", url, **kw)

def put(url, **kw):
    return request("PUT", url, **kw)

def patch(url, **kw):
    return request("PATCH", url, **kw)

def delete(url, **kw):
    return request("DELETE", url, **kw)

问题分析与修复方案

核心问题点

HTTPS证书验证缺失：原Python代码禁用了自签名证书验证，但MicroPython默认会验证证书，导致连接失败
HTTP协议版本兼容性：当前urequests使用HTTP/1.0，Azure端点对HTTP/1.1支持更完善
字符串与字节类型不兼容：发送请求时未将字符串类型的头部、方法转换为字节，导致MicroPython执行异常
响应读取可靠性差：一次性读取整个响应容易超时或失败
无超时设置与异常捕获：无法定位连接超时、证书错误等具体问题

修复后的urequests库代码

import usocket

class Response:

    def __init__(self, f):
        self.raw = f
        self.encoding = "utf-8"
        self._cached = None

    def close(self):
        if self.raw:
            self.raw.close()
            self.raw = None
        self._cached = None

    @property
    def content(self):
        if self._cached is None:
            try:
                # 分块读取响应，提升可靠性
                self._cached = b""
                while True:
                    chunk = self.raw.read(1024)
                    if not chunk:
                        break
                    self._cached += chunk
            finally:
                self.raw.close()
                self.raw = None
        return self._cached

    @property
    def text(self):
        return str(self.content, self.encoding)

    def json(self):
        import ujson
        return ujson.loads(self.content)


def request(method, url, data=None, json=None, headers={}, stream=None, timeout=10):
    try:
        proto, dummy, host, path = url.split("/", 3)
    except ValueError:
        proto, dummy, host = url.split("/", 2)
        path = ""
    if proto == "http:":
        port = 80
        use_ssl = False
    elif proto == "https:":
        import ussl
        port = 443
        use_ssl = True
    else:
        raise ValueError("不支持的协议: " + proto)

    if ":" in host:
        host, port = host.split(":", 1)
        port = int(port)

    ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
    ai = ai[0]

    s = usocket.socket(ai[0], ai[1], ai[2])
    s.settimeout(timeout)
    try:
        s.connect(ai[-1])
        if use_ssl:
            # 禁用证书验证，匹配原Python代码逻辑
            s = ussl.wrap_socket(s, server_hostname=host, cert_reqs=ussl.CERT_NONE)
        # 使用HTTP/1.1协议，提升兼容性
        s.write(b"%s /%s HTTP/1.1\r\n" % (method.encode(), path.encode()))
        if not "Host" in headers:
            s.write(b"Host: %s\r\n" % host.encode())
        # 确保所有头部都是字节类型
        for k, v in headers.items():
            s.write(k.encode() if isinstance(k, str) else k)
            s.write(b": ")
            s.write(v.encode() if isinstance(v, str) else v)
            s.write(b"\r\n")
        if json is not None:
            assert data is None
            import ujson
            data = ujson.dumps(json)
            if "Content-Type" not in headers:
                s.write(b"Content-Type: application/json\r\n")
        if data:
            if isinstance(data, str):
                data = data.encode()
            s.write(b"Content-Length: %d\r\n" % len(data))
        # 添加HTTP/1.1必需的连接关闭头部
        s.write(b"Connection: close\r\n")
        s.write(b"\r\n")
        if data:
            s.write(data)

        l = s.readline()
        if not l:
            raise OSError("读取响应头失败")
        l = l.split(None, 2)
        status = int(l[1])
        reason = ""
        if len(l) > 2:
            reason = l[2].rstrip().decode()
        while True:
            l = s.readline()
            if not l or l == b"\r\n":
                break
            if l.startswith(b"Transfer-Encoding:"):
                if b"chunked" in l:
                    raise ValueError("不支持分块编码: " + l.decode())
            elif l.startswith(b"Location:") and not 200 <= status <= 299:
                raise NotImplementedError("暂不支持重定向")
    except OSError as e:
        s.close()
        raise e

    resp = Response(s)
    resp.status_code = status
    resp.reason = reason
    return resp

def head(url, **kw):
    return request("HEAD", url, **kw)

def get(url, **kw):
    return request("GET", url, **kw)

def post(url, **kw):
    return request("POST", url, **kw)

def put(url, **kw):
    return request("PUT", url, **kw)

def patch(url, **kw):
    return request("PATCH", url, **kw)

def delete(url, **kw):
    return request("DELETE", url, **kw)

修复后的调用代码

import ujson
import urequests

# 定义请求数据
data =  {
  "input_data": {
    "columns": [
      "PLAT_PH",
      "PLAT_ORP",
      "PSU1",
      "PSU2",
      "PSU3"
    ],
    "index": [0],
    "data": [[3.0, 600, 4.0, 5.0, 0]]
  }
}

url = 'your-azure-endpoint-url'
api_key = 'your-api-key'
headers = {
    'Content-Type':'application/json', 
    'Authorization':'Bearer ' + api_key, 
    'azureml-model-deployment': 'your-deployment-name'
}

try:
    # 使用json参数自动处理序列化，避免手动编码
    response = urequests.post(url, json=data, headers=headers, timeout=15)
    print(f"响应状态码: {response.status_code}")
    print(f"响应内容: {response.text}")
    # 解析JSON结果
    result = response.json()
    print(f"解析后的结果: {result}")
except OSError as e:
    print(f"请求出错: {e}")
except Exception as e:
    print(f"其他错误: {e}")

关键修复说明

禁用证书验证：在ussl.wrap_socket中添加cert_reqs=ussl.CERT_NONE，匹配原Python代码的证书绕过逻辑
升级HTTP协议：改用HTTP/1.1并添加Connection: close头部，适配Azure端点要求
类型兼容处理：自动将字符串类型的请求内容转换为字节，避免MicroPython类型错误
分块读取响应：分块读取响应内容，解决一次性读取超时或失败问题
添加超时与异常捕获：设置socket超时，添加完整异常捕获，方便定位问题

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

树莓派Pico W调用Azure ML REST端点推理无响应问题排查

问题背景 #

可正常运行的Python代码 #

尝试的MicroPython代码 #

使用的urequests库源码 #

问题分析与修复方案 #

核心问题点 #

修复后的urequests库代码 #

修复后的调用代码 #

关键修复说明 #