MCP robot tools

Provide MCP tools for robot state, camera observations, pose editing, image analysis, and task construction. Most tools return JSON text; ping() returns plain text.

Import requires ROS, OpenCV, and MCP and probes display settings. Beamline configuration is cached after its first successful load; pose files are reread for each registry operation. Image-saving tools may not report a failed cv2.imwrite.

src.beambot.mcp_server.beambot_mcp_server · Source

Clients and integrations · Guide · Symbol index

How it is used

MCP clients call the decorated tool functions for robot state, saved poses, camera observations, detection, and task construction. main() runs the stdio server. ROS2Bridge creates its node and background executor on first node access; the bridge callbacks maintain the caches used by tools.

Choose the tool by its effect: capture_image() requests a Zivid capture or reads a ZED frame; detection tools analyse cached data; vision_target() and pickup_tip() return task JSON for a separate submission step. stop_robot() requests orchestrator cancellation, not a hardware emergency stop. Grid task generation shares expand_grid_target() with the orchestrator, so both paths apply the same grid rules.

Start here

Callable	Purpose
main	Configure stderr logging and run FastMCP’s stdio transport; on server exceptions append a traceback to the crash log, then always call bridge shutdown.
get_robot_state	Return cached joint positions, active tool, execution state, and beamline identity. `system_running` only indicates that joint positions have arrived; it does not establish their freshness or overall system health.
capture_image	For Zivid call the existing /capture Trigger and wait for image plus optional cloud; for ZED use the latest stream frame or wait briefly for its first frame. Save the RGB cache as a BGR-encoded image.
detect_objects	Detect objects in the cached image using HSV, ArUco, or YOLO. YOLO receives an RGB-to-BGR conversion. Optionally add point-cloud coordinates, transform them to `base_link`, and save detected objects. This function does not capture images or execute tasks. Depth lookup requires image-aligned, little-endian XYZ float32 records. It searches expanding square pixel rings and returns the first valid point without averaging.
vision_target	Build task JSON for a configured vision target and return it to the MCP client. Submission and execution are separate steps.

Example

For a configured tip_rack target, an MCP client can call vision_target with:

{"target_name": "tip_rack", "row": 0, "col": 0}

The response contains a task_json string plus target metadata, or an error. Generating this document reads configuration and saved poses; it does not submit the task. Grid rows and columns are zero-based.

Functions

`main()`

Configure stderr logging and run FastMCP’s stdio transport; on server exceptions append a traceback to the crash log, then always call bridge shutdown.

Entry point · Source

main()

Returns: None after server stops or a caught server Exception is logged.

Errors: MCP Exceptions are logged and swallowed after crash logging; crash-log writes and bridge cleanup can propagate. KeyboardInterrupt bypasses the Exception handler but still runs finally.

Side effects: Runs a blocking MCP server, logs to stderr, may append /tmp/beambot_mcp_crash.log and shuts down bridge/global ROS resources.

`get_robot_state()`

Return cached joint positions, active tool, execution state, and beamline identity. system_running only indicates that joint positions have arrived; it does not establish their freshness or overall system health.

Entry point · Source

@mcp.tool()
async get_robot_state() -> str

Returns: JSON with beamline (or null), system_running, gripper (unknown fallback), execution_state, joints_deg (name→two-decimal values or null), plus vacuum_status when active gripper is epick and a status exists. All cached joint positions are converted to degrees, including any nonrevolute entries.

Errors: Config errors are suppressed to beamline=null; bridge and malformed numeric data errors propagate.

Side effects: Initializes/reads ROS caches and reads beamline config; no motion.

`capture_image()`

For Zivid call the existing /capture Trigger and wait for image plus optional cloud; for ZED use the latest stream frame or wait briefly for its first frame. Save the RGB cache as a BGR-encoded image.

Entry point · Source

@mcp.tool()
async capture_image(
    camera: str = 'zivid',
    mode: str = '3d',
    save_path: str = DEFAULT_IMAGE_PATH,
    timeout: float = 30.0,
) -> str

Parameter	Meaning
`camera`	Exact zivid or zed camera selector.
`mode`	Only exact 3d sets need_cloud=True for Zivid; any other string behaves as image-only waiting. It does not select a distinct 2D acquisition service. ZED ignores it.
`save_path`	Image destination; parent directories are created and existing file may be overwritten.
`timeout`	Zivid response/data budget in seconds; ZED first-image wait is capped at five seconds.

Returns: JSON {image_path,camera,camera_frame,width,height,has_pointcloud,capture_timestamp}; error JSON for unknown/unavailable camera or failed capture. has_pointcloud merely reports a nonempty cache and may describe a previous Zivid capture.

Errors: Expected capture/data failures return error JSON. ROS, OpenCV and filesystem errors can propagate; cv2.imwrite False is ignored.

Side effects: May initialize ROS, trigger Zivid, wait for streaming data, create directories and overwrite an image.

`detect_objects()`

Detect objects in the cached image using HSV, ArUco, or YOLO. YOLO receives an RGB-to-BGR conversion. Optionally add point-cloud coordinates, transform them to base_link, and save detected objects. This function does not capture images or execute tasks.

Depth lookup requires image-aligned, little-endian XYZ float32 records. It searches expanding square pixel rings and returns the first valid point without averaging.

Entry point · Source

@mcp.tool()
async detect_objects(
    method: str = 'hsv_color',
    camera: str = 'zivid',
    hue_low: int = 100,
    hue_high: int = 130,
    sat_min: int = 80,
    val_min: int = 80,
    min_area: int = 200,
    marker_ids: str = '',
    yolo_model: str = 'yolov8n.pt',
    yolo_confidence: float = 0.25,
    yolo_classes: str = '',
    transform_to_base: bool = True,
    save_path: str = DEFAULT_ANNOTATED_PATH,
) -> str

Parameter	Meaning
`method`	Exact hsv_color, marker or yolo.
`camera`	Cached data source: zivid or zed.
`hue_low`	HSV lower hue bound on OpenCV scale; used only for hsv_color.
`hue_high`	HSV upper hue bound; one nonwrapping interval for hsv_color.
`sat_min`	HSV minimum saturation (0–255).
`val_min`	HSV minimum value/brightness (0–255).
`min_area`	Minimum external-contour area in square pixels for HSV.
`marker_ids`	Comma-separated integer IDs for marker mode; empty means all.
`yolo_model`	Generic YOLO weight/model path passed to the shared detector.
`yolo_confidence`	Minimum YOLO confidence fraction, conventionally 0–1, without local validation.
`yolo_classes`	Comma-separated numeric IDs if every token is digits, otherwise case-insensitive class names; empty keeps all.
`transform_to_base`	Request base_link coordinates for valid depth; failure leaves base_xyz null.
`save_path`	Annotated image destination; parent directories are created only for nonempty detections.

Returns: JSON with detections/count/method/annotated_image_path/camera/coordinate_frame. Each detection has pixel_x/y, camera_xyz and base_xyz, plus area (HSV), marker_id/corners (marker), or class/confidence/bbox (YOLO). Empty detections return a smaller count=0 response without image path. coordinate_frame says base_link when requested even if transforms fail.

Errors: Unknown method/camera or missing image returns error JSON. Bad ID text, detector/model, depth, filesystem and OpenCV exceptions propagate; image-write false results are ignored.

Side effects: Reads ROS caches and TF, may load generic YOLO weights (its library may resolve/download named weights), and writes annotation; does not trigger acquisition or move hardware.

`vision_target()`

Build task JSON for a configured vision target and return it to the MCP client. Submission and execution are separate steps.

Entry point · Source

@mcp.tool()
async vision_target(
    target_name: str,
    element_index: int = 0,
    row: int = -1,
    col: int = -1,
    tag_id: int = -1,
) -> str

Parameter	Meaning
`target_name`	Exact key in config.vision_targets.
`element_index`	Zero-based row-major grid index; ignored for offset mode or when both row and col are nonnegative.
`row`	Zero-based explicit row; used only together with a nonnegative col.
`col`	Zero-based explicit column; used only together with a nonnegative row.
`tag_id`	Nonnegative marker ID override; any negative value uses cfg.marker_id.

Returns: Indented JSON string containing the builder success schema or error.

Errors: Propagates errors not handled by _build_vision_target_tasks.

Side effects: Reads config and pose files through the builder; does not command hardware.

`ping()`

Confirm the MCP handler can execute without touching the lazy ROS bridge.

Entry point · Source

@mcp.tool()
async ping() -> str

Returns: Plain string pong, not JSON.

Errors: No explicit failure path.

`stop_robot()`

Request cancellation of orchestrator goals in a worker thread and return a JSON response. This is software action cancellation, not a hardware emergency stop.

Entry point · Source

@mcp.tool()
async stop_robot() -> str

Returns: JSON string with result containing the cancellation status message; no Boolean success field.

Errors: Bridge/future errors propagate; service/timeouts reported by cancel_all_goals remain ordinary result strings.

Side effects: Initializes ROS if needed and requests cancellation of all orchestrator goals; does not provide a hardware emergency stop.

`get_vacuum_status()`

Report whether any ePick status has been received and interpret statuses 1/2 as object_detected. It does not check status age or enforce an interlock.

Entry point · Source

@mcp.tool()
async get_vacuum_status() -> str

Returns: JSON with available and gripper; absent data adds note, present data adds status string and object_detected Boolean. Unknown integers are rendered UNKNOWN(n).

Errors: Bridge initialization errors propagate; no explicit service wait.

Side effects: Initializes ROS if needed and reads the cached vacuum status.

`get_saved_poses()`

Read saved poses and optionally filter names by a case-insensitive substring.

Entry point · Source

@mcp.tool()
async get_saved_poses(filter: str='') -> str

Parameter	Meaning
`filter`	Substring matched against lowercased keys; empty string keeps all entries.

Returns: JSON with poses and count, plus message for an initially empty registry; filtered-empty results omit that message.

Errors: File/YAML/invalid mapping or nonstring-key errors propagate rather than returning error JSON.

Side effects: Reads configuration and pose registry; does not initialize ROS.

`save_pose()`

Add or replace a pose, using six explicit degree values or every cached joint position converted from radians. Round values to two decimals and rewrite the registry.

Entry point · Source

@mcp.tool()
async save_pose(name: str, joints_deg: list=None, description: str='') -> str

Parameter	Meaning
`name`	Registry key; no empty-name or naming-pattern validation.
`joints_deg`	Explicit list must have length six and float-convertible values; None uses cached positions without reordering or enforcing six joints. Nonfinite values are not rejected.
`description`	Accepted but ignored; no comment or description is persisted.

Returns: JSON {saved,joints_deg,overwritten,total_poses}; error JSON for wrong explicit length or unavailable live joint data.

Errors: Numeric conversion, YAML/file and malformed-registry errors propagate; only the two explicit checks return error JSON.

Side effects: Reads/replaces pose YAML; initializes ROS only when joints are omitted. Does not command motion.

`delete_pose()`

Remove one exact pose key from the registry and replace the file.

Entry point · Source

@mcp.tool()
async delete_pose(name: str) -> str

Parameter	Meaning
`name`	Case-sensitive saved pose name.

Returns: JSON {deleted,remaining_poses}, or {error,available} listing keys when absent.

Errors: File/YAML/invalid registry errors propagate.

Side effects: Reads configuration/registry and writes the registry only when the name exists.

`set_cup_profile()`

Run ros2 param set for /beambot_orchestrator cup_profile; changing the string takes effect on a subsequent ePick MoveIt launch, not by rebuilding the current model.

Entry point · Source

@mcp.tool()
async set_cup_profile(name: str) -> str

Parameter	Meaning
`name`	Cup-profile string passed to ROS; this wrapper does not check suction_cups.yaml membership.

Returns: JSON {cup_profile,message} for exit code zero; otherwise error JSON, with note on nonzero exit.

Errors: All subprocess exceptions including the five-second timeout become error JSON; CLI return-code failures use stderr.

Side effects: Runs a ROS CLI subprocess and changes the orchestrator parameter; no immediate launch/rebuild.

`detect_sample_yolo()`

Infer with the custom sample detector, select the highest-confidence box across results, and look up center depth with a 20-pixel search before transforming to base_link. Shared depth lookup assumes image-aligned little-endian XYZ float32 cloud records; it returns the first valid point on expanding square pixel rings, not an averaged depth.

Entry point · Source

@mcp.tool()
async detect_sample_yolo(
    camera: str = 'zivid',
    confidence: float = 0.5,
    save_path: str = '/tmp/sample_detection_yolo.jpg',
) -> str

Parameter	Meaning
`camera`	zivid or zed cached RGB and cloud; both must exist.
`confidence`	YOLO confidence threshold fraction passed to model inference.
`save_path`	Annotation output path; parent directories are not created here.

Returns: JSON with pickup_base_xyz (metres or null), detection_pixel, confidence, bbox, bbox_size_px, num_detections and annotated_image_path. No detections returns error JSON and still attempts to plot/save the first result.

Errors: Unknown/missing camera data and absent weight file return error JSON; other inference/depth/TF/image errors propagate. cv2.imwrite False is ignored.

Side effects: May initialize ROS/model, perform inference/TF lookup and overwrite an annotation; no acquisition or motion.

`get_point_3d()`

Read a 3D point for a pixel in the cached RGB image. The optional GUI can adjust the pixel; the result can be transformed to base_link. A successful lookup attempts to save an annotated image even with confirm=False.

Depth lookup requires image-aligned, little-endian XYZ float32 records. It searches expanding square pixel rings and returns the first valid point without averaging.

Entry point · Source

@mcp.tool()
async get_point_3d(
    pixel_x: int,
    pixel_y: int,
    camera: str = 'zivid',
    transform_to_base: bool = True,
    search_radius: int = 10,
    confirm: bool = True,
    save_path: str = '/tmp/beambot_point3d.jpg',
) -> str

Parameter	Meaning
`pixel_x`	Original image column in pixels, checked before and after GUI adjustment.
`pixel_y`	Original image row in pixels.
`camera`	zivid or zed cached image/cloud selector.
`transform_to_base`	If true, try stamped/latest TF; missing TF leaves base_xyz null.
`search_radius`	Nonnegative integer pixel radius passed to shared depth lookup; exact pixel is tried first, then expanding square perimeters. Negative radius skips neighborhood search; no local validation.
`confirm`	Open the interactive point selector when true; it reuses /tmp/beambot_capture.jpg if present, which may not match the current cache/camera.
`save_path`	Annotation path; creates parent directories and may overwrite the file.

Returns: JSON with pixel_x/y, camera_xyz (metres), camera/camera_frame, base_xyz (metres or null), base_frame and annotated_image_path. Returns error JSON for invalid data/pixels/depth/GUI failure, or cancelled=true after user cancellation.

Errors: Handled availability/bounds/selection errors become JSON; unexpected OpenCV, filesystem, depth or worker errors propagate. Image write success flags are ignored.

Side effects: May initialize ROS, create a fallback capture image, open GUI, query TF and write annotation; no robot command.

`get_tf_transform()`

Look up the latest source-to-target TF in a worker and report translation, quaternion, Euler angles and homogeneous matrix.

Entry point · Source

@mcp.tool()
async get_tf_transform(
    source_frame: str = 'flange',
    target_frame: str = 'base_link',
    timeout: float = 2.0,
) -> str

Parameter	Meaning
`source_frame`	Frame coordinates are transformed from; default flange.
`target_frame`	Destination frame; default base_link.
`timeout`	Seconds passed to node.lookup_transform, which performs separate availability and lookup waits.

Returns: JSON with source_frame, target_frame, translation xyz in metres, rotation_quaternion xyzw, rotation_rpy_degrees and matrix_4x4. Translation/quaternion/matrix round to six decimals and RPY to four; unavailable TF returns error JSON.

Errors: Bridge initialization, numeric and unexpected TF errors propagate; ordinary unavailable transforms become error JSON.

Side effects: Initializes bridge in worker if needed and reads TF; default flange does not substitute tool0 orientation.

`get_recent_logs()`

Return recent matching ROS launch-log entries, newest first. Read the file while retaining its final 2,000 lines, then filter by severity and logger-name prefix.

Entry point · Source

@mcp.tool()
async get_recent_logs(severity: str='ERROR', logger: str='', count: int=30) -> str

Parameter	Meaning
`severity`	Case-insensitive minimum DEBUG/INFO/WARN/ERROR/FATAL; unrecognized values use ERROR threshold, while metadata retains the supplied text.
`logger`	Case-sensitive logger-name prefix; empty matches all.
`count`	Maximum accepted records; no validation means zero/negative still returns one matching record if one exists.

Returns: JSON {logs,count,filter}; each log contains process, logger, severity and message. Missing/unreadable file returns error plus empty logs/count. Multiline/unmatched lines are omitted.

Errors: Read failures become error JSON; parsing uses a fixed regex and unknown log severities count as zero.

Side effects: Reads all of /tmp/beambot_launch.log in a worker to retain the tail; memory is bounded but file I/O is not.

`pickup_tip()`

Build task JSON for the configured tip_rack target without executing it. It uses the same grid expansion as the parser macro, through the MCP request format.

Entry point · Source

@mcp.tool()
async pickup_tip(tip_index: int=0, row: int=-1, col: int=-1) -> str

Parameter	Meaning
`tip_index`	Zero-based row-major element index; valid range comes from configured grid size, not a hard-coded 96.
`row`	Zero-based row override when both row and col are nonnegative.
`col`	Zero-based column override when both row and col are nonnegative.

Returns: Indented JSON containing task_json/grid metadata or builder error.

Errors: Configured-target, index and unhandled config/pose errors follow _build_vision_target_tasks.

Side effects: Reads configuration/poses and returns proposed tasks only.

Internal callbacks and helpers

`_poses_file_path()`

Resolve poses_file from the active beamline config through the shared path resolver.

Internal helper · Source

_poses_file_path() -> str

Returns: Resolved path string; empty string when poses_file is empty or after a caught exception. Resolution is relative to the discovered workspace root, or the config directory if no src/ ancestor is found.

Errors: Catches all Exceptions and logs them. Empty resolution is hazardous for callers that resolve Path(“”) to the current directory.

Side effects: Reads beamline configuration and may log an error. The shared loader reads YAML only on its first successful call (or after explicit cache reset).

Referenced in source (3)

_read_poses_file — src.beambot.mcp_server.beambot_mcp_server
_write_poses_file — src.beambot.mcp_server.beambot_mcp_server
get_saved_poses — src.beambot.mcp_server.beambot_mcp_server

`_detect_display()`

Use DISPLAY/XAUTHORITY from the environment, otherwise inspect desktop-process environments for DISPLAY and common Xauthority paths.

Internal helper · Source

_detect_display() -> tuple[str, str]

Returns: (display, xauth) strings; display falls back to :1 and xauth may remain empty.

Errors: The shell probe is bounded to three seconds and any probe exception falls back to :1; filesystem probe errors are not explicitly caught.

Side effects: May spawn a bash inspection command and test auth paths. Called at module import; does not set environment variables itself.

`_detect_hsv_color()`

Threshold an RGB image in OpenCV HSV space, apply a 7×7 elliptical opening, and compute centroids of external contours passing the area threshold.

Internal helper · Source

_detect_hsv_color(
    rgb_image: np.ndarray,
    hue_low: int = 100,
    hue_high: int = 130,
    sat_min: int = 80,
    val_min: int = 80,
    min_area: int = 200,
) -> list[tuple[int, int, int]] | None

Parameter	Meaning
`rgb_image`	H×W×3 RGB image array accepted by cv2.cvtColor.
`hue_low`	Inclusive HSV hue lower threshold, OpenCV hue scale approximately 0–179.
`hue_high`	Inclusive hue upper threshold; a single contiguous interval is used, without wraparound.
`sat_min`	Inclusive saturation lower threshold, 0–255 scale.
`val_min`	Inclusive brightness/value lower threshold, 0–255 scale.
`min_area`	Minimum cv2 contour area in square pixels; equal area passes.

Returns: List of (integer center_x, center_y, truncated contour_area) tuples sorted by descending area, or None if none survive.

Errors: OpenCV and numeric errors propagate; thresholds are not validated. Zero-moment contours are skipped.

Side effects: Allocates masks/contours only.

Example

image = np.zeros((80, 80, 3), dtype=np.uint8)
image[20:60, 20:60] = [0, 0, 255]  # Blue in RGB.
hits = _detect_hsv_color(image)
assert hits and hits[0][:2] == (39, 39)

Referenced in source (1)

detect_objects — src.beambot.mcp_server.beambot_mcp_server

`_detect_aruco_markers()`

Detect ArUco corners through the shared detection implementation, then compute mean-corner center pixels and optionally retain selected IDs.

Internal helper · Source

_detect_aruco_markers(
    rgb_image: np.ndarray,
    marker_ids: list[int] | None = None,
    dictionary_name: str = 'DICT_4X4_50',
) -> list[tuple[int, int, int, list]] | None

Parameter	Meaning
`rgb_image`	RGB image array.
`marker_ids`	List of integer marker IDs to retain; None keeps every ID, while an empty list keeps none.
`dictionary_name`	One of DICT_4X4_50, DICT_4X4_100, DICT_5X5_50, DICT_5X5_100, DICT_6X6_50 or DICT_6X6_250; unknown names fall back to DICT_4X4_50.

Returns: List of (cx, cy, marker_id, four-corner-list) tuples in detector order, or None for no retained markers.

Errors: OpenCV/shared detector errors propagate.

Side effects: Image analysis only.

Referenced in source (1)

detect_objects — src.beambot.mcp_server.beambot_mcp_server

`_annotate_image()`

Copy RGB to BGR and draw a green/red center plus a label for every detection, preferring base coordinates, then camera coordinates, when available.

Internal helper · Source

_annotate_image(rgb_image: np.ndarray, detections: list[dict[str, Any]]) -> np.ndarray

Parameter	Meaning
`rgb_image`	RGB image array to copy.
`detections`	Dictionaries with pixel_x/pixel_y; optional marker_id, base_xyz or camera_xyz (metres) control labels.

Returns: New BGR image array suitable for cv2.imwrite.

Errors: Missing pixel keys, invalid coordinates and OpenCV errors propagate.

Side effects: Allocates/draws into a new image; does not write a file.

Referenced in source (1)

detect_objects — src.beambot.mcp_server.beambot_mcp_server

`_resolve_camera_state()`

Select the RGB/cloud/stamp caches and configured optical-frame name for a camera.

Internal helper · Source

_resolve_camera_state(
    node: ROS2BridgeNode,
    camera: str,
) -> tuple[np.ndarray | None, PointCloud2 | None, Any | None, str]

Parameter	Meaning
`node`	ROS2BridgeNode with cached Zivid and ZED data.
`camera`	Exact string zivid or zed.

Returns: (rgb, cloud, stamp, frame); any cached data may be None and references are not copied or synchronized.

Errors: ValueError for an unknown camera.

Side effects: Reads node fields only.

Referenced in source (3)

detect_objects — src.beambot.mcp_server.beambot_mcp_server
detect_sample_yolo — src.beambot.mcp_server.beambot_mcp_server
get_point_3d — src.beambot.mcp_server.beambot_mcp_server

`_read_poses_file()`

Resolve the configured pose registry and yaml.safe_load its contents, substituting {} for missing paths or falsey YAML values.

Internal helper · Source

_read_poses_file() -> dict

Returns: Loaded YAML value, normally a pose mapping, or {}; no mapping/entry validation is performed.

Errors: Read/YAML errors propagate. An unresolved empty configured path becomes the current directory and can raise IsADirectoryError.

Side effects: Reads beamline configuration and the pose registry.

Referenced in source (4)

_build_vision_target_tasks — src.beambot.mcp_server.beambot_mcp_server
delete_pose — src.beambot.mcp_server.beambot_mcp_server
get_saved_poses — src.beambot.mcp_server.beambot_mcp_server
save_pose — src.beambot.mcp_server.beambot_mcp_server

`_write_poses_file()`

Dump the complete pose mapping into a temporary file in the registry directory and replace the resolved destination.

Internal helper · Source

_write_poses_file(poses: dict)

Parameter	Meaning
`poses`	Serializable pose registry, normally name→degree-list; existing comments/formatting are not preserved.

Returns: None after replacement.

Errors: Catches BaseException during dumping/replacement to remove the temp file, then rethrows; cleanup errors may mask the original. Resolution, mkdir and mkstemp failures propagate. No concurrent read-modify-write lock is used.

Side effects: Creates parent directories and replaces the pose YAML atomically at rename; does not fsync. An empty resolved path can target a directory and fail.

Referenced in source (2)

delete_pose — src.beambot.mcp_server.beambot_mcp_server
save_pose — src.beambot.mcp_server.beambot_mcp_server

`capture_image._wait_for_zed()`

Poll the closed-over node every 0.1 seconds until a ZED RGB cache exists or min(timeout, 5) seconds expires.

Internal helper · Source

_wait_for_zed()

Returns: Boolean indicating whether zed_rgb is non-None after polling.

Errors: Negative timeout results in no polling; other timing/access errors propagate.

Side effects: Sleeps in an asyncio executor worker; does not trigger ZED.

Referenced in source (1)

capture_image — src.beambot.mcp_server.beambot_mcp_server

`_transform_point_to_base()`

Offload a stamped-then-latest TF lookup and homogeneous point transform to an asyncio worker.

Internal helper · Source

async _transform_point_to_base(
    node: ROS2BridgeNode,
    camera_xyz: tuple[float, float, float],
    camera_frame: str,
    capture_stamp = None,
) -> tuple[float, float, float] | None

Parameter	Meaning
`node`	Bridge node providing lookup_transform.
`camera_xyz`	Three camera-frame coordinates in metres.
`camera_frame`	Source optical-frame name.
`capture_stamp`	Optional ROS message timestamp; its lookup is preferred, but latest TF is used after failure.

Returns: (x,y,z) float tuple in base_link metres, or None when neither TF lookup succeeds.

Errors: Unavailable TF is logged and returns None; numeric/attribute/worker exceptions propagate. Latest-TF fallback can represent a different robot pose than capture.

Side effects: Queries TF in a worker thread and logs fallback/failure.

Referenced in source (3)

detect_objects — src.beambot.mcp_server.beambot_mcp_server
detect_sample_yolo — src.beambot.mcp_server.beambot_mcp_server
get_point_3d — src.beambot.mcp_server.beambot_mcp_server

`_transform_point_to_base._do_transform()`

Use closed-over camera coordinates and node to get stamped TF when available, fall back to latest, then multiply a homogeneous point by quaternion rotation plus translation.

Internal helper · Source

_do_transform()

Returns: Three base_link floats, or None after both TF attempts fail.

Errors: No broad catch; lookup helper handles TransformException, while malformed transforms/numbers propagate.

Side effects: Reads the TF buffer and logs fallback/errors in an executor worker.

Referenced in source (1)

_transform_point_to_base — src.beambot.mcp_server.beambot_mcp_server

`_confirm_point_via_gui()`

Launch the sibling point_selector_gui.py with initial coordinates and detected X11 environment, then parse its JSON stdout.

Internal helper · Source

async _confirm_point_via_gui(
    image_path: str,
    pixel_x: int,
    pixel_y: int,
    timeout: float = 120.0,
) -> dict[str, Any] | None

Parameter	Meaning
`image_path`	Image file displayed by the GUI subprocess.
`pixel_x`	Initial image-column coordinate in original pixels.
`pixel_y`	Initial image-row coordinate in original pixels.
`timeout`	Maximum communicate wait in seconds before killing the child.

Returns: Confirmed-point dictionary; None for a clean unconfirmed result; or an error dictionary. Nonzero child exits may return its parsed error JSON with stderr appended.

Errors: Missing script, launch/timeout and JSON parse failures become error dictionaries. A timeout kills without explicitly awaiting process reaping; malformed nonobject JSON can still raise later.

Side effects: Spawns a GUI process, displays an interactive window, captures stderr and may kill the child.

Referenced in source (1)

get_point_3d — src.beambot.mcp_server.beambot_mcp_server

`_get_yolo_model()`

Lazily load and cache Ultralytics YOLO from beambot/models/sample_detector.pt.

Internal helper · Source

_get_yolo_model()

Returns: The shared YOLO model instance; repeated calls return the cached object.

Errors: FileNotFoundError if weights are absent; ultralytics import/model-loading errors propagate.

Side effects: Reads model weights, initializes inference resources and logs first load; no locking around first load.

Referenced in source (1)

detect_sample_yolo — src.beambot.mcp_server.beambot_mcp_server

`get_tf_transform._lookup()`

Initialize/access the bridge node in the worker and fetch the latest transform using the outer frame/timeout arguments.

Internal helper · Source

_lookup()

Returns: TransformStamped, or None when the node lookup fails.

Errors: Propagates bridge/lookup errors outside handled TF failures.

Side effects: May initialize ROS and queries TF in an executor worker.

Referenced in source (1)

get_tf_transform — src.beambot.mcp_server.beambot_mcp_server

`get_recent_logs._read_tail()`

Read the complete launch log into a maxlen=2000 deque with replacement decoding for invalid text bytes.

Internal helper · Source

_read_tail()

Returns: Deque containing the last up to 2,000 lines, or an exception-message string.

Errors: Catches all Exceptions during open/iteration and returns str(exception).

Side effects: Reads /tmp/beambot_launch.log in a background worker.

Referenced in source (1)

get_recent_logs — src.beambot.mcp_server.beambot_mcp_server

`_load_beamline_config()`

Read the active beamline configuration with the shared loader and discard its path.

Internal helper · Source

_load_beamline_config() -> dict

Returns: Shared cached config dictionary, or {} on any Exception; callers must not mutate the successful shared result.

Errors: Logs all caught configuration/import failures; caller-specific shape errors are not validated here.

Side effects: Reads configuration and may log errors. The shared loader reads YAML only on its first successful call (or after explicit cache reset).

Referenced in source (1)

_load_vision_targets — src.beambot.mcp_server.beambot_mcp_server

`_load_vision_targets()`

Extract vision_targets from the shared beamline configuration.

Internal helper · Source

_load_vision_targets() -> dict

Returns: Configured vision_targets value (shared with the cached config), or {} when absent; no explicit mapping validation.

Errors: Shared loader errors yield {}; malformed nonmapping config can propagate attribute errors.

Side effects: Reads beamline config via _load_beamline_config. The shared loader reads YAML only on its first successful call (or after explicit cache reset).

Referenced in source (1)

_build_vision_target_tasks — src.beambot.mcp_server.beambot_mcp_server

`_build_vision_target_tasks()`

Build a configured scan-pose and vision approach without executing it. Offset mode inserts metre marker offsets; grid mode uses the same validated expansion as orchestrator pickup macros.

Internal helper · Source

_build_vision_target_tasks(
    target_name: str,
    element_index: int = 0,
    row: int = -1,
    col: int = -1,
    tag_id: int = -1,
) -> dict

Parameter	Meaning
`target_name`	Exact key in config.vision_targets.
`element_index`	Zero-based row-major grid index; ignored for offset mode or when both row and col are nonnegative.
`row`	Zero-based explicit row; used only together with a nonnegative col.
`col`	Zero-based explicit column; used only together with a nonnegative row.
`tag_id`	Nonnegative marker ID override; any negative value uses cfg.marker_id.

Returns: Dictionary containing task_json (a JSON string with start_gripper/tasks/poses), target and mode, or an error. Grid results add element {row,col,index} and offsets_mm by direction; near-zero row/column moves are omitted.

Errors: Unknown target/mode and ValueError from grid expansion return error dictionaries. Grid validation covers dimensions, indices, marker ID, finite offsets/pitches, directions and move definitions. Pose-file failures and offset-mode field/conversion errors can propagate; argument comparisons happen before grid validation. A scan pose can remain unresolved if absent from the registry.

Side effects: Reads configuration and poses; builds JSON only. No bridge initialization, detection, motion or action submission occurs.

Referenced in source (2)

pickup_tip — src.beambot.mcp_server.beambot_mcp_server
vision_target — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode`

Own ROS subscriptions, state caches, capture/cancel clients, and TF lookup for MCP tools. Cached values are observations, not freshness or system-health guarantees.

Inherits from: Node.

`ROS2BridgeNode.init()`

Create persistent Zivid/ZED image/cloud subscriptions, capture and cancel clients, a TF listener and robot-state caches. The ePick subscription is conditional on the optional epick_msgs import.

Lifecycle · Source

__init__(self)

Returns: None; all observation caches start empty and capture events cleared.

Errors: ROS/CvBridge/executor-related construction errors propagate.

Side effects: Registers beambot_mcp_bridge, ROS clients/subscriptions and TF listener; logs initialization. No capture or motion is requested.

Referenced in source (1)

ROS2Bridge._ensure_initialized — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode.cancel_all_goals()`

Send an empty CancelGoal request to /beambot_execution/_action/cancel_goal and summarize the number of goals being canceled. This is software action cancellation, not a hardware emergency stop.

Method · Source

cancel_all_goals(self) -> str

Returns: Human-readable status string for unavailable service, timeout, accepted cancellations, no active goals or no response.

Errors: Service discovery waits two seconds; future wait is five seconds. These failures become status strings; future exceptions propagate. The returned cancel return_code is not examined.

Side effects: Requests cancellation of all orchestrator goals; actual motion interruption depends on downstream action behavior.

`ROS2BridgeNode.trigger_capture()`

Trigger /capture, require its successful response and a newly signaled RGB image, and optionally require a newly signaled cloud. Capture timestamp is sampled just before the request; cached old data is retained.

Method · Source

trigger_capture(self, timeout: float=30.0, need_cloud: bool=True) -> bool

Parameter	Meaning
`timeout`	Seconds shared across response/image/cloud waits after discovery, with a minimum 0.1-second wait for each data event; service discovery can add three seconds.
`need_cloud`	True waits for both image and cloud; False waits only for image and does not change the capture service or clear old cloud data.

Returns: True after all requested events, False for service absence/failure or wait timeout.

Errors: Expected failures log and clear the waiting flag. Unexpected future exceptions or a None response can propagate without resetting the flag. No request-level serialization or timestamp matching is enforced.

Side effects: Triggers camera acquisition, records capture_stamp and blocks on events; updates waiting state.

`ROS2BridgeNode.lookup_transform()`

Check TF availability and fetch the transform at an explicit ROS stamp or latest time.

Method · Source

lookup_transform(
    self,
    target_frame: str,
    source_frame: str,
    stamp = None,
    timeout_sec: float = 2.0,
)

Parameter	Meaning
`target_frame`	Destination frame name.
`source_frame`	Source frame name.
`stamp`	ROS builtin time message; None requests latest transform.
`timeout_sec`	Seconds supplied independently to can_transform and lookup_transform, so total blocking can exceed one timeout.

Returns: TransformStamped mapping source_frame into target_frame, or None.

Errors: False availability and tf2 TransformException are logged and return None; invalid time/argument errors outside that exception class propagate.

Side effects: Queries the existing TF buffer and writes diagnostic logs.

Internal callbacks and helpers

`ROS2BridgeNode._on_zivid_image()`

Convert the incoming image to rgb8 and signal the image event only while a capture wait is active.

Callback · Source

_on_zivid_image(self, msg: Image)

Parameter	Meaning
`msg`	Zivid sensor_msgs/Image from /color/image_color.

Returns: None.

Errors: Conversion exceptions are logged and return early; last_rgb remains from an earlier successful frame.

Side effects: Updates last_rgb and may wake trigger_capture; does not retain the raw image message.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_zivid_cloud()`

Cache the most recent Zivid cloud and signal a capture waiter when active.

Callback · Source

_on_zivid_cloud(self, msg: PointCloud2)

Parameter	Meaning
`msg`	PointCloud2 from /points/xyzrgba.

Returns: None.

Errors: No explicit validation or exception handling.

Side effects: Replaces last_cloud and may set _cloud_event.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_zed_image()`

Store the streaming image stamp, then convert the image to RGB for subsequent tools.

Callback · Source

_on_zed_image(self, msg: Image)

Parameter	Meaning
`msg`	ZED sensor_msgs/Image with header.stamp.

Returns: None.

Errors: Conversion exceptions are logged; zed_stamp still advances while zed_rgb can remain stale.

Side effects: Updates zed_stamp and zed_rgb; does not retain the raw image message or synchronize image and cloud timestamps.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_zed_cloud()`

Keep the latest ZED registered cloud for image/depth tools.

Callback · Source

_on_zed_cloud(self, msg: PointCloud2)

Parameter	Meaning
`msg`	PointCloud2 from the configured ZED cloud topic.

Returns: None.

Errors: No explicit validation or error handling.

Side effects: Replaces zed_cloud.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_joint_states()`

Copy incoming joint names and positions into the robot-state cache without filtering for arm joints.

Callback · Source

_on_joint_states(self, msg: JointState)

Parameter	Meaning
`msg`	JointState; position values remain in ROS joint units, normally radians for revolute joints.

Returns: None.

Errors: Malformed-message errors propagate; unequal arrays are not rejected.

Side effects: Replaces joint_names and joint_positions.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_current_gripper()`

Cache the current tool/gripper name published by Beambot.

Callback · Source

_on_current_gripper(self, msg: String)

Parameter	Meaning
`msg`	std_msgs/String whose data is the active gripper identifier.

Returns: None.

Side effects: Updates current_gripper.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_execution_state()`

Cache the orchestrator execution-state string verbatim.

Callback · Source

_on_execution_state(self, msg: String)

Parameter	Meaning
`msg`	std_msgs/String from /beambot/execution_state; the bridge does not normalize the state vocabulary.

Returns: None.

Side effects: Updates execution_state.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2BridgeNode._on_epick_status()`

Convert and store the raw ePick detection-status integer.

Callback · Source

_on_epick_status(self, msg)

Parameter	Meaning
`msg`	ObjectDetectionStatus message; status is converted with int().

Returns: None.

Errors: Integer conversion and malformed-message errors propagate.

Side effects: Updates epick_status.

Referenced in source (1)

ROS2BridgeNode.__init__ — src.beambot.mcp_server.beambot_mcp_server

`ROS2Bridge`

Manage lazy ROS initialization and the background executor. Access the node property to obtain the shared bridge node; shutdown() releases its ROS resources.

`ROS2Bridge.init()`

Initialize a dormant bridge with empty node/executor/thread references and a lock for first-use initialization.

Lifecycle · Source

__init__(self)

Errors: Lock-allocation errors propagate.

Side effects: Allocates in-memory state only; does not initialize ROS.

`ROS2Bridge.node`

Return the shared ROS bridge node, initializing it and its background executor on first access.

Method · Source

@property
node(self) -> ROS2BridgeNode

Access this as an instance property, without parentheses.

Returns: ROS2BridgeNode instance.

Errors: Propagates first-use initialization failures.

Side effects: First access can initialize ROS and launch executor threads.

`ROS2Bridge.shutdown()`

Shut down the executor, destroy the node and shut down the ROS context if still active.

Lifecycle · Source

shutdown(self)

Returns: None.

Errors: Cleanup exceptions propagate and can prevent later steps. It does not reset _initialized, clear references or explicitly join the spin thread.

Side effects: Stops bridge ROS processing; can shut down a ROS context that existed before this bridge.

Internal callbacks and helpers

`ROS2Bridge._ensure_initialized()`

Use double-checked locking to initialize ROS if necessary, construct the bridge node and a four-thread executor, then start the daemon ros2-bridge spin thread once.

Internal helper · Source

_ensure_initialized(self)

Returns: None; later calls return immediately when _initialized is true.

Errors: Initialization errors propagate and partial resources are not rolled back here.

Side effects: May initialize the global ROS context and start subscriptions/executor threads.

Referenced in source (1)

ROS2Bridge.node — src.beambot.mcp_server.beambot_mcp_server

`ROS2Bridge._spin()`

Run the stored executor’s spin loop in the bridge thread.

Internal helper · Source

_spin(self)

Returns: None when spin exits or raises a caught Exception.

Errors: Logs any Exception without rethrowing or resetting _initialized.

Side effects: Processes ROS callbacks continuously in the background.

Referenced in source (1)

ROS2Bridge._ensure_initialized — src.beambot.mcp_server.beambot_mcp_server

MCP robot tools

How it is used

Start here

Example

Functions

main()

get_robot_state()

capture_image()

detect_objects()

vision_target()

ping()

stop_robot()

get_vacuum_status()

get_saved_poses()

save_pose()

delete_pose()

set_cup_profile()

detect_sample_yolo()

get_point_3d()

get_tf_transform()

get_recent_logs()

pickup_tip()

Internal callbacks and helpers

_poses_file_path()

_detect_display()

_detect_hsv_color()

_detect_aruco_markers()

_annotate_image()

_resolve_camera_state()

_read_poses_file()

_write_poses_file()

capture_image._wait_for_zed()

_transform_point_to_base()

_transform_point_to_base._do_transform()

_confirm_point_via_gui()

_get_yolo_model()

get_tf_transform._lookup()

get_recent_logs._read_tail()

_load_beamline_config()

_load_vision_targets()

_build_vision_target_tasks()

ROS2BridgeNode

ROS2BridgeNode.__init__()

ROS2BridgeNode.cancel_all_goals()

ROS2BridgeNode.trigger_capture()

ROS2BridgeNode.lookup_transform()

Internal callbacks and helpers

ROS2BridgeNode._on_zivid_image()

ROS2BridgeNode._on_zivid_cloud()

ROS2BridgeNode._on_zed_image()

ROS2BridgeNode._on_zed_cloud()

ROS2BridgeNode._on_joint_states()

ROS2BridgeNode._on_current_gripper()

ROS2BridgeNode._on_execution_state()

ROS2BridgeNode._on_epick_status()

ROS2Bridge

ROS2Bridge.__init__()

ROS2Bridge.node

ROS2Bridge.shutdown()

Internal callbacks and helpers

ROS2Bridge._ensure_initialized()

ROS2Bridge._spin()

`main()`

`get_robot_state()`

`capture_image()`

`detect_objects()`

`vision_target()`

`ping()`

`stop_robot()`

`get_vacuum_status()`

`get_saved_poses()`

`save_pose()`

`delete_pose()`

`set_cup_profile()`

`detect_sample_yolo()`

`get_point_3d()`

`get_tf_transform()`

`get_recent_logs()`

`pickup_tip()`

`_poses_file_path()`

`_detect_display()`

`_detect_hsv_color()`

`_detect_aruco_markers()`

`_annotate_image()`

`_resolve_camera_state()`

`_read_poses_file()`

`_write_poses_file()`

`capture_image._wait_for_zed()`

`_transform_point_to_base()`

`_transform_point_to_base._do_transform()`

`_confirm_point_via_gui()`

`_get_yolo_model()`

`get_tf_transform._lookup()`

`get_recent_logs._read_tail()`

`_load_beamline_config()`

`_load_vision_targets()`

`_build_vision_target_tasks()`

`ROS2BridgeNode`

`ROS2BridgeNode.init()`

`ROS2BridgeNode.cancel_all_goals()`

`ROS2BridgeNode.trigger_capture()`

`ROS2BridgeNode.lookup_transform()`

`ROS2BridgeNode._on_zivid_image()`

`ROS2BridgeNode._on_zivid_cloud()`

`ROS2BridgeNode._on_zed_image()`

`ROS2BridgeNode._on_zed_cloud()`

`ROS2BridgeNode._on_joint_states()`

`ROS2BridgeNode._on_current_gripper()`

`ROS2BridgeNode._on_execution_state()`

`ROS2BridgeNode._on_epick_status()`

`ROS2Bridge`

`ROS2Bridge.init()`

`ROS2Bridge.node`

`ROS2Bridge.shutdown()`

`ROS2Bridge._ensure_initialized()`

`ROS2Bridge._spin()`