行空板M10扩展板——行空车（MCP）
本帖最后由 云天 于 2025-6-28 00:54 编辑【项目背景】
之前我尝试搭建了一个智能家居小助手项目，通过 ESP32-S3 摄像头安装小智 AI，利用电脑运行 MCP 服务，再借助行空板 M10 上的物联网平台 SIOT 与行空板 M10 进行通信，实现对 LED 与风扇开关的控制。之所以选择让电脑作为中间服务的枢纽，主要是因为在行空板 M10 上安装 MCP 库时遇到了困难。在“行空板 M10 扩展板组合活动群”里，我提出了行空板 M10 上无法安装 MCP 库的问题。幸运的是，群内的大神“Sky”及时为我提供了一个离线安装包，这让我看到了解决问题的希望。此外，我偶然看到了“豆爸”分享的关于 K10 小智与行空板 M10 MCP 服务器进行院线热映电影查询的项目文章。虽然文章中并未详细阐述如何安装 MCP 库，但在文章配图里，我注意到行空板 M10 中的文件夹里有 python3.12.7 离线安装包的存在。基于这些线索，我决定对整个智能家居小助手项目进行优化，尝试在行空板 M10 上直接安装 MCP 库，摆脱对电脑中间服务的依赖，使系统架构更加简洁高效，提升整个智能家居控制系统的稳定性和便捷性，让智能家居小助手的运行更加流畅自如，更好地服务于日常生活场景中的智能控制需求。
【项目准备】
解决在行空板M10上安装mcp库问题描述：行空板 M10 上的 Python 版本为 3.7，安装依赖库时找不到对应的库。需要升级 Python 到 3.12 版本。https://mc.dfrobot.com.cn/forum.php?mod=attachment&aid=MTk4NjgzfDliMzY1NDZmfDE3NTA5NTM1OTN8ODI3Nzg0fDM0MzA5OQ%3D%3D&noupdate=yes
解决方案：1.安装 Pyenv：# 检查权限
ls -l install.sh
# 添加权限
chmod +x install.sh
# 运行脚本
./install.sh
# 加载配置
source ~/.bashrc
# 验证 Pyenv 是否可用
pyenv --versionhttps://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/077706c81f7c28e1df048f7ae745fe12.png
2.使用 Pyenv 安装 Python 3.12：cd 2-installPythonIntoPyenv
# 运行脚本
./install.shhttps://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/06e9a53b3b83249a946d8e7142af7ac2.png3.全局切换pyenv global 3.12.7
设置全局默认的 Python 版本
检查 Python 版本是否已安装可以手动检查：ls ~/.pyenv/versions/
如果确实存在 3.12.7 目录，说明该版本已经安装。
https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/0562b541e8df351005dffe3f9bf60be0.png
4.在终端，使用pip install mcp，安装mcp,同时安装python-dotenv>=1.0.0，websockets>=11.0.3 ，pydantic>=2.11.4
https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/718e03b0dafd6064717b1321f8a927ff.png
配置小智AI
https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/b9af510cae982193691d573839ded304.png
获取MCP接入点
https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/4ef68dcd9eacdb05d9cee2c91ebd111f.png
【程序编写】
编写mcp服务文件，move.py，功能描述：控制行空车的前进、后退、左转、右转和停车。————————————————————————
# move.py
from mcp.server.fastmcp import FastMCP
import sys
import logging
logger = logging.getLogger('MoveServer')
# 修复Windows控制台UTF-8编码问题
if sys.platform == 'win32':
sys.stderr.reconfigure(encoding='utf-8')
sys.stdout.reconfigure(encoding='utf-8')
# 创建MCP服务器
mcp = FastMCP("MoveServer")
@mcp.tool()
def forward() -> dict:
"""
控制行空车前进。
该函数将设置P5和P6引脚为低电平，P8和P16引脚为50%占空比的PWM输出，使行空车前进。
Returns:
dict: 返回一个字典，包含操作结果。
"""
result = "行空车前进"
logger.info(result)
return {"success": True, "result": result}
@mcp.tool()
def back() -> dict:
"""
控制行空车后退。
该函数将设置P5和P6引脚为高电平，P8和P16引脚为50%占空比的PWM输出，使行空车后退。
Returns:
dict: 返回一个字典，包含操作结果。
"""
result = "行空车后退"
logger.info(result)
return {"success": True, "result": result}
@mcp.tool()
def left() -> dict:
"""
控制行空车左转。
该函数将设置P5引脚为高电平，P6引脚为低电平，P8和P16引脚为50%占空比的PWM输出，使行空车左转。
Returns:
dict: 返回一个字典，包含操作结果。
"""
result = "行空车左转"
logger.info(result)
return {"success": True, "result": result}
@mcp.tool()
def right() -> dict:
"""
控制行空车右转。
该函数将设置P5引脚为低电平，P6引脚为高电平，P8和P16引脚为50%占空比的PWM输出，使行空车右转。
Returns:
dict: 返回一个字典，包含操作结果。
"""
result = "行空车右转"
logger.info(result)
return {"success": True, "result": result}
@mcp.tool()
def stop() -> dict:
"""
控制行空车停止。
该函数将设置P8和P16引脚的PWM输出为0，使行空车停止。
Returns:
dict: 返回一个字典，包含操作结果。
"""
result = "行空车停车"
logger.info(result)
return {"success": True, "result": result}
# 启动服务器
if __name__ == "__main__":
mcp.run(transport="stdio")
——————————————————————————————————————————————————————————————————
修改小智AI管道文件mymcp.py，功能描述：连接到 MCP 服务器，并通过 WebSocket 端点将输入和输出与指定的 Python 脚本进行管道通信。
————————————————————————————————
"""
This script is used to connect to the MCP server and pipe the input and output to the websocket endpoint.
Version: 0.1.0
Usage:
export MCP_ENDPOINT=
python mcp_pipe.py
"""
import asyncio
import websockets
import subprocess
import logging
import os
import signal
import sys
import random
from dotenv import load_dotenv
from pinpong.board import Board,Pin
#from pinpong.extension.unihiker import *
import json
Board().begin()
p_p5_out=Pin(Pin.P5, Pin.OUT)
p_p8_pwm=Pin(Pin.P8, Pin.PWM)
p_p6_out=Pin(Pin.P6, Pin.OUT)
p_p16_pwm=Pin(Pin.P16, Pin.PWM)
# 设置日志记录器
# Load environment variables from .env file
#load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('MCP_PIPE')
# Reconnection settings
INITIAL_BACKOFF = 1# Initial wait time in seconds
MAX_BACKOFF = 600# Maximum wait time in seconds
reconnect_attempt = 0
backoff = INITIAL_BACKOFF
async def connect_with_retry(uri):
"""Connect to WebSocket server with retry mechanism"""
global reconnect_attempt, backoff
while True:# Infinite reconnection
try:
if reconnect_attempt > 0:
wait_time = backoff * (1 + random.random() * 0.1)# Add some random jitter
logger.info(f"Waiting {wait_time:.2f} seconds before reconnection attempt {reconnect_attempt}...")
await asyncio.sleep(wait_time)
# Attempt to connect
await connect_to_server(uri)
except Exception as e:
reconnect_attempt += 1
logger.warning(f"Connection closed (attempt: {reconnect_attempt}): {e}")
# Calculate wait time for next reconnection (exponential backoff)
backoff = min(backoff * 2, MAX_BACKOFF)
async def connect_to_server(uri):
"""Connect to WebSocket server and establish bidirectional communication with `mcp_script`"""
global reconnect_attempt, backoff
try:
logger.info(f"Connecting to WebSocket server...")
async with websockets.connect(uri) as websocket:
logger.info(f"Successfully connected to WebSocket server")
# Reset reconnection counter if connection closes normally
reconnect_attempt = 0
backoff = INITIAL_BACKOFF
# Start mcp_script process
process = subprocess.Popen(
['python', mcp_script],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding='utf-8',
text=True# Use text mode
)
logger.info(f"Started {mcp_script} process")
# Create two tasks: read from WebSocket and write to process, read from process and write to WebSocket
await asyncio.gather(
pipe_websocket_to_process(websocket, process),
pipe_process_to_websocket(process, websocket),
pipe_process_stderr_to_terminal(process)
)
except websockets.exceptions.ConnectionClosed as e:
logger.error(f"WebSocket connection closed: {e}")
raise# Re-throw exception to trigger reconnection
except Exception as e:
logger.error(f"Connection error: {e}")
raise# Re-throw exception
finally:
# Ensure the child process is properly terminated
if 'process' in locals():
logger.info(f"Terminating {mcp_script} process")
try:
process.terminate()
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
logger.info(f"{mcp_script} process terminated")
async def pipe_websocket_to_process(websocket, process):
"""Read data from WebSocket and write to process stdin"""
try:
while True:
# Read message from WebSocket
message = await websocket.recv()
logger.debug(f"<< {message[:120]}...")
# Write to process stdin (in text mode)
if isinstance(message, bytes):
message = message.decode('utf-8')
process.stdin.write(message + '\n')
process.stdin.flush()
except Exception as e:
logger.error(f"Error in WebSocket to process pipe: {e}")
raise# Re-throw exception to trigger reconnection
finally:
# Close process stdin
if not process.stdin.closed:
process.stdin.close()
async def pipe_process_to_websocket(process, websocket):
"""Read data from process stdout and send to WebSocket"""
global p_p5_out,p_p8_pwm,p_p6_out,p_p16_pwm
try:
while True:
# Read data from process stdout
data = await asyncio.get_event_loop().run_in_executor(
None, process.stdout.readline
)
if not data:# If no data, the process may have ended
logger.info("Process has ended output")
break
# Send data to WebSocket
logger.debug(f">> {data[:120]}...")
print(data)
# 解析 JSON 字符串
json_str = json.loads(data)
print(json_str['id'])
if json_str['id']>1:
print(json_str['id'])
if json_str.get('result', {}):
text=json.loads(json_str['result']['content']['text'])
if text['success']:
print(text['result'])
L=480
if "前进" in text['result']:
print("******************************")
p_p5_out.write_digital(0)
p_p8_pwm.write_analog(L)
p_p6_out.write_digital(0)
p_p16_pwm.write_analog(512)
if "后退" in text['result']:
print("******************************")
p_p5_out.write_digital(1)
p_p8_pwm.write_analog(L)
p_p6_out.write_digital(1)
p_p16_pwm.write_analog(512)
if "左转" in text['result']:
print("******************************")
p_p5_out.write_digital(1)
p_p8_pwm.write_analog(L)
p_p6_out.write_digital(0)
p_p16_pwm.write_analog(512)
if "右转" in text['result']:
print("******************************")
p_p5_out.write_digital(0)
p_p8_pwm.write_analog(L)
p_p6_out.write_digital(1)
p_p16_pwm.write_analog(512)
if "停车" in text['result']:
print("******************************")
p_p8_pwm.write_analog(0)
p_p16_pwm.write_analog(0)
# In text mode, data is already a string, no need to decode
await websocket.send(data)
except Exception as e:
logger.error(f"Error in process to WebSocket pipe: {e}")
raise# Re-throw exception to trigger reconnection
async def pipe_process_stderr_to_terminal(process):
"""Read data from process stderr and print to terminal"""
try:
while True:
# Read data from process stderr
data = await asyncio.get_event_loop().run_in_executor(
None, process.stderr.readline
)
if not data:# If no data, the process may have ended
logger.info("Process has ended stderr output")
break
# Print stderr data to terminal (in text mode, data is already a string)
sys.stderr.write(data)
sys.stderr.flush()
except Exception as e:
logger.error(f"Error in process stderr pipe: {e}")
raise# Re-throw exception to trigger reconnection
def signal_handler(sig, frame):
"""Handle interrupt signals"""
logger.info("Received interrupt signal, shutting down...")
sys.exit(0)
if __name__ == "__main__":
# Register signal handler
signal.signal(signal.SIGINT, signal_handler)
# mcp_script
#if len(sys.argv) < 2:
# logger.error("Usage: mcp_pipe.py ")
# sys.exit(1)
mcp_script = "move.py"
# Get token from environment variable or command line arguments
#endpoint_url = os.environ.get('MCP_ENDPOINT')
endpoint_url="wss://api.xiaozhi.me/mcp/?token=eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VySWQiOjEyODIzNSwiYWdlbnRJZCI6MTkwMjQsImVuZHBvaW50SWQiOiJhZ2VudF8xOTAyNCIsInB1cnBvc2UiOiJtY3AtZW5kcG9pbnQiLCJpYXQiOjE3NTA5MDkyNzV9.ms9egq4Y1huI4OsEg6p2OaNHLbXSfCbHXyqecQQK0HbWHxsh6nl88Ad8yjzGF7Znyn4uUDBC2JtMoSXi1PMOcg"
if not endpoint_url:
logger.error("Please set the `MCP_ENDPOINT` environment variable")
sys.exit(1)
# Start main loop
try:
asyncio.run(connect_with_retry(endpoint_url))
except KeyboardInterrupt:
logger.info("Program interrupted by user")
except Exception as e:
logger.error(f"Program execution error: {e}")
………………………………………………………………………………………………………………
【硬件安装】1.行空板K10（安装小智AI）https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/d63fa7f333710b7506a01949e9d0cfce.jpg2.行空板M10与扩展板结合乐高车
https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/93008b32d07f8e7dbcee7b9b96d734f1.jpghttps://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/cb635d750cd9a52a45c1eb159e77122a.jpg3.行空板K10与行空板M10https://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/8c2d88b906eb1415380f8ec87f3642d4.jpghttps://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/aab11adbd610ae317c190f8caa0a283b.jpghttps://makelogimg.dfrobot.com.cn/makelog/5ce9d7d0c25af2251a7d7e30/36fc12679e125f4d60bf0566e0104375.jpg
【演示视频】https://www.bilibili.com/video/BV1MjKkzPEqy/?share_source=copy_web M10使用mcp这么麻烦么？有没有方便的部署环境的方式？ dreamful 发表于 2025-7-1 12:06
M10使用mcp这么麻烦么？有没有方便的部署环境的方式？
切换Python看这个
https://gitee.com/liliang9693/unihiker-pyenv-python Forgotten 发表于 2025-7-1 14:42
切换Python看这个
https://gitee.com/liliang9693/unihiker-pyenv-python
好的，谢谢哈，我看看 Forgotten 发表于 2025-7-1 14:42
切换Python看这个
https://gitee.com/liliang9693/unihiker-pyenv-python
下载了离线安装包后，安装不上，不知道什么原因。
root@unihiker:~# cd 1-OfflineInstallPyenv
root@unihiker:~/1-OfflineInstallPyenv# ls -l install.sh
-rwxr--r-- 1 root root 2293 5月19 17:40 install.sh
root@unihiker:~/1-OfflineInstallPyenv# chmod +x install.sh
root@unihiker:~/1-OfflineInstallPyenv# ./install.sh
正在部署离线仓库...
检测到现有APT源配置，跳过添加
正在更新软件仓库...
获取:1 file:/var/local/offline-repo ./ InRelease
忽略:1 file:/var/local/offline-repo ./ InRelease
获取:2 file:/var/local/offline-repo ./ Release
忽略:2 file:/var/local/offline-repo ./ Release
获取:3 file:/var/local/offline-repo ./ Packages
忽略:3 file:/var/local/offline-repo ./ Packages
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
获取:3 file:/var/local/offline-repo ./ Packages
忽略:3 file:/var/local/offline-repo ./ Packages
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
获取:3 file:/var/local/offline-repo ./ Packages
忽略:3 file:/var/local/offline-repo ./ Packages
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
获取:3 file:/var/local/offline-repo ./ Packages
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
获取:4 file:/var/local/offline-repo ./ Translation-zh_CN
忽略:4 file:/var/local/offline-repo ./ Translation-zh_CN
获取:5 file:/var/local/offline-repo ./ Translation-en
忽略:5 file:/var/local/offline-repo ./ Translation-en
获取:6 file:/var/local/offline-repo ./ Translation-zh
忽略:6 file:/var/local/offline-repo ./ Translation-zh
命中:7 http://security.debian.org buster/updates InRelease
命中:8 http://httpredir.debian.org/debian buster InRelease
命中:9 http://httpredir.debian.org/debian buster-updates InRelease
忽略:10 http://httpredir.debian.org/debian buster-backports InRelease
错误:11 http://httpredir.debian.org/debian buster-backports Release
404Not Found
正在读取软件包列表... 完成
E: 仓库 “http://httpredir.debian.org/debian buster-backports Release” 不再含有 Release 文件。
N: 无法安全地用该源进行更新，所以默认禁用该源。
N: 参见 apt-secure(8) 手册以了解仓库创建和用户配置方面的细节。
root@unihiker:~/1-OfflineInstallPyenv# source ~/.bashrc
bash: pyenv：未找到命令
root@unihiker:~/1-OfflineInstallPyenv# pyenv --version
bash: pyenv：未找到命令
root@unihiker:~/1-OfflineInstallPyenv#
页:
[1]