回到首页 返回首页
回到顶部 回到顶部
返回上一页 返回上一页
best-icon

行空板M10扩展板 ——小智AI车(MCP) 简单

头像 云天 2025.06.27 15 0

【项目背景】
 

  • 之前我尝试搭建了一个智能家居小助手项目,通过 ESP32-S3 摄像头安装小智 AI,利用电脑运行 MCP 服务,再借助行空板 M10 上的物联网平台 SIOT 与行空板 M10 进行通信,实现对 LED 与风扇开关的控制。之所以选择让电脑作为中间服务的枢纽,主要是因为在行空板 M10 上安装 MCP 库时遇到了困难。在“行空板 M10 扩展板组合活动群”里,我提出了行空板 M10 上无法安装 MCP 库的问题。幸运的是,群内的大神“Sky”及时为我提供了一个离线安装包,这让我看到了解决问题的希望。此外,我偶然看到了“豆爸”分享的关于 K10 小智与行空板 M10 MCP 服务器进行院线热映电影查询的项目文章。虽然文章中并未详细阐述如何安装 MCP 库,但在文章配图里,我注意到行空板 M10 中的文件夹里有 python3.12.7 离线安装包的存在。基于这些线索,我决定对整个智能家居小助手项目进行优化,尝试在行空板 M10 上直接安装 MCP 库,摆脱对电脑中间服务的依赖,使系统架构更加简洁高效,提升整个智能家居控制系统的稳定性和便捷性,让智能家居小助手的运行更加流畅自如,更好地服务于日常生活场景中的智能控制需求。
     

image.png


【项目准备】
解决在行空板M10上安装mcp库

问题描述:行空板 M10 上的 Python 版本为 3.7,安装依赖库时找不到对应的库。需要升级 Python 到 3.12 版本。
MonJune-202506233704..png
解决方案

1.安装 Pyenv:

——————————————————————————————————————————————————————

# 检查权限
ls -l install.sh
# 添加权限
chmod +x install.sh
# 运行脚本
./install.sh
# 加载配置
source ~/.bashrc
# 验证 Pyenv 是否可用
pyenv --version

——————————————————————————————————————————————————————
image.png

2.使用 Pyenv 安装 Python 3.12:

cd 2-installPythonIntoPyenv

# 运行脚本
./install.sh
image.png
(4)全局切换pyenv global 3.12.7
设置全局默认的 Python 版本

检查 Python 版本是否已安装可以手动检查:ls ~/.pyenv/versions/
如果确实存在 3.12.7 目录,说明该版本已经安装。
image.png

(5)在终端,使用pip install mcp,安装mcp,同时安装python-dotenv>=1.0.0,websockets>=11.0.3 ,pydantic>=2.11.4

image.png
配置小智AI
 image.png

获取MCP接入点

image.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'][0]['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)

069b42310a24af4ac9da9198a2ff3e3.jpg

2.行空板M10与扩展板结合乐高车

cd2332e0fab4d6dec9982e4c1bb79e5.jpg
d9b417645ed2fd99067b1d5b89c0271.jpg

3.行空板K10与行空板M10

b708a36ee796e81c786f29e1a9f16b3.jpg
e1e45478bf5e524b25b2cae35a35a19.jpg
6b397b1bee8f13b9786e9d2828d4c72.jpg
58f7783bdccc7ceb4000b4e6f88b268.jpg

 

【演示视频】

评论

user-avatar