px4-mcp-server by qkilani - MCP Server

PX4 MCP ROS2 Server

Production-grade Model Context Protocol (MCP) server enabling LLMs to safely control PX4 SITL quadcopters through standardized drone operation commands.

🚀 Quick Start

Prerequisites

Python 3.11+
ROS 2 Jazzy (optional, for real PX4 integration)
PX4 SITL (optional, for real drone simulation)

Installation

# Clone the repository
git clone <repository-url>
cd px4-mcp-ros2

# Install the package
pip install -e .

# Or install dependencies manually
pip install mcp pydantic pydantic-settings structlog anyio

🏃 Running the Server

1. Set Required Environment Variables

# Required: Safety token for authorization
export PX4_MCP_SAFETY_TOKEN="<your-token-here>"

2. Start the MCP Server

Option A: Using Python module (recommended)

python -m px4_mcp_ros2.mcp_server

Option B: Using the installed script

px4-mcp-server

Option C: Direct Python execution

python px4_mcp_ros2/mcp_server.py

3. Development Mode

For testing and development with enhanced logging:

export PX4_MCP_SAFETY_TOKEN="<your-token-here>"
export PX4_MCP_DRY_RUN_DEFAULT="true"
export PX4_MCP_LOG_LEVEL="DEBUG"
python -m px4_mcp_ros2.mcp_server

⚙️ Configuration

Configure the server using environment variables:

Required Configuration

export PX4_MCP_SAFETY_TOKEN="your-secure-token"  # Min 8 characters, no common passwords

Optional Configuration

# ROS 2 Settings
export PX4_MCP_ROS_DOMAIN_ID="0"              # ROS2 domain ID (0-101)
export PX4_MCP_FRAME_ID="1"                   # PX4 system/frame ID (1-255)

# Timeouts
export PX4_MCP_TIMEOUT_SEC="3.0"              # Command timeout (0.1-30.0s)
export PX4_MCP_ACK_TIMEOUT_SEC="2.0"          # PX4 acknowledgment timeout

# Logging
export PX4_MCP_LOG_LEVEL="INFO"               # DEBUG, INFO, WARNING, ERROR
export PX4_MCP_LOG_STRUCTURED="true"          # Enable JSON structured logging

# Development
export PX4_MCP_DRY_RUN_DEFAULT="false"        # Default dry-run mode
export PX4_MCP_MAX_RETRY_ATTEMPTS="3"         # Max command retry attempts (1-10)

# Health Monitoring
export PX4_MCP_STALE_MESSAGE_THRESHOLD_MS="1000"    # Message staleness threshold
export PX4_MCP_CRITICAL_MESSAGE_AGE_MS="5000"       # Critical message age warning

🛠️ Available MCP Tools

The server exposes the following tools for LLM integration:

Flight Control Tools

arm - Arm the quadcopter motors
disarm - Disarm the quadcopter motors
takeoff - Command takeoff to specified altitude
land - Command landing at current position
return_to_base - Return to launch/home position

Monitoring Tools

health_check - Get comprehensive system status

Tool Input Schema

All tools accept:

{
  "safety_token": "string",     // Required: Your safety authorization token
  "dry_run": false             // Optional: Test mode without actual execution
}

For takeoff, additional parameters are supported:

{
  "safety_token": "your-token",
  "parameters": {
    "altitude_amsl": 10.0,     // Required: Altitude in meters (1.0-100.0)
    "yaw_deg": 0.0             // Optional: Target yaw in degrees
  },
  "dry_run": false
}

🧪 Testing

Run the Test Suite

# Run all tests
python -m pytest tests/ -v

# Run unit tests only
python -m pytest tests/unit/ -v

# Run with coverage
python -m pytest tests/ --cov=px4_mcp_ros2 --cov-report=html

Test Individual Tools

# Test basic server functionality
PX4_MCP_SAFETY_TOKEN="<your-token-here>" python -c "
from px4_mcp_ros2.mcp_server import PX4MCPServer
server = PX4MCPServer()
print(f'Server initialized with {len(server.tools)} tools')
for name in server.tools:
    print(f'  - {name}')
"

Dry Run Testing

# Test arm command in dry-run mode
PX4_MCP_SAFETY_TOKEN="<your-token-here>" python -c "
from px4_mcp_ros2.tools.arm import ArmTool
import asyncio

async def test():
    tool = ArmTool()
    result = await tool.execute({
        'safety_token': '<your-token-here>',
        'dry_run': True
    })
    print('Arm result:', result)

asyncio.run(test())
"

🔌 MCP Client Integration

Claude Desktop Integration

Add to your mcp.json configuration:

{
  "servers": {
    "px4-mcp-ros2": {
      "command": "python",
      "args": ["-m", "px4_mcp_ros2.mcp_server"],
      "env": {
        "PX4_MCP_SAFETY_TOKEN": "your-secure-token-here",
        "PX4_MCP_LOG_LEVEL": "INFO"
      }
    }
  }
}

Generic MCP Client

The server communicates via stdio using the MCP protocol:

# Start server and pipe to MCP client
python -m px4_mcp_ros2.mcp_server | your-mcp-client

🏗️ Architecture

Project Structure

px4_mcp_ros2/
├── tools/           # MCP tool implementations
│   ├── arm.py
│   ├── disarm.py
│   ├── takeoff.py
│   ├── land.py
│   ├── return_to_base.py
│   └── health_check.py
├── models/          # Data models and schemas
│   ├── schemas.py   # Pydantic models for validation
│   └── errors.py    # Error handling classes
├── ros/             # ROS 2 integration layer
│   ├── node.py      # PX4 ROS 2 node
│   ├── mav_cmd.py   # MAVLink command definitions
│   └── qos.py       # ROS 2 QoS profiles
├── mcp_server.py    # Main MCP server implementation
└── settings.py      # Configuration management

Safety Features

Token-based Authorization - All operations require valid safety token
Input Validation - Pydantic schemas validate all inputs
Dry-run Mode - Test commands without actual execution
Structured Logging - Full audit trail with request correlation
Error Classification - Standardized error codes and messages

ROS 2 Integration

The server integrates with PX4 via ROS 2:

VehicleCommand publisher for sending commands
VehicleCommandAck subscriber for command acknowledgments
VehicleStatus subscriber for system state monitoring
Configurable QoS profiles for reliable communication

🚁 PX4 SITL Integration

For real PX4 simulation, run alongside PX4 SITL:

# Terminal 1: Start PX4 SITL
cd /path/to/PX4-Autopilot
make px4_sitl gazebo-classic

# Terminal 2: Start MicroXRCE Agent (if using DDS)
MicroXRCEAgent udp4 -p 8888

# Terminal 3: Start MCP Server
export PX4_MCP_SAFETY_TOKEN="your-token"
export PX4_MCP_ROS_DOMAIN_ID="0"
python -m px4_mcp_ros2.mcp_server

🐛 Troubleshooting

Common Issues

"Field required: safety_token"

# Ensure safety token is set
export PX4_MCP_SAFETY_TOKEN="<your-token-here>"

"Invalid safety token provided"

Token must be at least 8 characters
Cannot be common passwords (password, 123456, test, admin)
Must match PX4_MCP_SAFETY_TOKEN environment variable

ROS 2 Connection Issues

# Check ROS 2 domain
export PX4_MCP_ROS_DOMAIN_ID="0"
ros2 topic list  # Should show PX4 topics if connected

Import Errors

# Reinstall dependencies
pip install -e .
# Or install individual packages
pip install mcp pydantic pydantic-settings structlog

Debug Mode

Enable verbose logging for troubleshooting:

export PX4_MCP_LOG_LEVEL="DEBUG"
export PX4_MCP_LOG_STRUCTURED="true"
python -m px4_mcp_ros2.mcp_server

📖 Development

Code Quality

# Linting
ruff check px4_mcp_ros2/

# Type checking
mypy px4_mcp_ros2/

# Formatting
black px4_mcp_ros2/

Adding New Tools

Create tool class in px4_mcp_ros2/tools/
Implement execute() method with safety validation
Register in PX4MCPServer._register_tools()
Add contract tests in tests/unit/

📄 License

MIT License - see LICENSE file for details.

🤝 Contributing

Fork the repository
Create feature branch (git checkout -b feature/amazing-feature)
Commit changes (git commit -m 'Add amazing feature')
Push to branch (git push origin feature/amazing-feature)
Open Pull Request

📞 Support

For issues and questions:

Create GitHub issue with detailed description
Include logs with PX4_MCP_LOG_LEVEL="DEBUG"
Specify environment (Python version, ROS 2 distro, PX4 version)