Tools layer: 16 MCP tools over the three streams

connect/disconnect/status, list_commands/command_info, the generic
send_command, get_state/watch_state/preflight_check, camera snapshot and
record, FTP listing and fetch, the read-only shell, and arm/disarm.

The safety gate is tested exhaustively rather than by sample: every motion
and envelope command is asserted to refuse while unarmed, and Landing and
Emergency are asserted to work without arming, because refusing to bring
down an airborne aircraft is the more dangerous answer.

preflight_check reads the per-element sensor keys, so a single failing
self-test is visible instead of being hidden by whichever arrived last.
This commit is contained in:
2026-10-02 01:34:55 -06:00
parent 6f2c36936e
commit 607d14979d
13 changed files with 1053 additions and 10 deletions
+94
View File
@@ -0,0 +1,94 @@
"""Result models.
Pydantic for outputs so the schema reaches the caller; `Annotated`/`Field` is
used for inputs. These are deliberately plain: a tool result is read by a model
that has not seen the drone, so every field says what it is in its description.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
class ConnectionInfo(BaseModel):
connected: bool
target: str = Field(description="Address of the drone, or 'sim' for the built-in simulator.")
armed: bool = Field(description="Whether commands that can move the aircraft are currently unlocked.")
armed_until: str | None = Field(default=None, description="When the unlock expires, ISO 8601.")
telemetry_keys: int = Field(default=0, description="How many distinct values the drone has reported.")
last_update_age: float | None = Field(
default=None, description="Seconds since the most recent telemetry. Large means the link is stale."
)
undecodable_events: int = Field(
default=0, description="Events whose command id is absent from Parrot's XML. Normally zero."
)
class CommandSummary(BaseModel):
name: str
title: str = ""
tier: str = Field(description="observe, config, envelope or motion. motion and envelope need arm().")
direction: str = Field(description="to_drone if you can send it, from_drone if the drone emits it.")
bebop2: bool | None = Field(
description="Whether the XML says the Bebop 2 supports it. None means unstated."
)
deprecated: bool = False
class ArgumentInfo(BaseModel):
name: str
type: str
doc: str = ""
enum_values: list[str] = Field(default_factory=list, description="Allowed names, for an enum argument.")
class CommandDetail(CommandSummary):
ids: list[int] = Field(description="project, class and command id, as sent on the wire.")
args: list[ArgumentInfo] = Field(default_factory=list)
doc: str = ""
buffer: str = Field(description="Which link buffer carries it: ack, non_ack or high_prio.")
confirmed_by: list[str] = Field(
default_factory=list, description="Events the drone emits to confirm it, if the XML declares any."
)
class SendResult(BaseModel):
command: str
args: dict = Field(default_factory=dict)
acked: bool = Field(description="Whether the drone acknowledged receipt.")
confirmed: bool = Field(default=False, description="Whether the drone also emitted its confirming event.")
elapsed_ms: float = 0.0
note: str = ""
class StateValue(BaseModel):
value: object
age: float = Field(description="Seconds since the drone last reported this. Old values may be stale.")
class Check(BaseModel):
name: str
ok: bool | None = Field(description="True, False, or None when the drone has not reported it.")
detail: str
class Preflight(BaseModel):
ready: bool = Field(
description="False if any blocking check failed. Advisory, not an airworthiness test."
)
blocking: list[str] = Field(default_factory=list, description="Checks that would prevent a safe flight.")
advisory: list[str] = Field(default_factory=list, description="Worth knowing, but not disqualifying.")
checks: list[Check] = Field(default_factory=list)
class FileEntry(BaseModel):
name: str
size: int | None = None
is_dir: bool = False
class ArmState(BaseModel):
armed: bool
reason: str = ""
expires_at: str | None = None
note: str = ""
+18 -3
View File
@@ -8,6 +8,8 @@ through pyparrot, which prints from its receive thread.
import logging
import os
import sys
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastmcp import FastMCP
@@ -33,12 +35,25 @@ preflight_check() is the quickest way to learn whether the aircraft is healthy.
def build_server(settings: Settings | None = None) -> FastMCP:
"""Build the server. Tests call this directly so they can inject settings."""
settings = settings or Settings()
mcp = FastMCP("mcbebop", instructions=INSTRUCTIONS, version=__version__)
mcp.settings_obj = settings # type: ignore[attr-defined]
from mcbebop.tools import register_all
register_all(mcp, settings)
state_holder: dict = {}
@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[None]:
try:
yield
finally:
# A session outlives any single tool call, so closing it is the
# server's job. Without this a stopped server can leave the drone
# streaming video and holding the one controller slot it has.
app_state = state_holder.get("app")
if app_state is not None:
await app_state.close()
mcp = FastMCP("mcbebop", instructions=INSTRUCTIONS, version=__version__, lifespan=lifespan)
state_holder["app"] = register_all(mcp, settings)
return mcp
+73
View File
@@ -0,0 +1,73 @@
"""Server-wide state: the drone session and the arming lock.
Kept out of `tools/` deliberately. This module may use postponed annotations;
modules that define MCP tools may not, because FastMCP builds their schemas
from real annotation objects.
"""
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field
from mcbebop.config import Settings
SIM_TARGET = "sim"
@dataclass
class Arming:
"""Whether commands that can move the aircraft are unlocked.
Expiring is the point. A session that armed an hour ago to nudge the camera
should not still be able to spin motors, so this is checked on every send
rather than only at the moment of arming.
"""
reason: str = ""
expires_at: float = 0.0
@property
def armed(self) -> bool:
return self.expires_at > time.monotonic()
def arm(self, reason: str, minutes: int) -> None:
self.reason = reason
self.expires_at = time.monotonic() + minutes * 60
def disarm(self) -> None:
self.reason = ""
self.expires_at = 0.0
@property
def seconds_left(self) -> float:
return max(0.0, self.expires_at - time.monotonic())
@dataclass
class AppState:
settings: Settings
session: object | None = None # DroneSession once connected
sim: object | None = None # FakeBebop when target is "sim"
target: str = ""
arming: Arming = field(default_factory=Arming)
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
@property
def connected(self) -> bool:
return self.session is not None and getattr(self.session, "connected", False)
async def close(self) -> None:
if self.session is not None:
try:
await self.session.disconnect()
finally:
self.session = None
if self.sim is not None:
try:
self.sim.__exit__(None, None, None)
finally:
self.sim = None
self.target = ""
self.arming.disarm()
+10 -7
View File
@@ -1,17 +1,20 @@
"""Tool registration.
Each module exposes `register(mcp, settings)`; this is the only place that
knows the full set, so tool naming stays consistent.
The one place that knows the full set, so naming stays consistent across the
modules that would otherwise each invent their own.
"""
from fastmcp import FastMCP
from mcbebop.config import Settings
from mcbebop.state import AppState
from mcbebop.tools import camera, command, connection, files, protocol, safety, state
from mcbebop.tools._common import set_state
def register_all(mcp: FastMCP, settings: Settings) -> None:
# Modules are added here as each stream lands. Stage 0 registers nothing,
# so the server starts and answers list_tools with an empty set.
modules: list = []
for module in modules:
def register_all(mcp: FastMCP, settings: Settings) -> AppState:
app_state = AppState(settings=settings)
set_state(app_state)
for module in (connection, protocol, command, state, camera, files, safety):
module.register(mcp, settings)
return app_state
+34
View File
@@ -0,0 +1,34 @@
"""Helpers every tool module needs."""
from datetime import UTC, datetime, timedelta
from mcbebop import errors
from mcbebop.state import AppState
_STATE: AppState | None = None
def set_state(state: AppState) -> None:
"""Install the shared state. Called once by `register_all`."""
global _STATE
_STATE = state
def app() -> AppState:
if _STATE is None: # pragma: no cover - register_all always runs first
raise RuntimeError("server state was never installed")
return _STATE
def require_session():
"""The live session, or an error that says how to get one."""
state = app()
if not state.connected:
raise errors.not_connected()
return state.session
def expiry_iso(seconds_left: float) -> str | None:
if seconds_left <= 0:
return None
return (datetime.now(UTC) + timedelta(seconds=seconds_left)).isoformat(timespec="seconds")
+116
View File
@@ -0,0 +1,116 @@
"""Live video from the drone's camera."""
from pathlib import Path
from typing import Annotated, Any
from fastmcp import Context, FastMCP
from fastmcp.tools import ToolResult
from fastmcp.utilities.types import Image
from mcp.types import TextContent
from pydantic import Field
from mcbebop.config import Settings
from mcbebop.media import video
from mcbebop.protocol import xml_index
from mcbebop.tools._common import require_session
def _sender(session):
"""Adapt a session to the plain callable `media.video` expects.
The video module deliberately knows nothing about sessions or the protocol
index, so the wiring lives here.
"""
async def send(name: str, args: dict[str, Any]):
return await session.send(xml_index.get(name), args)
return send
def _stamp() -> str:
from datetime import UTC, datetime
return datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"openWorldHint": True})
async def camera_snapshot(
ctx: Context,
max_width: Annotated[
int, Field(description="Downscale to this width before returning it.", ge=64, le=1920)
] = 0,
exposure: Annotated[
float | None,
Field(description="Exposure while capturing, -1.5 to 1.5. Restored afterwards.", ge=-1.5, le=1.5),
] = None,
settle_seconds: Annotated[
float,
Field(
description=(
"Discard this much video first. The opening frames reference parameter sets that "
"have not arrived yet, and the camera's exposure needs a moment to settle."
),
ge=0,
le=20,
),
] = 4.0,
) -> ToolResult:
"""Take one frame from the live camera and return it as an image.
Enables the video stream, grabs a frame, then turns the stream off
again. The stream is 856x480: the drone records 1080p internally but
only streams the lower resolution.
"""
session = require_session()
capture_dir = Path(settings.capture_dir)
sdp = video.write_sdp(capture_dir / "bebop.sdp")
out = capture_dir / f"frame-{_stamp()}.jpg"
sink = video.snapshot(sdp, out, settle_seconds=settle_seconds)
async with video.StreamSession(
sink, _sender(session), exposure=exposure, read_state=session.values
) as stream:
await stream.wait(settle_seconds + 20)
if not out.exists() or out.stat().st_size == 0:
raise video.VideoUnavailable(
"no frame was captured. Is ffmpeg installed, and is the drone's camera unobstructed?"
)
raw = out.read_bytes()
shown = video.downscale(raw, max_width or settings.snapshot_max_width)
attitude = session.values(["AttitudeChanged", "AltitudeChanged", "GpsLocationChanged"])
return ToolResult(
content=[
TextContent(type="text", text=f"frame saved to {out} ({len(raw)} bytes)"),
Image(data=shown, format="jpeg").to_image_content(),
],
structured_content={"path": str(out), "bytes": len(raw), "state_at_capture": attitude},
)
@mcp.tool(annotations={"openWorldHint": True})
async def camera_record(
ctx: Context,
seconds: Annotated[float, Field(description="How long to record for.", ge=1, le=300)] = 15.0,
exposure: Annotated[
float | None, Field(description="Exposure while recording, -1.5 to 1.5.", ge=-1.5, le=1.5)
] = None,
) -> dict:
"""Record the live camera to an MP4, copying the drone's H.264 without re-encoding."""
session = require_session()
capture_dir = Path(settings.capture_dir)
sdp = video.write_sdp(capture_dir / "bebop.sdp")
out = capture_dir / f"video-{_stamp()}.mp4"
sink = video.record(sdp, out, seconds)
async with video.StreamSession(
sink, _sender(session), exposure=exposure, read_state=session.values
) as stream:
await stream.wait(seconds + 20)
size = out.stat().st_size if out.exists() else 0
if not size:
raise video.VideoUnavailable("nothing was recorded. Is ffmpeg installed?")
return {"path": str(out), "bytes": size, "seconds": seconds}
+90
View File
@@ -0,0 +1,90 @@
"""The generic sender: any of Parrot's commands, gated by what it can do."""
from typing import Annotated, Any
from fastmcp import Context, FastMCP
from pydantic import Field
from mcbebop import errors
from mcbebop.config import Settings
from mcbebop.models import SendResult
from mcbebop.protocol import xml_index
from mcbebop.protocol.safety import ALWAYS_ALLOWED
from mcbebop.protocol.types import Direction, Tier
from mcbebop.tools._common import app, require_session
LOCKED_TIERS = (Tier.MOTION, Tier.ENVELOPE)
def check_allowed(spec) -> None:
"""Refuse a command the session has not earned the right to send.
Landing and Emergency are exempt on purpose: if the aircraft is airborne,
refusing to bring it down is the more dangerous answer.
"""
if spec.full_name in ALWAYS_ALLOWED or spec.tier not in LOCKED_TIERS:
return
if not app().arming.armed:
raise errors.needs_arming(spec.full_name, str(spec.tier))
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"openWorldHint": True})
async def send_command(
ctx: Context,
name: Annotated[
str, Field(description="Full name, e.g. 'ardrone3.PictureSettings.PictureFormatSelection'.")
],
args: Annotated[
dict[str, Any],
Field(
description=(
"Arguments by name. Enum arguments take the member name as a string. "
"Call command_info() for the names, types and allowed values."
)
),
] = {}, # noqa: B006 - FastMCP reads the default into the schema; never mutated
wait_for_confirm: Annotated[
bool,
Field(
description=(
"Also wait for the event Parrot says confirms this command, where one is declared. "
"Stronger evidence than the acknowledgement alone."
)
),
] = True,
) -> SendResult:
"""Send any command the drone understands.
Commands are classified by consequence. Observing and configuring work
straight away; anything that can move the aircraft or change its flight
envelope refuses until arm() is called.
"""
spec = xml_index.get(name)
if spec is None:
raise errors.unknown_command(name, [c.full_name for c in xml_index.search(name)][:5])
if spec.direction is Direction.FROM_DRONE:
raise errors.not_sendable(name)
check_allowed(spec)
session = require_session()
note = ""
if xml_index.supports_bebop2(spec.support) is False:
note = f"The XML does not list the Bebop 2 as supporting this (support={spec.support!r})."
elif spec.deprecated:
note = "Parrot marks this command deprecated; the drone may ignore it."
try:
result = await session.send(spec, dict(args), confirm=wait_for_confirm)
except ValueError as exc:
raise errors.bad_argument(name, str(exc)) from exc
return SendResult(
command=spec.full_name,
args=dict(args),
acked=bool(result.get("acked")),
confirmed=bool(result.get("confirmed")),
elapsed_ms=float(result.get("elapsed_ms", 0.0)),
note=note,
)
+100
View File
@@ -0,0 +1,100 @@
"""Opening and closing the link to the aircraft."""
import asyncio
from typing import Annotated
from fastmcp import Context, FastMCP
from pydantic import Field
from mcbebop import errors
from mcbebop.config import Settings
from mcbebop.models import ConnectionInfo
from mcbebop.state import SIM_TARGET
from mcbebop.tools._common import app, expiry_iso
def _info() -> ConnectionInfo:
state = app()
session = state.session
stats = session.link_stats() if state.connected else {}
return ConnectionInfo(
connected=state.connected,
target=state.target or "none",
armed=state.arming.armed,
armed_until=expiry_iso(state.arming.seconds_left),
telemetry_keys=stats.get("telemetry_keys", 0) or 0,
last_update_age=stats.get("last_event_age"),
undecodable_events=stats.get("undecodable_events", 0) or 0,
)
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"openWorldHint": True})
async def connect(
ctx: Context,
target: Annotated[
str,
Field(
description=(
"Where to connect: an IP address, 'drone' for the configured address "
f"({settings.drone_ip}), or 'sim' to run a built-in simulator instead of the "
"aircraft. Use 'sim' to rehearse anything that would move a real drone."
)
),
] = "drone",
) -> ConnectionInfo:
"""Open a session with the drone. Nothing else works until this succeeds.
The drone serves one controller at a time, so close any phone app first.
This machine must already have joined the drone's own Wi-Fi network.
"""
state = app()
async with state.lock:
if state.connected:
raise errors.already_connected(state.target)
from mcbebop.arsdk.session import DroneSession
from mcbebop.arsdk.types import HandshakeError
if target == SIM_TARGET:
from mcbebop.sim import FakeBebop
sim = FakeBebop()
sim.__enter__()
state.sim = sim
session = DroneSession(ip=sim.host, discovery_port=sim.discovery_port)
else:
ip = settings.drone_ip if target == "drone" else target
session = DroneSession(ip=ip)
try:
await session.connect()
except HandshakeError as exc:
await state.close()
status = getattr(exc, "status", None)
raise errors.handshake_refused(status) if status else errors.not_connected() from exc
except OSError as exc:
await state.close()
raise errors.not_connected() from exc
state.session = session
state.target = target if target == SIM_TARGET else getattr(session, "ip", target)
await session.request_full_state()
# The drone answers the state request as a burst over the next
# moment, so pause before reporting a count: returning zero keys
# from a healthy connection reads as a failure.
await asyncio.sleep(1.0)
return _info()
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def connection_status(ctx: Context) -> ConnectionInfo:
"""Whether a session is open, how fresh its telemetry is, and whether it is armed."""
return _info()
@mcp.tool(annotations={"openWorldHint": False})
async def disconnect(ctx: Context) -> ConnectionInfo:
"""Close the session. Also disarms, and stops the drone's video if it was started."""
state = app()
async with state.lock:
await state.close()
return _info()
+91
View File
@@ -0,0 +1,91 @@
"""Reading files off the drone over FTP, and its debug shell."""
import asyncio
from typing import Annotated, Literal
from fastmcp import Context, FastMCP
from fastmcp.exceptions import ToolError
from pydantic import Field
from mcbebop.config import Settings
from mcbebop.files import ftp, shell
from mcbebop.models import FileEntry
from mcbebop.tools._common import app
AreaName = Literal["media", "flightplans", "logs"]
def _host(settings: Settings) -> str:
"""FTP talks to the aircraft directly, so the simulator has nothing to serve."""
state = app()
if state.target == "sim":
raise ToolError(
"File access talks to the real aircraft's FTP server; the simulator has no filesystem. "
"Connect to the drone to use this."
)
return state.target or settings.drone_ip
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True})
async def list_files(
ctx: Context,
area: Annotated[
AreaName,
Field(
description=(
"media: photos and videos. flightplans: stored missions. "
"logs: the drone's own blackbox and debug files."
)
),
] = "media",
path: Annotated[str, Field(description="Subdirectory within the area. Empty is its root.")] = "",
) -> list[FileEntry]:
"""List files on the drone.
Read-only. The firmware-update channel is deliberately not reachable
through this tool.
"""
host = _host(settings)
entries = await asyncio.to_thread(ftp.list_dir, ftp.resolve_area(area), path, host=host)
return [FileEntry(name=e.name, size=e.size, is_dir=e.is_dir) for e in entries]
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True})
async def fetch_file(
ctx: Context,
area: Annotated[AreaName, Field(description="Which area the path is in.")],
path: Annotated[str, Field(description="Path within the area, as shown by list_files.")],
max_mb: Annotated[int, Field(description="Refuse anything larger.", ge=1, le=512)] = 32,
) -> dict:
"""Download one file from the drone to this machine, and return where it landed."""
host = _host(settings)
dest = await asyncio.to_thread(
ftp.fetch,
ftp.resolve_area(area),
path,
host=host,
capture_dir=settings.capture_dir,
max_bytes=max_mb * 1024 * 1024,
)
return {"path": str(dest), "bytes": dest.stat().st_size}
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True})
async def shell_read(
ctx: Context,
command: Annotated[str, Field(description=f"One of: {', '.join(sorted(shell.allowed_commands()))}.")],
args: Annotated[
list[str],
Field(description="Arguments. Shell metacharacters are refused rather than escaped."),
] = [], # noqa: B006 - read into the schema by FastMCP; never mutated
) -> dict:
"""Run one read-only command on the drone's debug shell.
The shell only exists after the drone's debug mode is switched on with
four short presses of its power button, and it does not survive a
reboot. It is an unauthenticated root shell, which is why this tool
takes an allow-list of command names rather than trying to filter a
free-form line.
"""
host = _host(settings)
result = await asyncio.to_thread(shell.run, command, list(args), host=host)
return {"command": result.command, "stdout": result.stdout}
+93
View File
@@ -0,0 +1,93 @@
"""Browsing Parrot's command set, so nothing has to be known in advance."""
from typing import Annotated, Literal
from fastmcp import Context, FastMCP
from pydantic import Field
from mcbebop import errors
from mcbebop.config import Settings
from mcbebop.models import ArgumentInfo, CommandDetail, CommandSummary
from mcbebop.protocol import xml_index
def _summary(spec) -> CommandSummary:
return CommandSummary(
name=spec.full_name,
title=spec.title,
tier=str(spec.tier),
direction=str(spec.direction),
bebop2=xml_index.supports_bebop2(spec.support),
deprecated=spec.deprecated,
)
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def list_commands(
ctx: Context,
search: Annotated[
str, Field(description="Case-insensitive substring of the name or title. Empty lists everything.")
] = "",
direction: Annotated[
Literal["any", "to_drone", "from_drone"],
Field(description="to_drone are commands you can send; from_drone are events the drone emits."),
] = "any",
tier: Annotated[
Literal["any", "observe", "config", "envelope", "motion"],
Field(description="Filter by what a command can do. motion and envelope require arm()."),
] = "any",
bebop2_only: Annotated[
bool,
Field(description="Hide commands the XML says the Bebop 2 does not support. Unstated ones stay."),
] = True,
) -> list[CommandSummary]:
"""Search the 264 commands Parrot defines. Start here when you do not know the name."""
out = []
for spec in xml_index.all_commands():
if search and search.lower() not in f"{spec.full_name} {spec.title}".lower():
continue
if direction != "any" and str(spec.direction) != direction:
continue
if tier != "any" and str(spec.tier) != tier:
continue
if bebop2_only and xml_index.supports_bebop2(spec.support) is False:
continue
out.append(_summary(spec))
return sorted(out, key=lambda c: c.name)
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def command_info(
ctx: Context,
name: Annotated[str, Field(description="Full name, e.g. 'ardrone3.MediaStreaming.VideoEnable'.")],
) -> CommandDetail:
"""Everything Parrot documents about one command: arguments, enum values, support, tier."""
spec = xml_index.get(name)
if spec is None:
near = [c.full_name for c in xml_index.search(name)][:5]
raise errors.unknown_command(name, near)
confirmed = []
for exp in spec.expectations:
other = xml_index.by_ids(exp.ids)
label = other.full_name if other else "-".join(str(i) for i in exp.ids)
if exp.fields:
label += " where " + ", ".join(f"{k}={v}" for k, v in exp.fields.items())
confirmed.append(label)
return CommandDetail(
**_summary(spec).model_dump(),
ids=list(spec.ids),
doc=spec.doc,
buffer=str(spec.buffer),
confirmed_by=confirmed,
args=[
ArgumentInfo(
name=a.name,
type=a.type,
doc=a.doc,
enum_values=[m.name for m in a.members],
)
for a in spec.args
],
)
+44
View File
@@ -0,0 +1,44 @@
"""Arming: the deliberate step before anything can move the aircraft."""
from typing import Annotated
from fastmcp import Context, FastMCP
from pydantic import Field
from mcbebop.config import Settings
from mcbebop.models import ArmState
from mcbebop.tools._common import app, expiry_iso
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"openWorldHint": False})
async def arm(
ctx: Context,
reason: Annotated[
str,
Field(description="Why, in your own words. Recorded, and shown back in connection_status."),
],
minutes: Annotated[
int, Field(description="How long the unlock lasts before it expires on its own.", ge=1, le=60)
] = 0,
) -> ArmState:
"""Unlock commands that can move the aircraft or change its flight envelope.
Only do this when the propellers are off, or the drone is somewhere it
is safe for it to move. The unlock expires on its own and is cleared by
disconnect. Landing and Emergency never need it.
"""
state = app()
state.arming.arm(reason, minutes or settings.arm_minutes)
return ArmState(
armed=True,
reason=reason,
expires_at=expiry_iso(state.arming.seconds_left),
note="Motion and envelope commands are unlocked until this expires.",
)
@mcp.tool(annotations={"openWorldHint": False})
async def disarm(ctx: Context) -> ArmState:
"""Re-lock motion commands immediately, without waiting for the timer."""
app().arming.disarm()
return ArmState(armed=False, note="Motion and envelope commands are locked again.")
+159
View File
@@ -0,0 +1,159 @@
"""Reading what the drone reports about itself."""
import asyncio
from typing import Annotated
from fastmcp import Context, FastMCP
from pydantic import Field
from mcbebop.config import Settings
from mcbebop.models import Check, Preflight, StateValue
from mcbebop.tools._common import require_session
# The drone says a position is unavailable by reporting this, not by omitting it.
GPS_UNAVAILABLE = 500.0
SENSOR_PREFIX = "SensorsStatesListChanged["
def _sensor_results(values: dict) -> dict[str, bool]:
"""Pull the self-test results back out of the per-element keys.
These arrive as one event per sensor, all carrying the same argument names,
which is why the session stores them under `Command[element]_arg`.
"""
out = {}
for key, value in values.items():
if key.startswith(SENSOR_PREFIX) and key.endswith("_sensorState"):
out[key.split("[", 1)[1].split("]", 1)[0]] = bool(value)
return out
def register(mcp: FastMCP, settings: Settings) -> None:
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def get_state(
ctx: Context,
keys: Annotated[
list[str],
Field(
description=(
"Which values to read, as '<Command>_<arg>' (e.g. 'BatteryStateChanged_percent'). "
"A bare prefix matches everything under it. Empty returns everything reported."
)
),
] = [], # noqa: B006 - read into the schema by FastMCP; never mutated
) -> dict[str, StateValue]:
"""Read the drone's reported state, with the age of each value.
Age matters: these are the last thing the drone said, not a fresh read.
A large age means the link went quiet, not that the value is current.
"""
session = require_session()
raw = session.state(keys or None)
return {k: StateValue(value=v["value"], age=round(v["age"], 2)) for k, v in raw.items()}
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def watch_state(
ctx: Context,
keys: Annotated[
list[str], Field(description="Values to watch, same form as get_state. A bare prefix matches.")
],
seconds: Annotated[float, Field(description="How long to watch for.", ge=0.5, le=60)] = 5.0,
) -> dict[str, list]:
"""Watch values change over a window, to see motion rather than a snapshot."""
session = require_session()
start = session.state(keys or None)
await asyncio.sleep(seconds)
end = session.state(keys or None)
changes: dict[str, list] = {}
for key, after in end.items():
before = start.get(key, {}).get("value")
if before != after["value"]:
changes[key] = [before, after["value"]]
return changes or {"(no change)": [None, None]}
@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
async def preflight_check(ctx: Context) -> Preflight:
"""A go/no-go summary: battery, link, sensor self-tests, motors, GPS and limits.
Advisory. It reports what the drone says about itself and cannot see
whether the propellers are on, the weather, or what is in the way.
"""
session = require_session()
v = session.values()
checks: list[Check] = []
blocking: list[str] = []
advisory: list[str] = []
def add(name: str, ok: bool | None, detail: str, blocks: bool = False) -> None:
checks.append(Check(name=name, ok=ok, detail=detail))
if ok is False:
(blocking if blocks else advisory).append(f"{name}: {detail}")
elif ok is None:
advisory.append(f"{name}: not reported")
battery = v.get("BatteryStateChanged_percent")
add(
"battery",
None if battery is None else battery >= 30,
"not reported" if battery is None else f"{battery}%",
blocks=True,
)
rssi = v.get("WifiSignalChanged_rssi")
add("link", None if rssi is None else rssi > -75, "not reported" if rssi is None else f"{rssi} dBm")
sensors = _sensor_results(v)
failed = sorted(n for n, ok in sensors.items() if not ok)
add(
"sensor self-tests",
None if not sensors else not failed,
"none reported"
if not sensors
else (f"failing: {', '.join(failed)}" if failed else f"{len(sensors)} ok"),
blocks=True,
)
motor = v.get("MotorErrorStateChanged_motorError")
motor = getattr(motor, "value", motor)
add(
"motors", None if motor is None else motor == "noError", str(motor or "not reported"), blocks=True
)
alert = getattr(v.get("AlertStateChanged_state"), "value", v.get("AlertStateChanged_state"))
add(
"alert state",
None if alert is None else alert == "none",
str(alert or "not reported"),
blocks=True,
)
flying = getattr(v.get("FlyingStateChanged_state"), "value", v.get("FlyingStateChanged_state"))
add("flying state", None if flying is None else flying == "landed", str(flying or "not reported"))
fixed = v.get("GPSFixStateChanged_fixed")
sats = v.get("NumberOfSatelliteChanged_numberOfSatellite")
add(
"gps",
None if fixed is None else bool(fixed),
"no fix" + (f", {sats} satellites" if sats is not None else "")
if not fixed
else f"fix, {sats} satellites",
)
calib = v.get("MagnetoCalibrationRequiredState_required")
add(
"compass calibration",
None if calib is None else not calib,
"not reported" if calib is None else ("required" if calib else "not required"),
)
geofence = v.get("NoFlyOverMaxDistanceChanged_shouldNotFlyOver")
max_dist = v.get("MaxDistanceChanged_current")
add(
"geofence",
None if geofence is None else bool(geofence),
f"off, max distance {max_dist} m" if geofence == 0 else f"on at {max_dist} m",
)
return Preflight(ready=not blocking, blocking=blocking, advisory=advisory, checks=checks)
+131
View File
@@ -0,0 +1,131 @@
"""The tools layer, driven through a real MCP client against the simulator."""
import pytest
from fastmcp import Client
from fastmcp.exceptions import ToolError
from mcbebop.protocol import xml_index
from mcbebop.protocol.safety import ALWAYS_ALLOWED
from mcbebop.protocol.types import Direction, Tier
from mcbebop.server import build_server
def data(result):
return result.structured_content if result.structured_content is not None else result.data
@pytest.fixture
async def client():
async with Client(build_server()) as c:
yield c
@pytest.fixture
async def connected(client):
await client.call_tool("connect", {"target": "sim"})
yield client
await client.call_tool("disconnect", {})
async def test_every_tool_is_registered(client):
names = {t.name for t in await client.list_tools()}
assert names == {
"connect", "disconnect", "connection_status", "list_commands", "command_info",
"send_command", "get_state", "watch_state", "preflight_check", "camera_snapshot",
"camera_record", "list_files", "fetch_file", "shell_read", "arm", "disarm",
} # fmt: skip
async def test_tools_refuse_before_connecting(client):
calls = (
("get_state", {}),
("preflight_check", {}),
("send_command", {"name": "common.Common.AllStates"}),
)
for tool, args in calls:
with pytest.raises(ToolError, match=r"[Nn]ot connected"):
await client.call_tool(tool, args)
async def test_connect_reports_live_telemetry(connected):
info = data(await connected.call_tool("connection_status", {}))
assert info["connected"] and info["target"] == "sim"
assert info["telemetry_keys"] > 20, "the drone's state burst should have arrived"
assert info["last_update_age"] is not None and info["last_update_age"] < 10
assert info["undecodable_events"] == 0, "every event the simulator sends should decode"
@pytest.mark.parametrize(
"name",
[c.full_name for c in xml_index.all_commands()
if c.direction is Direction.TO_DRONE and c.tier in (Tier.MOTION, Tier.ENVELOPE)
and c.full_name not in ALWAYS_ALLOWED],
) # fmt: skip
async def test_every_locked_command_is_refused_unarmed(connected, name):
"""The gate must hold for all of them, not just the obvious ones."""
with pytest.raises(ToolError, match="locked"):
await connected.call_tool("send_command", {"name": name})
@pytest.mark.parametrize("name", sorted(ALWAYS_ALLOWED))
async def test_landing_and_emergency_are_never_locked(connected, name):
"""Refusing to bring down an airborne aircraft is the more dangerous answer."""
result = data(await connected.call_tool("send_command", {"name": name}))
assert result["acked"]
async def test_arming_unlocks_and_disarming_relocks(connected):
takeoff = "ardrone3.Piloting.TakeOff"
with pytest.raises(ToolError, match="locked"):
await connected.call_tool("send_command", {"name": takeoff})
armed = data(await connected.call_tool("arm", {"reason": "test", "minutes": 1}))
assert armed["armed"] and armed["expires_at"]
assert data(await connected.call_tool("send_command", {"name": takeoff}))["acked"]
await connected.call_tool("disarm", {})
with pytest.raises(ToolError, match="locked"):
await connected.call_tool("send_command", {"name": takeoff})
async def test_events_cannot_be_sent(connected):
with pytest.raises(ToolError, match="event the drone emits"):
await connected.call_tool("send_command", {"name": "common.CommonState.BatteryStateChanged"})
async def test_unknown_command_suggests_alternatives(connected):
with pytest.raises(ToolError, match="No command named"):
await connected.call_tool("send_command", {"name": "ardrone3.Piloting.TakeOf"})
async def test_preflight_sees_a_failing_sensor(connected):
"""The simulator fails its magnetometer; a flat telemetry store would hide it."""
report = data(await connected.call_tool("preflight_check", {}))
sensors = next(c for c in report["checks"] if c["name"] == "sensor self-tests")
assert sensors["ok"] is False and "magnetometer" in sensors["detail"]
assert report["ready"] is False
assert any("magnetometer" in b for b in report["blocking"])
async def test_command_info_carries_the_confirming_event(connected):
info = data(await connected.call_tool("command_info", {"name": "ardrone3.MediaStreaming.VideoEnable"}))
assert info["ids"] == [1, 21, 0]
assert info["args"][0]["name"] == "enable"
assert any("VideoEnableChanged" in c for c in info["confirmed_by"])
async def test_list_commands_filters(connected):
args = {"direction": "to_drone", "bebop2_only": False}
sendable = data(await connected.call_tool("list_commands", args))
sendable = sendable.get("result", sendable)
# 101 in Parrot's XML, plus FlatTrim which the firmware accepts but the XML omits.
assert len(sendable) == 102
motion = data(await connected.call_tool("list_commands", {"tier": "motion", "bebop2_only": False}))
motion = motion.get("result", motion)
assert all(c["tier"] == "motion" for c in motion)
async def test_file_tools_reject_the_simulator(connected):
"""The simulator has no filesystem; saying so beats a confusing timeout."""
with pytest.raises(ToolError, match="simulator has no filesystem"):
await connected.call_tool("list_files", {"area": "media"})