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）发送数据，结果还是报错：

用这段测试：

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机测试通过：

python3 tp1.py 
200

micropython报buff有问题

再次测试micropython下request

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文件内容：

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测试代码：

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的中转服务器，连接成功：

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)

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

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