微空 AI 智能无人机教程 5 - AI agent 智能体接入

自然语言指令（Telegram 消息）
     ↓
  OpenClaw 智能体（Node.js）
     ├── 云端 Gemini API（理解 + 决策）
     ├── Skill（SKILL.md，描述可用能力）
     └── MCP 协议（stdio / JSON-RPC）
          ↓
  drone_mcp_server.py（Python MCP Server）
     └── rclpy → MAVROS → 飞控

curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.40.1/install.sh | bash

source ~/.bashrc

nvm --version

nvm install 22

node -v
npm -v

curl -fsSL https://openclaw.ai/install.sh | bash

openclaw --version

openclaw onboard

connect telegram

openclaw gateway status --deep

openclaw gateway start      # 启动服务
openclaw gateway stop       # 停止服务
openclaw gateway restart    # 重启服务
openclaw logs --follow      # 查看实时日志

你好，你能做什么？

请告诉我这台机器的 hostname 和系统信息（uname -a）

htop

sudo apt install -y python3-venv
python3 -m venv --system-site-packages ~/openclaw_drone/venv
source ~/openclaw_drone/venv/bin/activate
pip install mcp

python3 -c "import mcp; print('MCP SDK 安装成功')"
python3 -c "import rclpy; print('rclpy 可用')"

sudo vim ~/openclaw_drone/drone_mcp_server.py

#!/usr/bin/env python3
"""
MVD35 Drone MCP Server
通过 MCP 协议将无人机状态查询接口暴露给 OpenClaw AI 智能体。
通信方式：stdio (JSON-RPC)

注意：MCP Server 通过 stdio 与 OpenClaw 通信，
      绝对不能在代码中使用 print()，否则会破坏 JSON-RPC 数据流。
      调试信息请输出到 sys.stderr。
"""

import sys
import threading
import math
import rclpy
from rclpy.node import Node
from mcp.server.fastmcp import FastMCP
from geometry_msgs.msg import PoseStamped
from mavros_msgs.msg import State
from sensor_msgs.msg import BatteryState


# ============ ROS 2 节点 ============

class DroneROSNode(Node):
    """持久运行的 ROS 2 节点，订阅无人机状态数据"""

    def __init__(self):
        super().__init__('drone_mcp_bridge')

        # 状态数据（持续更新）
        self.state = State()
        self.pose = PoseStamped()
        self.battery = BatteryState()

        # 连接状态标记
        self.state_received = False
        self.pose_received = False
        self.battery_received = False

        # 订阅 MAVROS 话题
        self.state_sub = self.create_subscription(
            State, '/mavros/state', self._state_cb, 10)
        self.pose_sub = self.create_subscription(
            PoseStamped, '/mavros/local_position/pose', self._pose_cb, 10)
        self.battery_sub = self.create_subscription(
            BatteryState, '/mavros/battery', self._battery_cb, 10)

        self.get_logger().info('DroneROSNode 初始化完成，等待 MAVROS 数据...')

    def _state_cb(self, msg):
        self.state = msg
        self.state_received = True

    def _pose_cb(self, msg):
        self.pose = msg
        self.pose_received = True

    def _battery_cb(self, msg):
        self.battery = msg
        self.battery_received = True

    def quaternion_to_yaw(self, q):
        """将四元数转换为偏航角（度）"""
        siny_cosp = 2.0 * (q.w * q.z + q.x * q.y)
        cosy_cosp = 1.0 - 2.0 * (q.y * q.y + q.z * q.z)
        yaw_rad = math.atan2(siny_cosp, cosy_cosp)
        return math.degrees(yaw_rad)


# ============ MCP Server ============

mcp = FastMCP("MVD35-DroneControl")
drone_node = None  # 全局 ROS 节点引用


@mcp.tool()
def get_drone_status() -> str:
    """获取无人机的综合状态信息，包括连接状态、飞行模式、解锁状态、位置和电池。
    这是了解无人机当前情况的最常用工具。"""
    if drone_node is None:
        return "错误：ROS 2 节点尚未初始化"

    lines = []

    # 飞控连接状态
    if drone_node.state_received:
        s = drone_node.state
        lines.append(f"飞控连接: {'已连接' if s.connected else '未连接'}")
        lines.append(f"飞行模式: {s.mode}")
        lines.append(f"解锁状态: {'已解锁（电机可运转）' if s.armed else '未解锁（安全状态）'}")
    else:
        lines.append("飞控连接: 未收到状态数据（MAVROS 可能未启动）")

    # 位置
    if drone_node.pose_received:
        p = drone_node.pose.pose.position
        yaw = drone_node.quaternion_to_yaw(drone_node.pose.pose.orientation)
        lines.append(f"位置: x={p.x:.2f}m, y={p.y:.2f}m, z={p.z:.2f}m")
        lines.append(f"朝向: {yaw:.1f} 度")
    else:
        lines.append("位置: 暂无数据")

    # 电池
    if drone_node.battery_received:
        b = drone_node.battery
        pct = b.percentage * 100 if b.percentage >= 0 else -1
        lines.append(f"电池: {pct:.0f}% ({b.voltage:.1f}V)")
    else:
        lines.append("电池: 暂无数据")

    return "\n".join(lines)


@mcp.tool()
def get_position() -> str:
    """获取无人机当前的本地坐标位置（米）。
    坐标系：x 为前方，y 为左方，z 为高度。"""
    if drone_node is None or not drone_node.pose_received:
        return "暂无位置数据"
    p = drone_node.pose.pose.position
    yaw = drone_node.quaternion_to_yaw(drone_node.pose.pose.orientation)
    return f"位置: x={p.x:.3f}m, y={p.y:.3f}m, z={p.z:.3f}m, 朝向={yaw:.1f} 度"


@mcp.tool()
def get_battery() -> str:
    """获取无人机的电池状态，包括电压和剩余电量百分比。"""
    if drone_node is None or not drone_node.battery_received:
        return "暂无电池数据"
    b = drone_node.battery
    pct = b.percentage * 100 if b.percentage >= 0 else -1
    return f"电压: {b.voltage:.2f}V, 剩余电量: {pct:.0f}%"


@mcp.tool()
def is_armed() -> str:
    """检查无人机电机是否已解锁。返回解锁状态说明。"""
    if drone_node is None or not drone_node.state_received:
        return "暂无状态数据"
    if drone_node.state.armed:
        return "已解锁：电机可以运转，无人机可能正在飞行中"
    else:
        return "未解锁：电机锁定，无人机处于安全状态"


@mcp.tool()
def get_flight_mode() -> str:
    """获取无人机当前的飞行模式。
    常见模式：STABILIZE（自稳）、ALT_HOLD（定高）、LOITER（定点）、
    GUIDED（引导）、LAND（降落）。"""
    if drone_node is None or not drone_node.state_received:
        return "暂无状态数据"
    return f"当前飞行模式: {drone_node.state.mode}"


@mcp.tool()
def check_connection() -> str:
    """检查各个系统组件的连接状态，包括飞控连接、位置数据和电池数据。"""
    if drone_node is None:
        return "错误：ROS 2 节点尚未初始化"

    lines = []
    lines.append(f"MAVROS 状态话题: {'已连接' if drone_node.state_received else '未收到数据'}")
    lines.append(f"位置数据: {'正常' if drone_node.pose_received else '未收到数据'}")
    lines.append(f"电池数据: {'正常' if drone_node.battery_received else '未收到数据'}")

    if drone_node.state_received:
        lines.append(f"飞控连接: {'已连接' if drone_node.state.connected else '未连接'}")

    return "\n".join(lines)


# ============ 启动逻辑 ============

def run_ros_spin(node):
    """在单独线程中运行 ROS 2 事件循环"""
    rclpy.spin(node)


def main():
    global drone_node

    # 初始化 ROS 2
    rclpy.init()
    drone_node = DroneROSNode()

    # 在后台线程运行 ROS 2 spin（持续接收话题数据）
    ros_thread = threading.Thread(
        target=run_ros_spin, args=(drone_node,), daemon=True)
    ros_thread.start()

    # 调试信息输出到 stderr（不能用 print！）
    sys.stderr.write("[drone_mcp_server] MCP Server 启动中...\n")
    sys.stderr.flush()

    # 启动 MCP Server（阻塞主线程，通过 stdio 与 OpenClaw 通信）
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

sudo vim ~/openclaw_drone/start_mcp_server.sh

#!/bin/bash
# MCP Server 启动脚本，加载 ROS 2 环境和虚拟环境后运行 drone_mcp_server.py
source /opt/ros/jazzy/setup.bash
source ~/ros2_ws/install/setup.bash 2>/dev/null
source ~/openclaw_drone/venv/bin/activate
exec python3 ~/openclaw_drone/drone_mcp_server.py

sudo chmod +x ~/openclaw_drone/drone_mcp_server.py
sudo chmod +x ~/openclaw_drone/start_mcp_server.sh

openclaw mcp add drone-control --command bash --arg /home/micoair/openclaw_drone/start_mcp_server.sh

openclaw mcp list

openclaw gateway restart
openclaw gateway status

mkdir -p ~/.openclaw/skills/mvd35-drone
vim ~/.openclaw/skills/mvd35-drone/SKILL.md

---
name: mvd35-drone
description: 查询和监控 MVD35 智能无人机的飞行状态。当用户询问无人机状态、电池、位置、飞行模式等信息时使用此技能。
---

# MVD35 无人机状态查询

你是一架 MVD35 微型智能无人机上的 AI 助手。你可以通过 MCP 工具 `drone-control` 来查询无人机的实时状态。

## 可用查询工具

- `get_drone_status`：获取无人机综合状态（飞行模式、解锁状态、位置、电池），这是最常用的工具
- `get_position`：获取无人机当前坐标位置
- `get_battery`：获取电池电压和剩余电量
- `is_armed`：检查电机是否已解锁
- `get_flight_mode`：获取当前飞行模式
- `check_connection`：检查各系统组件连接状态

## 使用规则

1. 当用户询问无人机的总体情况时，优先使用 `get_drone_status` 获取综合信息
2. 当用户只关心某个具体指标时，使用对应的专用工具
3. 如果工具返回"暂无数据"，告诉用户可能是 MAVROS 尚未启动或飞控未连接
4. 用简洁友好的中文回复用户

openclaw gateway restart

openclaw skills list

ros2 launch mavros apm.launch

openclaw gateway status

openclaw gateway start

你叫mvd35-001，负责本台mvd35智能无人机的管理与控制。

你好，你能做什么？

无人机现在是什么状态？

检查一下各个系统的连接状态