ESP32开发板在micropython里直接用requests向web服务器发送请求：ESP32S3开发板通过fastapi中转成功连接星河大模型

在micropython里用requests连web服务器

本来想在ESP32开发板里直接连百度星河大模型，但是一直有报错，没调通，于是转而用fastapi进行中转，也就是先用ESP32连fastapi的中转服务器，该中转服务器再去连百度星河大模型。

WEB服务器是自己用fastapi启动的，地址：http://192.168.0.99:8000/login/

这台WE服务器测试了很久，其WEB服务代码也修改了多次，最终代码没有login部分，而是openai模型接入的链接："/v1/chat/completions/" ，中间测试代码就不提供了，最终WEB服务器代码见后面。

测试requests

这里一开始不懂，所以直接把字典当数据使用data参数传给web服务器，引起后面的报错：

import requests
data ={"model": "ernie-speed-8k","username": "testusername","password": "passtestusername"}
response = requests.post('http://192.168.0.99:8000/login/',headers = {'accept': 'application/json','Content-Type': 'application/x-www-form-urlencoded', },data = data
)
print(response.status_code)

Traceback (most recent call last):
File "<stdin>", line 5, in <module>
File "requests/__init__.py", line 205, in post
File "requests/__init__.py", line 144, in request
TypeError: object with buffer protocol required
有报错，看来直接用requests有点问题，需要再看看文档。

错误提示表明requests.post方法在尝试发送数据时遇到了问题。在标准的requests库中，这通常是因为传递给data或json参数的对象不是字符串、字节序列或类似的可缓冲对象。

因为看到报错里有“buffer”字样，刚开始还以为是板子出了问题，中间还换了开发板，从ESP32C3换成ESP32S3。现在回过头来看，应该是传输的数据有问题，data后面应该跟的是字符串（后来知道是application/x-www-form-urlencoded类型），json参数后面才带json数据。

看micropython的文档

里面讲了urequests.request

urequests.request(function, url, data=None, json=None, files=None, headers={}, auth=None)¶

向服务器发送 HTTP 请求。

function - 要使用的 HTTP 方法

url - 要发送的 URL

data - 要附加到请求主体的数据。如果提供了字典或元组列表，则将对其进行编码。

json - 用于附加到请求主体的 json 数据。

files - 用于文件上传，类型为 2 元组，定义了文件名、文件路径和内容类型。如下，{‘name’，（文件路径，内容类型）}

headers - 要发送的标头字典。

auth - 启用 Basic/Digest/自定义 HTTP Auth 的 Auth 元组。

学着用urequests.request(function, url, data=data）发送数据，结果还是报错：

用这段测试：

python">cat tp1.py 
import requests
data ={"model": "ernie-speed-8k","username": "testusername","password": "passtestusername"}
response = requests.post('http://192.168.0.99:8000/login/',headers = {'accept': 'application/json','Content-Type': 'application/x-www-form-urlencoded', },data = data
)
print(response.status_code)

服务器端用这个：

cat testpost.py 
from fastapi import FastAPI, Formapp = FastAPI()@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
#    print(username, model)print(username, )return {"username": username}
(py311) skywalk@rbpi:~/work/fastapi $ uvicorn testpost:app --reload --host 0.0.0.0
INFO:     Will watch for changes in these directories: ['/home/skywalk/work/fastapi']
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1617] using WatchFiles
INFO:     Started server process [1619]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

pc机测试通过：

python">python3 tp1.py 
200

micropython报buff有问题

再次测试micropython下request

python">import requests as urequests
# import urequestsdef sendGetRequest(url):try:response = urequests.get(url)print("Response Status Code:", response.status_code)print("Response Text:", response.text)response.close()except Exception as e:print("An error occurred:", e)# Main function
def main():ssid = 'your_wifi_name'password = 'your_wifi_password'# connectWiFi(ssid, password)url = 'http://192.168.0.99:8000/login/'sendGetRequest(url)if __name__ == "__main__" :main()

这次连上了：

main()
Response Status Code: 422
Response Text: {"detail":[{"type":"missing","loc":["body","username"],"msg":"Field required"
,"input":null},{"type":"missing","loc":["body","password"],"msg":"Field required","input":nu
ll}]}

尽管有报错422,但证明报文送到了服务器，服务器返回了信息。

尝试解决422报错问题

查找422故障原因：422状态码是属于客户端错误的一种,表示服务器能够理解请求,但是请求格式正确,却无法处理

最终经过艰苦卓绝的战斗，成功解决了422报错问题，原来是json格式不对，需要把数据用json格式化好才行，解决问题过程见：走进科学json版：在 JSON 格式中，字符串值必须使用双引号 “ 来界定，而不能使用单引号 ‘-CSDN博客

最终测试成功！下面为成功后的代码和配置：

在树莓派里启动fastapi服务

启动服务

服务器app.py文件放置在~/work/fastapiagent目录，进入该目录，然后使用uvicorn启动服务：

uvicorn app:app --host 0.0.0.0 --reload

服务器代码，app.py文件内容：

python">cat app.py 
from typing import Annotated
import json
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel
from openai import OpenAIimport os
import time
from typing import Union, Optionalapp = FastAPI()
API_KEY = "xxxx"
model="ernie-speed-8k"class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass ChatItem(BaseModel):messages: listmodel: str | None = Nonedef ChatCompletions(messages: list,model: str,# provider: Optional[ProviderType] = None,stream: Optional[bool] = False,# proxy: Optional[str] = None,response_format: Optional[dict] = None,max_tokens: Optional[int] = None,stop: Optional[Union[list[str], str]] = None,api_key: Optional[str] = None,# ignored: Optional[list[str]] = None,# ignore_working: Optional[bool] = False,ignore_stream: Optional[bool] = False,**kwargs):client = OpenAI(api_key=api_key,  # 含有 AI Studio 访问令牌的环境变量，https://aistudio.baidu.com/account/accessToken,base_url="https://aistudio.baidu.com/llm/lmapi/v3",  # aistudio 大模型 api 服务域名
)chat_completion = client.chat.completions.create(messages=messages,model=model,
)print(f"==log for app chat_completion:{chat_completion} ")# response = chat_completion.result# print(f"==log for app  response:{response}")return chat_completion@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(title="The ID of the item to get")],q: Annotated[str | None, Query(alias="item-query")] = None,
):results = {"item_id": item_id}if q:results.update({"q": q})return results@app.post("/v1/chat/completions/")
async def chat_completions(chatitem: ChatItem):print(chatitem)chatitemdict = chatitem.dict()print(f"==== items:{chatitemdict}")#print ("=" * 20 , messages, type(messages))#yjson = json.loads(messages)#print("="*10, yjson, type(yjson))#zmessages = yjson["messages"]#print("="*10, zmessages, typr(zmessages))model="ernie-speed-8k"messages = chatitem.messagesprint(f"==== messages=chatitem.msg:{messages}")y = ChatCompletions(messages=messages, model=model, api_key=API_KEY)print("="*10, y)z = y.choices[0].message.contentreturn z@app.post("/items/")
async def create_item(item: Item):print(f"==== get the Item type:{type(item)}, item:{item}")item_dict = item.dict()print(f"==== item_dict{item_dict}")if item.tax:price_with_tax = item.price + item.taxitem_dict.update({"price_with_tax": price_with_tax})return item_dict

服务器启动后，服务器侦听192.168.0.99地址的8000端口。

客户端curl连通测试

curl测试代码：

python">curl -X 'POST' \'http://192.168.0.99:8000/v1/chat/completions/' \-H 'Content-Type: application/json' \-d '{"messages":[{"role": "user", "content": "hello"}]}'

测试通过！

ESP32S3开发板通过中转连星河大模型成功

在ESP32S3开发板的MicroPython环境下，使用requests来连fastapi的中转服务器，连接成功：

python">import requestsresponse = requests.post('http://192.168.0.99:8000/v1/chat/completions/',headers = {'Content-Type': 'application/json',},json = {'model': "ernie-speed-8k","messages": [{"role": "user", "content": "hello"}]}
)print(response.status_code, response.reason)
print(response.text)

python">print(response.text)
"你好！有什么我可以帮助你的吗？"

自此，ESP32开发板在micropython里直接用requests向web服务器发送请求，测试成功！