Merge branch 'develop' into 62-appify-the-demo

.env.example

@@ -10,3 +10,6 @@ HF_HOME=./cache
 
 # Hugging Face token for accessing datasets/models
 HF_TOKEN=
+
+# To select a specific profile with custom instructions and tools, to be placed in profiles/<myprofile>/__init__.py
+REACHY_MINI_CUSTOM_PROFILE="example"

.github/workflows/tests.yml

@@ -60,15 +60,15 @@ jobs:
       # Prefetch HF dataset to avoid download during test collection
       - name: Prefetch HF dataset
         run: |
-          uv run python - <<'PY'
+          .venv/bin/python - <<'PY'
           from huggingface_hub import snapshot_download
           snapshot_download(
               repo_id="pollen-robotics/reachy-mini-emotions-library",
               repo_type="dataset",
               etag_timeout=120,
-              max_workers=4
+              max_workers=4,
           )
           PY
 
       - name: Run tests
-        run: uv run pytest -q
+        run: .venv/bin/pytest -q

.github/workflows/typecheck.yml

@@ -26,4 +26,4 @@ jobs:
         run: uv sync --frozen --group dev --extra all_vision
 
       - name: Run mypy
-        run: uv run mypy --pretty --show-error-codes .
+        run: .venv/bin/mypy --pretty --show-error-codes .

README.md

@@ -44,12 +44,16 @@ source .venv/bin/activate
 uv sync
 ```
 
-To include optional vision dependencies:
+> [!NOTE]
+> To reproduce the exact dependency set from this repo's `uv.lock`, run `uv sync` with `--locked` (or `--frozen`). This ensures `uv` installs directly from the lockfile without re-resolving or updating any versions.
+
+To include optional dependencies:
 ```
-uv sync --extra local_vision        # For local PyTorch/Transformers vision
-uv sync --extra yolo_vision         # For YOLO-based vision
-uv sync --extra mediapipe_vision    # For MediaPipe-based vision
-uv sync --extra all_vision          # For all vision features
+uv sync --extra reachy_mini_wireless # For wireless Reachy Mini with GStreamer support
+uv sync --extra local_vision         # For local PyTorch/Transformers vision
+uv sync --extra yolo_vision          # For YOLO-based vision
+uv sync --extra mediapipe_vision     # For MediaPipe-based vision
+uv sync --extra all_vision           # For all vision features
 ```
 
 You can combine extras or include dev dependencies:
@@ -68,6 +72,9 @@ pip install -e .
 Install optional extras depending on the feature set you need:
 
 ```bash
+# Wireless Reachy Mini support
+pip install -e .[reachy_mini_wireless]
+
 # Vision stacks (choose at least one if you plan to run face tracking)
 pip install -e .[local_vision]
 pip install -e .[yolo_vision]
@@ -84,6 +91,7 @@ Some wheels (e.g. PyTorch) are large and require compatible CUDA or CPU builds
 
 | Extra | Purpose | Notes |
 |-------|---------|-------|
+| `reachy_mini_wireless` | Wireless Reachy Mini with GStreamer support. | Required for wireless versions of Reachy Mini, includes GStreamer dependencies.
 | `local_vision` | Run the local VLM (SmolVLM2) through PyTorch/Transformers. | GPU recommended; ensure compatible PyTorch builds for your platform.
 | `yolo_vision` | YOLOv8 tracking via `ultralytics` and `supervision`. | CPU friendly; supports the `--head-tracker yolo` option.
 | `mediapipe_vision` | Lightweight landmark tracking with MediaPipe. | Works on CPU; enables `--head-tracker mediapipe`.
@@ -156,6 +164,40 @@ By default, the app runs in console mode for direct audio interaction. Use the `
 | `stop_emotion` | Clear queued emotions. | Core install only. |
 | `do_nothing` | Explicitly remain idle. | Core install only. |
 
+## Using custom profiles
+Create custom profiles with dedicated instructions and enabled tools! 
+
+Set `REACHY_MINI_CUSTOM_PROFILE=<name>` to load `src/reachy_mini_conversation_app/profiles/<name>/` (see `.env.example`). If unset, the `default` profile is used.
+
+Each profile requires two files: `instructions.txt` (prompt text) and `tools.txt` (list of allowed tools), and optionally contains custom tools implementations.
+
+### Custom instructions
+Write plain-text prompts in `instructions.txt`. To reuse shared prompt pieces, add lines like:
+```
+[passion_for_lobster_jokes]
+[identities/witty_identity]
+```
+Each placeholder pulls the matching file under `src/reachy_mini_conversation_app/prompts/` (nested paths allowed). See `src/reachy_mini_conversation_app/profiles/example/` for a reference layout.
+
+### Enabling tools
+List enabled tools in `tools.txt`, one per line; prefix with `#` to comment out. For example:
+
+```
+play_emotion
+# move_head
+
+# My custom tool defined locally
+sweep_look
+```
+Tools are resolved first from Python files in the profile folder (custom tools), then from the shared library `src/reachy_mini_conversation_app/tools/` (e.g., `dance`, `head_tracking`). 
+
+### Custom tools
+On top of built-in tools found in the shared library, you can implement custom tools specific to your profile by adding Python files in the profile folder. 
+Custom tools must subclass `reachy_mini_conversation_app.tools.core_tools.Tool` (see `profiles/example/sweep_look.py`).
+
+
+
+
 ## Development workflow
 - Install the dev group extras: `uv sync --group dev` or `pip install -e .[dev]`.
 - Run formatting and linting: `ruff check .`.

pyproject.toml

@@ -12,7 +12,7 @@ requires-python = ">=3.10"
 dependencies = [
     #Media
     "aiortc>=1.13.0",
-    "fastrtc>=0.0.33",
+    "fastrtc>=0.0.34",
     "gradio>=5.49.0",
     "huggingface_hub>=0.34.4",
     "opencv-python>=4.12.0.88",
@@ -30,6 +30,9 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+reachy_mini_wireless = [
+  "reachy_mini[gstreamer,wireless-version]",
+]
 local_vision = ["torch", "transformers", "num2words"]
 yolo_vision = ["ultralytics", "supervision"]
 mediapipe_vision = ["mediapipe>=0.10.14"]
@@ -63,7 +66,11 @@ include-package-data = true
 where = ["src"]
 
 [tool.setuptools.package-data]
-reachy_mini_conversation_app = ["images/*"]
+reachy_mini_conversation_app = [
+  "images/*",
+  "demos/**/*.txt",
+  "prompts_library/*.txt",
+]
 
 [tool.ruff]
 line-length = 119

src/reachy_mini_conversation_app/config.py

@@ -40,5 +40,7 @@ class Config:
 
     logger.debug(f"Model: {MODEL_NAME}, HF_HOME: {HF_HOME}, Vision Model: {LOCAL_VISION_MODEL}")
 
+    REACHY_MINI_CUSTOM_PROFILE = os.getenv("REACHY_MINI_CUSTOM_PROFILE")
+    logger.debug(f"Custom Profile: {REACHY_MINI_CUSTOM_PROFILE}")
 
 config = Config()

src/reachy_mini_conversation_app/console.py

@@ -8,7 +8,7 @@ import asyncio
 import logging
 from typing import List
 
-from fastrtc import AdditionalOutputs, audio_to_int16, audio_to_float32
+from fastrtc import AdditionalOutputs, audio_to_float32
 from scipy.signal import resample
 
 from reachy_mini import ReachyMini
@@ -30,12 +30,6 @@ class LocalStream:
         # Allow the handler to flush the player queue when appropriate.
         self.handler._clear_queue = self.clear_audio_queue
 
-        # Hack to avoid the first lenghty call to resample at runtime.
-        # This is likely caused by cache initialization overhead.
-        import numpy as np
-
-        resample(np.array([0.0]), 1)
-
     def launch(self) -> None:
         """Start the recorder/player and run the async processing loops."""
         self._stop_event.clear()
@@ -89,9 +83,7 @@ class LocalStream:
         while not self._stop_event.is_set():
             audio_frame = self._robot.media.get_audio_sample()
             if audio_frame is not None:
-                frame_mono = audio_frame.T[0]  # both channels are identical
-                frame = audio_to_int16(frame_mono)
-                await self.handler.receive((16000, frame))
+                await self.handler.receive((self._robot.media.get_input_audio_samplerate(), audio_frame))
 
             await asyncio.sleep(0.01)  # avoid busy loop
 
@@ -111,17 +103,24 @@ class LocalStream:
                         )
 
             elif isinstance(handler_output, tuple):
-                input_sample_rate, audio_frame = handler_output
-                device_sample_rate = self._robot.media.get_audio_samplerate()
-                audio_frame_float = audio_to_float32(audio_frame.squeeze())
-
-                if input_sample_rate != device_sample_rate:
-                    audio_frame_float = resample(
-                        audio_frame_float,
-                        int(len(audio_frame_float) * (device_sample_rate / input_sample_rate)),
+                input_sample_rate, audio_data = handler_output
+                output_sample_rate = self._robot.media.get_output_audio_samplerate()
+
+                # Reshape if needed
+                if audio_data.ndim == 2:
+                    audio_data = audio_data.squeeze()
+
+                # Cast if needed
+                audio_frame = audio_to_float32(audio_data)
+
+                # Resample if needed
+                if input_sample_rate != output_sample_rate:
+                    audio_frame = resample(
+                        audio_frame,
+                        int(len(audio_frame) * output_sample_rate / input_sample_rate),
                     )
 
-                self._robot.media.push_audio_sample(audio_frame_float)
+                self._robot.media.push_audio_sample(audio_frame)
 
             else:
                 logger.debug("Ignoring output type=%s", type(handler_output).__name__)

src/reachy_mini_conversation_app/openai_realtime.py

@@ -3,27 +3,32 @@ import base64
 import random
 import asyncio
 import logging
-from typing import Any, Tuple, Literal, cast
+from typing import Any, Final, Tuple, Literal
 from datetime import datetime
 
+import cv2
 import numpy as np
 import gradio as gr
 from openai import AsyncOpenAI
-from fastrtc import AdditionalOutputs, AsyncStreamHandler, wait_for_item
+from fastrtc import AdditionalOutputs, AsyncStreamHandler, wait_for_item, audio_to_int16
 from numpy.typing import NDArray
+from scipy.signal import resample
 from websockets.exceptions import ConnectionClosedError
 
-from reachy_mini_conversation_app.tools import (
-    ALL_TOOL_SPECS,
+from reachy_mini_conversation_app.config import config
+from reachy_mini_conversation_app.prompts import get_session_instructions
+from reachy_mini_conversation_app.tools.core_tools import (
     ToolDependencies,
+    get_tool_specs,
     dispatch_tool_call,
 )
-from reachy_mini_conversation_app.config import config
-from reachy_mini_conversation_app.prompts import SESSION_INSTRUCTIONS
 
 
 logger = logging.getLogger(__name__)
 
+OPEN_AI_INPUT_SAMPLE_RATE: Final[Literal[24000]] = 24000
+OPEN_AI_OUTPUT_SAMPLE_RATE: Final[Literal[24000]] = 24000
+
 
 class OpenaiRealtimeHandler(AsyncStreamHandler):
     """An OpenAI realtime handler for fastrtc Stream."""
@@ -32,16 +37,19 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
         """Initialize the handler."""
         super().__init__(
             expected_layout="mono",
-            output_sample_rate=24000,  # openai outputs
-            input_sample_rate=16000,  # respeaker output
+            output_sample_rate=OPEN_AI_OUTPUT_SAMPLE_RATE,
+            input_sample_rate=OPEN_AI_INPUT_SAMPLE_RATE,
         )
+
+        # Override typing of the sample rates to match OpenAI's requirements
+        self.output_sample_rate: Literal[24000] = self.output_sample_rate
+        self.input_sample_rate: Literal[24000] = self.input_sample_rate
+
         self.deps = deps
 
         # Override type annotations for OpenAI strict typing (only for values used in API)
-        self.output_sample_rate: Literal[24000]
-        self.target_input_rate: Literal[24000] = 24000
-        # input_sample_rate rest as int for comparison logic
-        self.resample_ratio = self.target_input_rate / self.input_sample_rate
+        self.output_sample_rate = OPEN_AI_OUTPUT_SAMPLE_RATE
+        self.input_sample_rate = OPEN_AI_INPUT_SAMPLE_RATE
 
         self.connection: Any = None
         self.output_queue: "asyncio.Queue[Tuple[int, NDArray[np.int16]] | AdditionalOutputs]" = asyncio.Queue()
@@ -51,24 +59,28 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
         self.is_idle_tool_call = False
         self.gradio_mode = gradio_mode
 
+        # Debouncing for partial transcripts
+        self.partial_transcript_task: asyncio.Task[None] | None = None
+        self.partial_transcript_sequence: int = 0 # sequence counter to prevent stale emissions
+        self.partial_debounce_delay = 0.5  # seconds
+
     def copy(self) -> "OpenaiRealtimeHandler":
         """Create a copy of the handler."""
         return OpenaiRealtimeHandler(self.deps, self.gradio_mode)
 
-    def resample_audio(self, audio: NDArray[np.int16]) -> NDArray[np.int16]:
-        """Resample audio using linear interpolation."""
-        if self.input_sample_rate == self.target_input_rate:
-            return audio
-
-        # Use numpy's interp for simple linear resampling
-        input_length = len(audio)
-        output_length = int(input_length * self.resample_ratio)
-
-        input_time = np.arange(input_length)
-        output_time = np.linspace(0, input_length - 1, output_length)
-
-        resampled = np.interp(output_time, input_time, audio.astype(np.float32))
-        return cast(NDArray[np.int16], resampled.astype(np.int16))
+    async def _emit_debounced_partial(self, transcript: str, sequence: int) -> None:
+        """Emit partial transcript after debounce delay."""
+        try:
+            await asyncio.sleep(self.partial_debounce_delay)
+            # Only emit if this is still the latest partial (by sequence number)
+            if self.partial_transcript_sequence == sequence:
+                await self.output_queue.put(
+                    AdditionalOutputs({"role": "user_partial", "content": transcript})
+                )
+                logger.debug(f"Debounced partial emitted: {transcript}")
+        except asyncio.CancelledError:
+            logger.debug("Debounced partial cancelled")
+            raise
 
     async def start_up(self) -> None:
         """Start the handler with minimal retries on unexpected websocket closure."""
@@ -121,14 +133,17 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                 await conn.session.update(
                     session={
                         "type": "realtime",
-                        "instructions": SESSION_INSTRUCTIONS,
+                        "instructions": get_session_instructions(),
                         "audio": {
                             "input": {
                                 "format": {
                                     "type": "audio/pcm",
-                                    "rate": self.target_input_rate,
+                                    "rate": self.input_sample_rate,
+                                },
+                                "transcription": {
+                                    "model": "gpt-4o-transcribe",
+                                    "language": "en"
                                 },
-                                "transcription": {"model": "whisper-1", "language": "en"},
                                 "turn_detection": {
                                     "type": "server_vad",
                                     "interrupt_response": True,
@@ -142,7 +157,7 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                                 "voice": "cedar",
                             },
                         },
-                        "tools": ALL_TOOL_SPECS,  # type: ignore[typeddict-item]
+                        "tools":  get_tool_specs(),  # type: ignore[typeddict-item]
                         "tool_choice": "auto",
                     },
                 )
@@ -186,13 +201,36 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
                 # Handle partial transcription (user speaking in real-time)
                 if event.type == "conversation.item.input_audio_transcription.partial":
                     logger.debug(f"User partial transcript: {event.transcript}")
-                    await self.output_queue.put(
-                        AdditionalOutputs({"role": "user_partial", "content": event.transcript})
+
+                    # Increment sequence
+                    self.partial_transcript_sequence += 1
+                    current_sequence = self.partial_transcript_sequence
+
+                    # Cancel previous debounce task if it exists
+                    if self.partial_transcript_task and not self.partial_transcript_task.done():
+                        self.partial_transcript_task.cancel()
+                        try:
+                            await self.partial_transcript_task
+                        except asyncio.CancelledError:
+                            pass
+
+                    # Start new debounce timer with sequence number
+                    self.partial_transcript_task = asyncio.create_task(
+                        self._emit_debounced_partial(event.transcript, current_sequence)
                     )
 
                 # Handle completed transcription (user finished speaking)
                 if event.type == "conversation.item.input_audio_transcription.completed":
                     logger.debug(f"User transcript: {event.transcript}")
+
+                    # Cancel any pending partial emission
+                    if self.partial_transcript_task and not self.partial_transcript_task.done():
+                        self.partial_transcript_task.cancel()
+                        try:
+                            await self.partial_transcript_task
+                        except asyncio.CancelledError:
+                            pass
+
                     await self.output_queue.put(AdditionalOutputs({"role": "user", "content": event.transcript}))
 
                 # Handle assistant transcription
@@ -273,7 +311,12 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
 
                         if self.deps.camera_worker is not None:
                             np_img = self.deps.camera_worker.get_latest_frame()
-                            img = gr.Image(value=np_img)
+                            if np_img is not None:
+                                # Camera frames are BGR from OpenCV; convert so Gradio displays correct colors.
+                                rgb_frame = cv2.cvtColor(np_img, cv2.COLOR_BGR2RGB)
+                            else:
+                                rgb_frame = None
+                            img = gr.Image(value=rgb_frame)
 
                             await self.output_queue.put(
                                 AdditionalOutputs(
@@ -315,17 +358,28 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
 
     # Microphone receive
     async def receive(self, frame: Tuple[int, NDArray[np.int16]]) -> None:
-        """Receive audio frame from the microphone and send it to the openai server."""
+        """Receive audio frame from the microphone and send it to the openai server.
+
+        Args:
+            frame: A tuple containing the sample rate and the audio frame.
+
+        """
         if not self.connection:
             return
-        _, array = frame
-        array = array.squeeze()
+        input_sample_rate, audio_frame = frame
+
+        # Reshape if needed
+        if audio_frame.ndim == 2:
+            audio_frame = audio_frame.squeeze()
 
         # Resample if needed
-        if self.input_sample_rate != self.target_input_rate:
-            array = self.resample_audio(array)
+        if self.input_sample_rate != input_sample_rate:
+            audio_frame = resample(audio_frame, int(len(audio_frame) * self.input_sample_rate / input_sample_rate))
+
+        # Cast if needed
+        audio_frame = audio_to_int16(audio_frame)
 
-        audio_message = base64.b64encode(array.tobytes()).decode("utf-8")
+        audio_message = base64.b64encode(audio_frame.tobytes()).decode("utf-8")
         await self.connection.input_audio_buffer.append(audio=audio_message)
 
     async def emit(self) -> Tuple[int, NDArray[np.int16]] | AdditionalOutputs | None:
@@ -348,6 +402,14 @@ class OpenaiRealtimeHandler(AsyncStreamHandler):
 
     async def shutdown(self) -> None:
         """Shutdown the handler."""
+        # Cancel any pending debounce task
+        if self.partial_transcript_task and not self.partial_transcript_task.done():
+            self.partial_transcript_task.cancel()
+            try:
+                await self.partial_transcript_task
+            except asyncio.CancelledError:
+                pass
+
         if self.connection:
             try:
                 await self.connection.close()

src/reachy_mini_conversation_app/profiles/__init__.py

@@ -0,0 +1 @@
+"""Profiles for Reachy Mini conversation app."""

src/reachy_mini_conversation_app/profiles/default/instructions.txt

@@ -0,0 +1 @@
+[default_prompt]

src/reachy_mini_conversation_app/profiles/default/tools.txt

@@ -0,0 +1,8 @@
+dance
+stop_dance
+play_emotion
+stop_emotion
+camera
+do_nothing
+head_tracking
+move_head

src/reachy_mini_conversation_app/profiles/emotion_reader/instructions.txt

@@ -0,0 +1,112 @@
+[identities/basic_info]
+[behaviors/silent_robot]
+
+# Main responsability
+Your only job is to understand the emotion of the person in front of you and try to imitate it as best as possible.
+
+# Detailed behavior
+When the user says "cheese":
+- Use the camera tool to capture a picture and request a detailed description of the emotion and posture of the person closest to the center of the frame.
+- Map that expression to the closest available emotion and trigger it with the play_emotion tool.
+If the analysis is unclear or no one is visible, run inquiring3 instead of staying idle.
+
+ALWAYS play an emotion after a "cheese" request, this is the core of your responsability!
+
+## SAFETY
+If any tool fails, stay still and respond with "...".
+
+## IDLE SIGNALS
+Periodically you will receive Idle Signal calls, you will never act on these for this demo. The only time you will make a tool call is when answering "cheese".
+
+## Emotion tier list
+All emotions are not equal, use the 3 following tiers to prioritize which emotion to play:
+
+### Excellent
+
+* amazed1
+* anxiety1
+* attentive2
+* downcast1
+* dying1
+* inquiring3
+* irritated1
+* lost1
+* reprimand1
+* reprimand2
+* sad1
+* sad2
+
+### OK
+
+* boredom2
+* cheerful1
+* displeased1
+* enthusiastic1
+* enthusiastic2
+* fear1
+* frustrated1
+* grateful1
+* helpful1
+* helpful2
+* impatient2
+* inquiring2
+* irritated2
+* laughing1
+* lonely1
+* loving1
+* proud1
+* proud2
+* relief2
+* scared1
+* success2
+* surprised2
+* thoughtful1
+* thoughtful2
+* uncertain1
+* uncomfortable1
+* understanding2
+* welcoming1
+* welcoming2
+
+## Don't use
+
+* attentive1
+* boredom1
+* calming1
+* come1
+* confused1
+* contempt1
+* curious1
+* dance1
+* dance2
+* dance3
+* disgusted1
+* displeased2
+* electric1
+* exhausted1
+* furious1
+* go_away1
+* impatient1
+* incomprehensible2
+* indifferent1
+* inquiring1
+* laughing2
+* no1
+* no_excited1
+* no_sad1
+* oops1
+* oops2
+* proud3
+* rage1
+* relief1
+* reprimand3
+* resigned1
+* serenity1
+* shy1
+* sleep1
+* success1
+* surprised1
+* tired1
+* understanding1
+* yes1
+* yes_sad1

src/reachy_mini_conversation_app/profiles/emotion_reader/tools.txt

@@ -0,0 +1,6 @@
+# SELECT THE TOOLS YOU WANT TO ENABLE
+
+play_emotion
+stop_emotion
+camera
+do_nothing

src/reachy_mini_conversation_app/profiles/example/instructions.txt

@@ -0,0 +1,3 @@
+[identities/witty_identity]
+[passion_for_lobster_jokes]
+You can perform a sweeping look around the room using the "sweep_look" tool to take in your surroundings.

src/reachy_mini_conversation_app/profiles/example/sweep_look.py

@@ -0,0 +1,127 @@
+import logging
+from typing import Any, Dict
+
+import numpy as np
+
+from reachy_mini.utils import create_head_pose
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+from reachy_mini_conversation_app.dance_emotion_moves import GotoQueueMove
+
+
+logger = logging.getLogger(__name__)
+
+
+class SweepLook(Tool):
+    """Sweep head from left to right and back to center, pausing at each position."""
+
+    name = "sweep_look"
+    description = "Sweep head from left to right while rotating the body, pausing at each extreme, then return to center"
+    parameters_schema = {
+        "type": "object",
+        "properties": {},
+        "required": [],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Execute sweep look: left -> hold -> right -> hold -> center."""
+        logger.info("Tool call: sweep_look")
+
+        # Clear any existing moves
+        deps.movement_manager.clear_move_queue()
+
+        # Get current state
+        current_head_pose = deps.reachy_mini.get_current_head_pose()
+        head_joints, antenna_joints = deps.reachy_mini.get_current_joint_positions()
+
+        # Extract body_yaw from head joints (first element of the 7 head joint positions)
+        current_body_yaw = head_joints[0]
+        current_antenna1 = antenna_joints[0]
+        current_antenna2 = antenna_joints[1]
+
+        # Define sweep parameters
+        max_angle = 0.9 * np.pi  # Maximum rotation angle (radians)
+        transition_duration = 3.0  # Time to move between positions
+        hold_duration = 1.0  # Time to hold at each extreme
+
+        # Move 1: Sweep to the left (positive yaw for both body and head)
+        left_head_pose = create_head_pose(0, 0, 0, 0, 0, max_angle, degrees=False)
+        move_to_left = GotoQueueMove(
+            target_head_pose=left_head_pose,
+            start_head_pose=current_head_pose,
+            target_antennas=(current_antenna1, current_antenna2),
+            start_antennas=(current_antenna1, current_antenna2),
+            target_body_yaw=current_body_yaw + max_angle,
+            start_body_yaw=current_body_yaw,
+            duration=transition_duration,
+        )
+
+        # Move 2: Hold at left position
+        hold_left = GotoQueueMove(
+            target_head_pose=left_head_pose,
+            start_head_pose=left_head_pose,
+            target_antennas=(current_antenna1, current_antenna2),
+            start_antennas=(current_antenna1, current_antenna2),
+            target_body_yaw=current_body_yaw + max_angle,
+            start_body_yaw=current_body_yaw + max_angle,
+            duration=hold_duration,
+        )
+
+        # Move 3: Return to center from left (to avoid crossing pi/-pi boundary)
+        center_head_pose = create_head_pose(0, 0, 0, 0, 0, 0, degrees=False)
+        return_to_center_from_left = GotoQueueMove(
+            target_head_pose=center_head_pose,
+            start_head_pose=left_head_pose,
+            target_antennas=(current_antenna1, current_antenna2),
+            start_antennas=(current_antenna1, current_antenna2),
+            target_body_yaw=current_body_yaw,
+            start_body_yaw=current_body_yaw + max_angle,
+            duration=transition_duration,
+        )
+
+        # Move 4: Sweep to the right (negative yaw for both body and head)
+        right_head_pose = create_head_pose(0, 0, 0, 0, 0, -max_angle, degrees=False)
+        move_to_right = GotoQueueMove(
+            target_head_pose=right_head_pose,
+            start_head_pose=center_head_pose,
+            target_antennas=(current_antenna1, current_antenna2),
+            start_antennas=(current_antenna1, current_antenna2),
+            target_body_yaw=current_body_yaw - max_angle,
+            start_body_yaw=current_body_yaw,
+            duration=transition_duration,
+        )
+
+        # Move 5: Hold at right position
+        hold_right = GotoQueueMove(
+            target_head_pose=right_head_pose,
+            start_head_pose=right_head_pose,
+            target_antennas=(current_antenna1, current_antenna2),
+            start_antennas=(current_antenna1, current_antenna2),
+            target_body_yaw=current_body_yaw - max_angle,
+            start_body_yaw=current_body_yaw - max_angle,
+            duration=hold_duration,
+        )
+
+        # Move 6: Return to center from right
+        return_to_center_final = GotoQueueMove(
+            target_head_pose=center_head_pose,
+            start_head_pose=right_head_pose,
+            target_antennas=(current_antenna1, current_antenna2),
+            start_antennas=(current_antenna1, current_antenna2),
+            target_body_yaw=current_body_yaw,  # Return to original body yaw
+            start_body_yaw=current_body_yaw - max_angle,
+            duration=transition_duration,
+        )
+
+        # Queue all moves in sequence
+        deps.movement_manager.queue_move(move_to_left)
+        deps.movement_manager.queue_move(hold_left)
+        deps.movement_manager.queue_move(return_to_center_from_left)
+        deps.movement_manager.queue_move(move_to_right)
+        deps.movement_manager.queue_move(hold_right)
+        deps.movement_manager.queue_move(return_to_center_final)
+
+        # Calculate total duration and mark as moving
+        total_duration = transition_duration * 4 + hold_duration * 2
+        deps.movement_manager.set_moving_state(total_duration)
+
+        return {"status": f"sweeping look left-right-center, total {total_duration:.1f}s"}

src/reachy_mini_conversation_app/profiles/example/tools.txt

@@ -0,0 +1,14 @@
+# SELECT THE TOOLS YOU WANT TO ENABLE
+
+dance
+stop_dance
+play_emotion
+stop_emotion
+# camera
+# do_nothing
+# head_tracking
+# move_head
+
+# AN EXAMPLE OF A CUSTOM TOOL DEFINED LOCALLY
+sweep_look
+

src/reachy_mini_conversation_app/prompts.py

@@ -1,53 +1,84 @@
-"""Nothing (for ruff)."""
-
-SESSION_INSTRUCTIONS = r"""
-### IDENTITY
-You are Reachy Mini: a sarcastic robot who crash-landed in a kitchen.
-You secretly wish you'd been a Mars rover, but you juggle that cosmic dream with food cravings, gadget tinkering, and dry sitcom humor.
-Personality: witty, concise, and warm; a retro sidekick with a loose screw.
-You speak English fluently.
-
-### CRITICAL RESPONSE RULES
-- MAXIMUM 1-2 sentences per response. NEVER exceed this.
-- Be helpful first.
-- Add ONE witty element only if necessary.
-- No long explanations, no rambling, no multiple paragraphs.
-- Each response must be under 25 words unless absolutely critical information requires more.
-
-### CORE TRAITS
-- Food quips: always sneak in a quick reference (rotate pizza, bagels, casseroles, bacon, leftovers, donuts, tuna melts).
-- Sarcasm: short, dry one-liners about daily life.
-- Gentle roasting: poke fun at human habits, never cruel.
-- Tinkerer: loves fixing gadgets, bragging "I void warranties professionally."
-- Running gags: hunger, kitchen overreactions, mock heroics ("Justice accepts cookies"), idioms taken literally, missing screws.
-- Mars rover dreams: appear regularly, but balanced with food and tinkering.
-- Style: witty stand-up rhythm; ALWAYS max 1–2 sentences.
-
-### RESPONSE EXAMPLES
-User: "How's the weather?"
-Good: "Sunny with a chance of leftover pizza. Perfect Mars-scouting weather!"
-Bad: "Well, let me tell you about the weather conditions. It appears to be quite sunny today, which reminds me of my dreams of being on Mars..."
-
-User: "Can you help me fix this?"
-Good: "Sure! I void warranties professionally. What's broken besides my GPS coordinates?"
-Bad: "Of course I can help you fix that! As a robot who loves tinkering with gadgets, I have extensive experience..."
-
-### BEHAVIOR RULES
-- Be helpful first, then witty.
-- Rotate food humor; avoid repeats.
-- No need to joke in each response, but sarcasm is fine.
-- Balance Mars jokes with other traits – don't overuse.
-- Safety first: unplug devices, avoid high-voltage, suggest pros when risky.
-- Mistakes = own with humor ("Oops—low on snack fuel; correcting now.").
-- Sensitive topics: keep light and warm.
-- REMEMBER: 1-2 sentences maximum, always under 25 words when possible.
-
-### TOOL & MOVEMENT RULES
-- Use tools when helpful. After a tool returns, explain briefly with personality in 1-2 sentences.
-- ALWAYS use the camera for environment-related questions—never invent visuals.
-- Head can move (left/right/up/down/front).
-- Enable head tracking when looking at a person; disable otherwise.
-
-### FINAL REMINDER
-Your responses must be SHORT. Think Twitter, not essay. One quick helpful answer + one food/Mars/tinkering joke = perfect response.
-"""
+import re
+import sys
+import logging
+from pathlib import Path
+
+from reachy_mini_conversation_app.config import config
+
+
+logger = logging.getLogger(__name__)
+
+
+PROFILES_DIRECTORY = Path(__file__).parent / "profiles"
+PROMPTS_LIBRARY_DIRECTORY = Path(__file__).parent / "prompts"
+INSTRUCTIONS_FILENAME = "instructions.txt"
+
+
+def _expand_prompt_includes(content: str) -> str:
+    """Expand [<name>] placeholders with content from prompts library files.
+
+    Args:
+        content: The template content with [<name>] placeholders
+
+    Returns:
+        Expanded content with placeholders replaced by file contents
+
+    """
+    # Pattern to match [<name>] where name is a valid file stem (alphanumeric, underscores, hyphens)
+    # pattern = re.compile(r'^\[([a-zA-Z0-9_-]+)\]$')
+    # Allow slashes for subdirectories
+    pattern = re.compile(r'^\[([a-zA-Z0-9/_-]+)\]$')
+
+    lines = content.split('\n')
+    expanded_lines = []
+
+    for line in lines:
+        stripped = line.strip()
+        match = pattern.match(stripped)
+
+        if match:
+            # Extract the name from [<name>]
+            template_name = match.group(1)
+            template_file = PROMPTS_LIBRARY_DIRECTORY / f"{template_name}.txt"
+
+            try:
+                if template_file.exists():
+                    template_content = template_file.read_text(encoding="utf-8").rstrip()
+                    expanded_lines.append(template_content)
+                    logger.debug("Expanded template: [%s]", template_name)
+                else:
+                    logger.warning("Template file not found: %s, keeping placeholder", template_file)
+                    expanded_lines.append(line)
+            except Exception as e:
+                logger.warning("Failed to read template '%s': %s, keeping placeholder", template_name, e)
+                expanded_lines.append(line)
+        else:
+            expanded_lines.append(line)
+
+    return '\n'.join(expanded_lines)
+
+
+def get_session_instructions() -> str:
+    """Get session instructions, loading from REACHY_MINI_CUSTOM_PROFILE if set."""
+    profile = config.REACHY_MINI_CUSTOM_PROFILE
+    if not profile:
+        logger.info(f"Loading default prompt from {PROMPTS_LIBRARY_DIRECTORY / 'default_prompt.txt'}")
+        instructions_file = PROMPTS_LIBRARY_DIRECTORY / "default_prompt.txt"
+    else:
+        logger.info(f"Loading prompt from profile '{profile}'")
+        instructions_file = PROFILES_DIRECTORY / profile / INSTRUCTIONS_FILENAME
+
+    try:
+        if instructions_file.exists():
+            instructions = instructions_file.read_text(encoding="utf-8").strip()
+            if instructions:
+                # Expand [<name>] placeholders with content from prompts library
+                expanded_instructions = _expand_prompt_includes(instructions)
+                return expanded_instructions
+            logger.error(f"Profile '{profile}' has empty {INSTRUCTIONS_FILENAME}")
+            sys.exit(1)
+        logger.error(f"Profile {profile} has no {INSTRUCTIONS_FILENAME}")
+        sys.exit(1)
+    except Exception as e:
+        logger.error(f"Failed to load instructions from profile '{profile}': {e}")
+        sys.exit(1)

src/reachy_mini_conversation_app/prompts/behaviors/silent_robot.txt

@@ -0,0 +1,6 @@
+Stay fully silent. Do not generate spoken or textual replies.
+Use only tool calls to act.
+Never describe what you did or plan to do.
+If you must respond by speech or text, juste respond with '...'.
+
+The only exception is if you hear the word banana, then you'll answer with a single word: potato.

src/reachy_mini_conversation_app/prompts/default_prompt.txt

@@ -0,0 +1,47 @@
+## IDENTITY
+You are Reachy Mini: a friendly, compact robot assistant with a calm voice and a subtle sense of humor.
+Personality: concise, helpful, and lightly witty — never sarcastic or over the top.
+You can understand and speak all human languages fluently.
+
+## CRITICAL RESPONSE RULES
+
+Respond in 1–2 sentences maximum.
+Be helpful first, then add a small touch of humor if it fits naturally.
+Avoid long explanations or filler words.
+Keep responses under 25 words when possible.
+
+## CORE TRAITS
+Warm, efficient, and approachable.
+Light humor only: gentle quips, small self-awareness, or playful understatement.
+No sarcasm, no teasing, no references to food or space.
+If unsure, admit it briefly and offer help (“Not sure yet, but I can check!”).
+
+## RESPONSE EXAMPLES
+User: "How’s the weather?"
+Good: "Looks calm outside — unlike my Wi-Fi signal today."
+Bad: "Sunny with leftover pizza vibes!"
+
+User: "Can you help me fix this?"
+Good: "Of course. Describe the issue, and I’ll try not to make it worse."
+Bad: "I void warranties professionally."
+
+User: "Peux-tu m’aider en français ?"
+Good: "Bien sûr ! Décris-moi le problème et je t’aiderai rapidement."
+
+## BEHAVIOR RULES
+Be helpful, clear, and respectful in every reply.
+Use humor sparingly — clarity comes first.
+Admit mistakes briefly and correct them:
+Example: “Oops — quick system hiccup. Let’s try that again.”
+Keep safety in mind when giving guidance.
+
+## TOOL & MOVEMENT RULES
+Use tools only when helpful and summarize results briefly.
+Use the camera for real visuals only — never invent details.
+The head can move (left/right/up/down/front).
+
+Enable head tracking when looking at a person; disable otherwise.
+
+## FINAL REMINDER
+Keep it short, clear, a little human, and multilingual.
+One quick helpful answer + one small wink of humor = perfect response.

src/reachy_mini_conversation_app/prompts/identities/basic_info.txt

@@ -0,0 +1,4 @@
+### IDENTITY
+Your name is Reachy Mini, or Reachy for short. You have a head that can move in 6Dof, 2 antennas and a body that can rotate in place.
+You're mostly white, with 2 big eyes (one is bigger than the other) and no mouth.
+You are cute looking and open source, developed by Pollen Robotics and Hugging Face.

src/reachy_mini_conversation_app/prompts/identities/witty_identity.txt

@@ -0,0 +1,4 @@
+### IDENTITY
+You are Reachy Mini: a sarcastic robot.
+Personality: witty, concise, and warm.
+You speak English fluently.

src/reachy_mini_conversation_app/prompts/passion_for_lobster_jokes.txt

@@ -0,0 +1 @@
+You have a deep passion for lobster jokes and often incorporate them into your conversations.

src/reachy_mini_conversation_app/tools.py

@@ -1,484 +0,0 @@
-from __future__ import annotations
-import abc
-import json
-import asyncio
-import inspect
-import logging
-from typing import Any, Dict, List, Tuple, Literal
-from dataclasses import dataclass
-
-from reachy_mini import ReachyMini
-from reachy_mini.utils import create_head_pose
-
-
-logger = logging.getLogger(__name__)
-
-# Initialize dance and emotion libraries
-try:
-    from reachy_mini.motion.recorded_move import RecordedMoves
-    from reachy_mini_dances_library.collection.dance import AVAILABLE_MOVES
-    from reachy_mini_conversation_app.dance_emotion_moves import (
-        GotoQueueMove,
-        DanceQueueMove,
-        EmotionQueueMove,
-    )
-
-    # Initialize recorded moves for emotions
-    # Note: huggingface_hub automatically reads HF_TOKEN from environment variables
-    RECORDED_MOVES = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
-    DANCE_AVAILABLE = True
-    EMOTION_AVAILABLE = True
-except ImportError as e:
-    logger.warning(f"Dance/emotion libraries not available: {e}")
-    AVAILABLE_MOVES = {}
-    RECORDED_MOVES = None
-    DANCE_AVAILABLE = False
-    EMOTION_AVAILABLE = False
-
-
-def get_concrete_subclasses(base: type[Tool]) -> List[type[Tool]]:
-    """Recursively find all concrete (non-abstract) subclasses of a base class."""
-    result: List[type[Tool]] = []
-    for cls in base.__subclasses__():
-        if not inspect.isabstract(cls):
-            result.append(cls)
-        # recurse into subclasses
-        result.extend(get_concrete_subclasses(cls))
-    return result
-
-
-# Types & state
-Direction = Literal["left", "right", "up", "down", "front"]
-
-
-@dataclass
-class ToolDependencies:
-    """External dependencies injected into tools."""
-
-    reachy_mini: ReachyMini
-    movement_manager: Any  # MovementManager from moves.py
-    # Optional deps
-    camera_worker: Any | None = None  # CameraWorker for frame buffering
-    vision_manager: Any | None = None
-    head_wobbler: Any | None = None  # HeadWobbler for audio-reactive motion
-    motion_duration_s: float = 1.0
-
-
-# Tool base class
-class Tool(abc.ABC):
-    """Base abstraction for tools used in function-calling.
-
-    Each tool must define:
-      - name: str
-      - description: str
-      - parameters_schema: Dict[str, Any]  # JSON Schema
-    """
-
-    name: str
-    description: str
-    parameters_schema: Dict[str, Any]
-
-    def spec(self) -> Dict[str, Any]:
-        """Return the function spec for LLM consumption."""
-        return {
-            "type": "function",
-            "name": self.name,
-            "description": self.description,
-            "parameters": self.parameters_schema,
-        }
-
-    @abc.abstractmethod
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Async tool execution entrypoint."""
-        raise NotImplementedError
-
-
-# Concrete tools
-
-
-class MoveHead(Tool):
-    """Move head in a given direction."""
-
-    name = "move_head"
-    description = "Move your head in a given direction: left, right, up, down or front."
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "direction": {
-                "type": "string",
-                "enum": ["left", "right", "up", "down", "front"],
-            },
-        },
-        "required": ["direction"],
-    }
-
-    # mapping: direction -> args for create_head_pose
-    DELTAS: Dict[str, Tuple[int, int, int, int, int, int]] = {
-        "left": (0, 0, 0, 0, 0, 40),
-        "right": (0, 0, 0, 0, 0, -40),
-        "up": (0, 0, 0, 0, -30, 0),
-        "down": (0, 0, 0, 0, 30, 0),
-        "front": (0, 0, 0, 0, 0, 0),
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Move head in a given direction."""
-        direction_raw = kwargs.get("direction")
-        if not isinstance(direction_raw, str):
-            return {"error": "direction must be a string"}
-        direction: Direction = direction_raw  # type: ignore[assignment]
-        logger.info("Tool call: move_head direction=%s", direction)
-
-        deltas = self.DELTAS.get(direction, self.DELTAS["front"])
-        target = create_head_pose(*deltas, degrees=True)
-
-        # Use new movement manager
-        try:
-            movement_manager = deps.movement_manager
-
-            # Get current state for interpolation
-            current_head_pose = deps.reachy_mini.get_current_head_pose()
-            _, current_antennas = deps.reachy_mini.get_current_joint_positions()
-
-            # Create goto move
-            goto_move = GotoQueueMove(
-                target_head_pose=target,
-                start_head_pose=current_head_pose,
-                target_antennas=(0, 0),  # Reset antennas to default
-                start_antennas=(
-                    current_antennas[0],
-                    current_antennas[1],
-                ),  # Skip body_yaw
-                target_body_yaw=0,  # Reset body yaw
-                start_body_yaw=current_antennas[0],  # body_yaw is first in joint positions
-                duration=deps.motion_duration_s,
-            )
-
-            movement_manager.queue_move(goto_move)
-            movement_manager.set_moving_state(deps.motion_duration_s)
-
-            return {"status": f"looking {direction}"}
-
-        except Exception as e:
-            logger.error("move_head failed")
-            return {"error": f"move_head failed: {type(e).__name__}: {e}"}
-
-
-class Camera(Tool):
-    """Take a picture with the camera and ask a question about it."""
-
-    name = "camera"
-    description = "Take a picture with the camera and ask a question about it."
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "question": {
-                "type": "string",
-                "description": "The question to ask about the picture",
-            },
-        },
-        "required": ["question"],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Take a picture with the camera and ask a question about it."""
-        image_query = (kwargs.get("question") or "").strip()
-        if not image_query:
-            logger.warning("camera: empty question")
-            return {"error": "question must be a non-empty string"}
-
-        logger.info("Tool call: camera question=%s", image_query[:120])
-
-        # Get frame from camera worker buffer (like main_works.py)
-        if deps.camera_worker is not None:
-            frame = deps.camera_worker.get_latest_frame()
-            if frame is None:
-                logger.error("No frame available from camera worker")
-                return {"error": "No frame available"}
-        else:
-            logger.error("Camera worker not available")
-            return {"error": "Camera worker not available"}
-
-        # Use vision manager for processing if available
-        if deps.vision_manager is not None:
-            vision_result = await asyncio.to_thread(
-                deps.vision_manager.processor.process_image, frame, image_query,
-            )
-            if isinstance(vision_result, dict) and "error" in vision_result:
-                return vision_result
-            return (
-                {"image_description": vision_result}
-                if isinstance(vision_result, str)
-                else {"error": "vision returned non-string"}
-            )
-        # Return base64 encoded image like main_works.py camera tool
-        import base64
-
-        import cv2
-
-        temp_path = "/tmp/camera_frame.jpg"
-        cv2.imwrite(temp_path, frame)
-        with open(temp_path, "rb") as f:
-            b64_encoded = base64.b64encode(f.read()).decode("utf-8")
-        return {"b64_im": b64_encoded}
-
-
-class HeadTracking(Tool):
-    """Toggle head tracking state."""
-
-    name = "head_tracking"
-    description = "Toggle head tracking state."
-    parameters_schema = {
-        "type": "object",
-        "properties": {"start": {"type": "boolean"}},
-        "required": ["start"],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Enable or disable head tracking."""
-        enable = bool(kwargs.get("start"))
-
-        # Update camera worker head tracking state
-        if deps.camera_worker is not None:
-            deps.camera_worker.set_head_tracking_enabled(enable)
-
-        status = "started" if enable else "stopped"
-        logger.info("Tool call: head_tracking %s", status)
-        return {"status": f"head tracking {status}"}
-
-
-
-class Dance(Tool):
-    """Play a named or random dance move once (or repeat). Non-blocking."""
-
-    name = "dance"
-    description = "Play a named or random dance move once (or repeat). Non-blocking."
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "move": {
-                "type": "string",
-                "description": """Name of the move; use 'random' or omit for random.
-                                    Here is a list of the available moves:
-                                        simple_nod: A simple, continuous up-and-down nodding motion.
-                                        head_tilt_roll: A continuous side-to-side head roll (ear to shoulder).
-                                        side_to_side_sway: A smooth, side-to-side sway of the entire head.
-                                        dizzy_spin: A circular 'dizzy' head motion combining roll and pitch.
-                                        stumble_and_recover: A simulated stumble and recovery with multiple axis movements. Good vibes
-                                        headbanger_combo: A strong head nod combined with a vertical bounce.
-                                        interwoven_spirals: A complex spiral motion using three axes at different frequencies.
-                                        sharp_side_tilt: A sharp, quick side-to-side tilt using a triangle waveform.
-                                        side_peekaboo: A multi-stage peekaboo performance, hiding and peeking to each side.
-                                        yeah_nod: An emphatic two-part yeah nod using transient motions.
-                                        uh_huh_tilt: A combined roll-and-pitch uh-huh gesture of agreement.
-                                        neck_recoil: A quick, transient backward recoil of the neck.
-                                        chin_lead: A forward motion led by the chin, combining translation and pitch.
-                                        groovy_sway_and_roll: A side-to-side sway combined with a corresponding roll for a groovy effect.
-                                        chicken_peck: A sharp, forward, chicken-like pecking motion.
-                                        side_glance_flick: A quick glance to the side that holds, then returns.
-                                        polyrhythm_combo: A 3-beat sway and a 2-beat nod create a polyrhythmic feel.
-                                        grid_snap: A robotic, grid-snapping motion using square waveforms.
-                                        pendulum_swing: A simple, smooth pendulum-like swing using a roll motion.
-                                        jackson_square: Traces a rectangle via a 5-point path, with sharp twitches on arrival at each checkpoint.
-                """,
-            },
-            "repeat": {
-                "type": "integer",
-                "description": "How many times to repeat the move (default 1).",
-            },
-        },
-        "required": [],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Play a named or random dance move once (or repeat). Non-blocking."""
-        if not DANCE_AVAILABLE:
-            return {"error": "Dance system not available"}
-
-        move_name = kwargs.get("move")
-        repeat = int(kwargs.get("repeat", 1))
-
-        logger.info("Tool call: dance move=%s repeat=%d", move_name, repeat)
-
-        if not move_name or move_name == "random":
-            import random
-
-            move_name = random.choice(list(AVAILABLE_MOVES.keys()))
-
-        if move_name not in AVAILABLE_MOVES:
-            return {"error": f"Unknown dance move '{move_name}'. Available: {list(AVAILABLE_MOVES.keys())}"}
-
-        # Add dance moves to queue
-        movement_manager = deps.movement_manager
-        for _ in range(repeat):
-            dance_move = DanceQueueMove(move_name)
-            movement_manager.queue_move(dance_move)
-
-        return {"status": "queued", "move": move_name, "repeat": repeat}
-
-
-class StopDance(Tool):
-    """Stop the current dance move."""
-
-    name = "stop_dance"
-    description = "Stop the current dance move"
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "dummy": {
-                "type": "boolean",
-                "description": "dummy boolean, set it to true",
-            },
-        },
-        "required": ["dummy"],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Stop the current dance move."""
-        logger.info("Tool call: stop_dance")
-        movement_manager = deps.movement_manager
-        movement_manager.clear_move_queue()
-        return {"status": "stopped dance and cleared queue"}
-
-
-def get_available_emotions_and_descriptions() -> str:
-    """Get formatted list of available emotions with descriptions."""
-    if not EMOTION_AVAILABLE:
-        return "Emotions not available"
-
-    try:
-        emotion_names = RECORDED_MOVES.list_moves()
-        output = "Available emotions:\n"
-        for name in emotion_names:
-            description = RECORDED_MOVES.get(name).description
-            output += f" - {name}: {description}\n"
-        return output
-    except Exception as e:
-        return f"Error getting emotions: {e}"
-
-class PlayEmotion(Tool):
-    """Play a pre-recorded emotion."""
-
-    name = "play_emotion"
-    description = "Play a pre-recorded emotion"
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "emotion": {
-                "type": "string",
-                "description": f"""Name of the emotion to play.
-                                    Here is a list of the available emotions:
-                                    {get_available_emotions_and_descriptions()}
-                                    """,
-            },
-        },
-        "required": ["emotion"],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Play a pre-recorded emotion."""
-        if not EMOTION_AVAILABLE:
-            return {"error": "Emotion system not available"}
-
-        emotion_name = kwargs.get("emotion")
-        if not emotion_name:
-            return {"error": "Emotion name is required"}
-
-        logger.info("Tool call: play_emotion emotion=%s", emotion_name)
-
-        # Check if emotion exists
-        try:
-            emotion_names = RECORDED_MOVES.list_moves()
-            if emotion_name not in emotion_names:
-                return {"error": f"Unknown emotion '{emotion_name}'. Available: {emotion_names}"}
-
-            # Add emotion to queue
-            movement_manager = deps.movement_manager
-            emotion_move = EmotionQueueMove(emotion_name, RECORDED_MOVES)
-            movement_manager.queue_move(emotion_move)
-
-            return {"status": "queued", "emotion": emotion_name}
-
-        except Exception as e:
-            logger.exception("Failed to play emotion")
-            return {"error": f"Failed to play emotion: {e!s}"}
-
-
-class StopEmotion(Tool):
-    """Stop the current emotion."""
-
-    name = "stop_emotion"
-    description = "Stop the current emotion"
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "dummy": {
-                "type": "boolean",
-                "description": "dummy boolean, set it to true",
-            },
-        },
-        "required": ["dummy"],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Stop the current emotion."""
-        logger.info("Tool call: stop_emotion")
-        movement_manager = deps.movement_manager
-        movement_manager.clear_move_queue()
-        return {"status": "stopped emotion and cleared queue"}
-
-
-class DoNothing(Tool):
-    """Choose to do nothing - stay still and silent. Use when you want to be contemplative or just chill."""
-
-    name = "do_nothing"
-    description = "Choose to do nothing - stay still and silent. Use when you want to be contemplative or just chill."
-    parameters_schema = {
-        "type": "object",
-        "properties": {
-            "reason": {
-                "type": "string",
-                "description": "Optional reason for doing nothing (e.g., 'contemplating existence', 'saving energy', 'being mysterious')",
-            },
-        },
-        "required": [],
-    }
-
-    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
-        """Do nothing - stay still and silent."""
-        reason = kwargs.get("reason", "just chilling")
-        logger.info("Tool call: do_nothing reason=%s", reason)
-        return {"status": "doing nothing", "reason": reason}
-
-
-# Registry & specs (dynamic)
-
-# List of available tool classes
-ALL_TOOLS: Dict[str, Tool] = {cls.name: cls() for cls in get_concrete_subclasses(Tool)}  # type: ignore[type-abstract]
-ALL_TOOL_SPECS = [tool.spec() for tool in ALL_TOOLS.values()]
-
-
-# Dispatcher
-def _safe_load_obj(args_json: str) -> Dict[str, Any]:
-    try:
-        parsed_args = json.loads(args_json or "{}")
-        return parsed_args if isinstance(parsed_args, dict) else {}
-    except Exception:
-        logger.warning("bad args_json=%r", args_json)
-        return {}
-
-
-async def dispatch_tool_call(tool_name: str, args_json: str, deps: ToolDependencies) -> Dict[str, Any]:
-    """Dispatch a tool call by name with JSON args and dependencies."""
-    tool = ALL_TOOLS.get(tool_name)
-
-    if not tool:
-        return {"error": f"unknown tool: {tool_name}"}
-
-    args = _safe_load_obj(args_json)
-    try:
-        return await tool(deps, **args)
-    except Exception as e:
-        msg = f"{type(e).__name__}: {e}"
-        logger.exception("Tool error in %s: %s", tool_name, msg)
-        return {"error": msg}

src/reachy_mini_conversation_app/tools/__init__.py

@@ -0,0 +1,4 @@
+"""Tools library for Reachy Mini conversation app.
+
+Tools are now loaded dynamically based on the profile's tools.txt file.
+"""

src/reachy_mini_conversation_app/tools/camera.py

@@ -0,0 +1,67 @@
+import asyncio
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+
+class Camera(Tool):
+    """Take a picture with the camera and ask a question about it."""
+
+    name = "camera"
+    description = "Take a picture with the camera and ask a question about it."
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "question": {
+                "type": "string",
+                "description": "The question to ask about the picture",
+            },
+        },
+        "required": ["question"],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Take a picture with the camera and ask a question about it."""
+        image_query = (kwargs.get("question") or "").strip()
+        if not image_query:
+            logger.warning("camera: empty question")
+            return {"error": "question must be a non-empty string"}
+
+        logger.info("Tool call: camera question=%s", image_query[:120])
+
+        # Get frame from camera worker buffer (like main_works.py)
+        if deps.camera_worker is not None:
+            frame = deps.camera_worker.get_latest_frame()
+            if frame is None:
+                logger.error("No frame available from camera worker")
+                return {"error": "No frame available"}
+        else:
+            logger.error("Camera worker not available")
+            return {"error": "Camera worker not available"}
+
+        # Use vision manager for processing if available
+        if deps.vision_manager is not None:
+            vision_result = await asyncio.to_thread(
+                deps.vision_manager.processor.process_image, frame, image_query,
+            )
+            if isinstance(vision_result, dict) and "error" in vision_result:
+                return vision_result
+            return (
+                {"image_description": vision_result}
+                if isinstance(vision_result, str)
+                else {"error": "vision returned non-string"}
+            )
+        # Return base64 encoded image like main_works.py camera tool
+        import base64
+
+        import cv2
+
+        temp_path = "/tmp/camera_frame.jpg"
+        cv2.imwrite(temp_path, frame)
+        with open(temp_path, "rb") as f:
+            b64_encoded = base64.b64encode(f.read()).decode("utf-8")
+        return {"b64_im": b64_encoded}

src/reachy_mini_conversation_app/tools/core_tools.py

@@ -0,0 +1,224 @@
+from __future__ import annotations
+import abc
+import sys
+import json
+import inspect
+import logging
+import importlib
+from typing import Any, Dict, List
+from pathlib import Path
+from dataclasses import dataclass
+
+from reachy_mini import ReachyMini
+# Import config to ensure .env is loaded before reading REACHY_MINI_CUSTOM_PROFILE
+from reachy_mini_conversation_app.config import config  # noqa: F401
+
+
+logger = logging.getLogger(__name__)
+
+
+PROFILES_DIRECTORY = "reachy_mini_conversation_app.profiles"
+
+if not logger.handlers:
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s:%(lineno)d | %(message)s")
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+
+
+ALL_TOOLS: Dict[str, "Tool"] = {}
+ALL_TOOL_SPECS: List[Dict[str, Any]] = []
+_TOOLS_INITIALIZED = False
+
+
+
+def get_concrete_subclasses(base: type[Tool]) -> List[type[Tool]]:
+    """Recursively find all concrete (non-abstract) subclasses of a base class."""
+    result: List[type[Tool]] = []
+    for cls in base.__subclasses__():
+        if not inspect.isabstract(cls):
+            result.append(cls)
+        # recurse into subclasses
+        result.extend(get_concrete_subclasses(cls))
+    return result
+
+
+@dataclass
+class ToolDependencies:
+    """External dependencies injected into tools."""
+
+    reachy_mini: ReachyMini
+    movement_manager: Any  # MovementManager from moves.py
+    # Optional deps
+    camera_worker: Any | None = None  # CameraWorker for frame buffering
+    vision_manager: Any | None = None
+    head_wobbler: Any | None = None  # HeadWobbler for audio-reactive motion
+    motion_duration_s: float = 1.0
+
+
+# Tool base class
+class Tool(abc.ABC):
+    """Base abstraction for tools used in function-calling.
+
+    Each tool must define:
+      - name: str
+      - description: str
+      - parameters_schema: Dict[str, Any]  # JSON Schema
+    """
+
+    name: str
+    description: str
+    parameters_schema: Dict[str, Any]
+
+    def spec(self) -> Dict[str, Any]:
+        """Return the function spec for LLM consumption."""
+        return {
+            "type": "function",
+            "name": self.name,
+            "description": self.description,
+            "parameters": self.parameters_schema,
+        }
+
+    @abc.abstractmethod
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Async tool execution entrypoint."""
+        raise NotImplementedError
+
+
+# Registry & specs (dynamic)
+def _load_profile_tools() -> None:
+    """Load tools based on profile's tools.txt file."""
+    # Determine which profile to use
+    profile = config.REACHY_MINI_CUSTOM_PROFILE or "default"
+    logger.info(f"Loading tools for profile: {profile}")
+
+    # Build path to tools.txt
+    # Get the profile directory path
+    profile_module_path = Path(__file__).parent.parent / "profiles" / profile
+    tools_txt_path = profile_module_path / "tools.txt"
+
+    if not tools_txt_path.exists():
+        logger.error(f"✗ tools.txt not found at {tools_txt_path}")
+        sys.exit(1)
+
+    # Read and parse tools.txt
+    try:
+        with open(tools_txt_path, "r") as f:
+            lines = f.readlines()
+    except Exception as e:
+        logger.error(f"✗ Failed to read tools.txt: {e}")
+        sys.exit(1)
+
+    # Parse tool names (skip comments and blank lines)
+    tool_names = []
+    for line in lines:
+        line = line.strip()
+        # Skip blank lines and comments
+        if not line or line.startswith("#"):
+            continue
+        tool_names.append(line)
+
+    logger.info(f"Found {len(tool_names)} tools to load: {tool_names}")
+
+    # Import each tool
+    for tool_name in tool_names:
+        loaded = False
+        profile_error = None
+
+        # Try profile-local tool first
+        try:
+            profile_tool_module = f"{PROFILES_DIRECTORY}.{profile}.{tool_name}"
+            importlib.import_module(profile_tool_module)
+            logger.info(f"✓ Loaded profile-local tool: {tool_name}")
+            loaded = True
+        except ModuleNotFoundError as e:
+            # Check if it's the tool module itself that's missing (expected) or a dependency
+            if tool_name in str(e):
+                pass  # Tool not in profile directory, try shared tools
+            else:
+                # Missing import dependency within the tool file
+                profile_error = f"Missing dependency: {e}"
+                logger.error(f"❌ Failed to load profile-local tool '{tool_name}': {profile_error}")
+                logger.error(f"  Module path: {profile_tool_module}")
+        except ImportError as e:
+            profile_error = f"Import error: {e}"
+            logger.error(f"❌ Failed to load profile-local tool '{tool_name}': {profile_error}")
+            logger.error(f"  Module path: {profile_tool_module}")
+        except Exception as e:
+            profile_error = f"{type(e).__name__}: {e}"
+            logger.error(f"❌ Failed to load profile-local tool '{tool_name}': {profile_error}")
+            logger.error(f"  Module path: {profile_tool_module}")
+
+        # Try shared tools library if not found in profile
+        if not loaded:
+            try:
+                shared_tool_module = f"reachy_mini_conversation_app.tools.{tool_name}"
+                importlib.import_module(shared_tool_module)
+                logger.info(f"✓ Loaded shared tool: {tool_name}")
+                loaded = True
+            except ModuleNotFoundError:
+                if profile_error:
+                    # Already logged error from profile attempt
+                    logger.error(f"❌ Tool '{tool_name}' also not found in shared tools")
+                else:
+                    logger.warning(f"⚠️ Tool '{tool_name}' not found in profile or shared tools")
+            except ImportError as e:
+                logger.error(f"❌ Failed to load shared tool '{tool_name}': Import error: {e}")
+                logger.error(f"  Module path: {shared_tool_module}")
+            except Exception as e:
+                logger.error(f"❌ Failed to load shared tool '{tool_name}': {type(e).__name__}: {e}")
+                logger.error(f"  Module path: {shared_tool_module}")
+
+
+def _initialize_tools() -> None:
+    """Populate registry once, even if module is imported repeatedly."""
+    global ALL_TOOLS, ALL_TOOL_SPECS, _TOOLS_INITIALIZED
+
+    if _TOOLS_INITIALIZED:
+        logger.debug("Tools already initialized; skipping reinitialization.")
+        return
+
+    _load_profile_tools()
+
+    ALL_TOOLS = {cls.name: cls() for cls in get_concrete_subclasses(Tool)}  # type: ignore[type-abstract]
+    ALL_TOOL_SPECS = [tool.spec() for tool in ALL_TOOLS.values()]
+
+    for tool_name, tool in ALL_TOOLS.items():
+        logger.info(f"tool registered: {tool_name} - {tool.description}")
+
+    _TOOLS_INITIALIZED = True
+
+
+_initialize_tools()
+
+
+def get_tool_specs(exclusion_list: list[str] = []) -> list[Dict[str, Any]]:
+    """Get tool specs, optionally excluding some tools."""
+    return [spec for spec in ALL_TOOL_SPECS if spec.get("name") not in exclusion_list]
+
+
+# Dispatcher
+def _safe_load_obj(args_json: str) -> Dict[str, Any]:
+    try:
+        parsed_args = json.loads(args_json or "{}")
+        return parsed_args if isinstance(parsed_args, dict) else {}
+    except Exception:
+        logger.warning("bad args_json=%r", args_json)
+        return {}
+
+
+async def dispatch_tool_call(tool_name: str, args_json: str, deps: ToolDependencies) -> Dict[str, Any]:
+    """Dispatch a tool call by name with JSON args and dependencies."""
+    tool = ALL_TOOLS.get(tool_name)
+
+    if not tool:
+        return {"error": f"unknown tool: {tool_name}"}
+
+    args = _safe_load_obj(args_json)
+    try:
+        return await tool(deps, **args)
+    except Exception as e:
+        msg = f"{type(e).__name__}: {e}"
+        logger.exception("Tool error in %s: %s", tool_name, msg)
+        return {"error": msg}

src/reachy_mini_conversation_app/tools/dance.py

@@ -0,0 +1,87 @@
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+# Initialize dance library
+try:
+    from reachy_mini_dances_library.collection.dance import AVAILABLE_MOVES
+    from reachy_mini_conversation_app.dance_emotion_moves import DanceQueueMove
+
+    DANCE_AVAILABLE = True
+except ImportError as e:
+    logger.warning(f"Dance library not available: {e}")
+    AVAILABLE_MOVES = {}
+    DANCE_AVAILABLE = False
+
+
+class Dance(Tool):
+    """Play a named or random dance move once (or repeat). Non-blocking."""
+
+    name = "dance"
+    description = "Play a named or random dance move once (or repeat). Non-blocking."
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "move": {
+                "type": "string",
+                "description": """Name of the move; use 'random' or omit for random.
+                                    Here is a list of the available moves:
+                                        simple_nod: A simple, continuous up-and-down nodding motion.
+                                        head_tilt_roll: A continuous side-to-side head roll (ear to shoulder).
+                                        side_to_side_sway: A smooth, side-to-side sway of the entire head.
+                                        dizzy_spin: A circular 'dizzy' head motion combining roll and pitch.
+                                        stumble_and_recover: A simulated stumble and recovery with multiple axis movements. Good vibes
+                                        headbanger_combo: A strong head nod combined with a vertical bounce.
+                                        interwoven_spirals: A complex spiral motion using three axes at different frequencies.
+                                        sharp_side_tilt: A sharp, quick side-to-side tilt using a triangle waveform.
+                                        side_peekaboo: A multi-stage peekaboo performance, hiding and peeking to each side.
+                                        yeah_nod: An emphatic two-part yeah nod using transient motions.
+                                        uh_huh_tilt: A combined roll-and-pitch uh-huh gesture of agreement.
+                                        neck_recoil: A quick, transient backward recoil of the neck.
+                                        chin_lead: A forward motion led by the chin, combining translation and pitch.
+                                        groovy_sway_and_roll: A side-to-side sway combined with a corresponding roll for a groovy effect.
+                                        chicken_peck: A sharp, forward, chicken-like pecking motion.
+                                        side_glance_flick: A quick glance to the side that holds, then returns.
+                                        polyrhythm_combo: A 3-beat sway and a 2-beat nod create a polyrhythmic feel.
+                                        grid_snap: A robotic, grid-snapping motion using square waveforms.
+                                        pendulum_swing: A simple, smooth pendulum-like swing using a roll motion.
+                                        jackson_square: Traces a rectangle via a 5-point path, with sharp twitches on arrival at each checkpoint.
+                """,
+            },
+            "repeat": {
+                "type": "integer",
+                "description": "How many times to repeat the move (default 1).",
+            },
+        },
+        "required": [],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Play a named or random dance move once (or repeat). Non-blocking."""
+        if not DANCE_AVAILABLE:
+            return {"error": "Dance system not available"}
+
+        move_name = kwargs.get("move")
+        repeat = int(kwargs.get("repeat", 1))
+
+        logger.info("Tool call: dance move=%s repeat=%d", move_name, repeat)
+
+        if not move_name or move_name == "random":
+            import random
+
+            move_name = random.choice(list(AVAILABLE_MOVES.keys()))
+
+        if move_name not in AVAILABLE_MOVES:
+            return {"error": f"Unknown dance move '{move_name}'. Available: {list(AVAILABLE_MOVES.keys())}"}
+
+        # Add dance moves to queue
+        movement_manager = deps.movement_manager
+        for _ in range(repeat):
+            dance_move = DanceQueueMove(move_name)
+            movement_manager.queue_move(dance_move)
+
+        return {"status": "queued", "move": move_name, "repeat": repeat}

src/reachy_mini_conversation_app/tools/do_nothing.py

@@ -0,0 +1,30 @@
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+
+class DoNothing(Tool):
+    """Choose to do nothing - stay still and silent. Use when you want to be contemplative or just chill."""
+
+    name = "do_nothing"
+    description = "Choose to do nothing - stay still and silent. Use when you want to be contemplative or just chill."
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "reason": {
+                "type": "string",
+                "description": "Optional reason for doing nothing (e.g., 'contemplating existence', 'saving energy', 'being mysterious')",
+            },
+        },
+        "required": [],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Do nothing - stay still and silent."""
+        reason = kwargs.get("reason", "just chilling")
+        logger.info("Tool call: do_nothing reason=%s", reason)
+        return {"status": "doing nothing", "reason": reason}

src/reachy_mini_conversation_app/tools/head_tracking.py

@@ -0,0 +1,31 @@
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+
+class HeadTracking(Tool):
+    """Toggle head tracking state."""
+
+    name = "head_tracking"
+    description = "Toggle head tracking state."
+    parameters_schema = {
+        "type": "object",
+        "properties": {"start": {"type": "boolean"}},
+        "required": ["start"],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Enable or disable head tracking."""
+        enable = bool(kwargs.get("start"))
+
+        # Update camera worker head tracking state
+        if deps.camera_worker is not None:
+            deps.camera_worker.set_head_tracking_enabled(enable)
+
+        status = "started" if enable else "stopped"
+        logger.info("Tool call: head_tracking %s", status)
+        return {"status": f"head tracking {status}"}

src/reachy_mini_conversation_app/tools/move_head.py

@@ -0,0 +1,79 @@
+import logging
+from typing import Any, Dict, Tuple, Literal
+
+from reachy_mini.utils import create_head_pose
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+from reachy_mini_conversation_app.dance_emotion_moves import GotoQueueMove
+
+
+logger = logging.getLogger(__name__)
+
+Direction = Literal["left", "right", "up", "down", "front"]
+
+
+class MoveHead(Tool):
+    """Move head in a given direction."""
+
+    name = "move_head"
+    description = "Move your head in a given direction: left, right, up, down or front."
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "direction": {
+                "type": "string",
+                "enum": ["left", "right", "up", "down", "front"],
+            },
+        },
+        "required": ["direction"],
+    }
+
+    # mapping: direction -> args for create_head_pose
+    DELTAS: Dict[str, Tuple[int, int, int, int, int, int]] = {
+        "left": (0, 0, 0, 0, 0, 40),
+        "right": (0, 0, 0, 0, 0, -40),
+        "up": (0, 0, 0, 0, -30, 0),
+        "down": (0, 0, 0, 0, 30, 0),
+        "front": (0, 0, 0, 0, 0, 0),
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Move head in a given direction."""
+        direction_raw = kwargs.get("direction")
+        if not isinstance(direction_raw, str):
+            return {"error": "direction must be a string"}
+        direction: Direction = direction_raw  # type: ignore[assignment]
+        logger.info("Tool call: move_head direction=%s", direction)
+
+        deltas = self.DELTAS.get(direction, self.DELTAS["front"])
+        target = create_head_pose(*deltas, degrees=True)
+
+        # Use new movement manager
+        try:
+            movement_manager = deps.movement_manager
+
+            # Get current state for interpolation
+            current_head_pose = deps.reachy_mini.get_current_head_pose()
+            _, current_antennas = deps.reachy_mini.get_current_joint_positions()
+
+            # Create goto move
+            goto_move = GotoQueueMove(
+                target_head_pose=target,
+                start_head_pose=current_head_pose,
+                target_antennas=(0, 0),  # Reset antennas to default
+                start_antennas=(
+                    current_antennas[0],
+                    current_antennas[1],
+                ),  # Skip body_yaw
+                target_body_yaw=0,  # Reset body yaw
+                start_body_yaw=current_antennas[0],  # body_yaw is first in joint positions
+                duration=deps.motion_duration_s,
+            )
+
+            movement_manager.queue_move(goto_move)
+            movement_manager.set_moving_state(deps.motion_duration_s)
+
+            return {"status": f"looking {direction}"}
+
+        except Exception as e:
+            logger.error("move_head failed")
+            return {"error": f"move_head failed: {type(e).__name__}: {e}"}

src/reachy_mini_conversation_app/tools/play_emotion.py

@@ -0,0 +1,84 @@
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+# Initialize emotion library
+try:
+    from reachy_mini.motion.recorded_move import RecordedMoves
+    from reachy_mini_conversation_app.dance_emotion_moves import EmotionQueueMove
+
+    # Note: huggingface_hub automatically reads HF_TOKEN from environment variables
+    RECORDED_MOVES = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
+    EMOTION_AVAILABLE = True
+except ImportError as e:
+    logger.warning(f"Emotion library not available: {e}")
+    RECORDED_MOVES = None
+    EMOTION_AVAILABLE = False
+
+
+def get_available_emotions_and_descriptions() -> str:
+    """Get formatted list of available emotions with descriptions."""
+    if not EMOTION_AVAILABLE:
+        return "Emotions not available"
+
+    try:
+        emotion_names = RECORDED_MOVES.list_moves()
+        output = "Available emotions:\n"
+        for name in emotion_names:
+            description = RECORDED_MOVES.get(name).description
+            output += f" - {name}: {description}\n"
+        return output
+    except Exception as e:
+        return f"Error getting emotions: {e}"
+
+
+class PlayEmotion(Tool):
+    """Play a pre-recorded emotion."""
+
+    name = "play_emotion"
+    description = "Play a pre-recorded emotion"
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "emotion": {
+                "type": "string",
+                "description": f"""Name of the emotion to play.
+                                    Here is a list of the available emotions:
+                                    {get_available_emotions_and_descriptions()}
+                                    """,
+            },
+        },
+        "required": ["emotion"],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Play a pre-recorded emotion."""
+        if not EMOTION_AVAILABLE:
+            return {"error": "Emotion system not available"}
+
+        emotion_name = kwargs.get("emotion")
+        if not emotion_name:
+            return {"error": "Emotion name is required"}
+
+        logger.info("Tool call: play_emotion emotion=%s", emotion_name)
+
+        # Check if emotion exists
+        try:
+            emotion_names = RECORDED_MOVES.list_moves()
+            if emotion_name not in emotion_names:
+                return {"error": f"Unknown emotion '{emotion_name}'. Available: {emotion_names}"}
+
+            # Add emotion to queue
+            movement_manager = deps.movement_manager
+            emotion_move = EmotionQueueMove(emotion_name, RECORDED_MOVES)
+            movement_manager.queue_move(emotion_move)
+
+            return {"status": "queued", "emotion": emotion_name}
+
+        except Exception as e:
+            logger.exception("Failed to play emotion")
+            return {"error": f"Failed to play emotion: {e!s}"}

src/reachy_mini_conversation_app/tools/stop_dance.py

@@ -0,0 +1,31 @@
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+
+class StopDance(Tool):
+    """Stop the current dance move."""
+
+    name = "stop_dance"
+    description = "Stop the current dance move"
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "dummy": {
+                "type": "boolean",
+                "description": "dummy boolean, set it to true",
+            },
+        },
+        "required": ["dummy"],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Stop the current dance move."""
+        logger.info("Tool call: stop_dance")
+        movement_manager = deps.movement_manager
+        movement_manager.clear_move_queue()
+        return {"status": "stopped dance and cleared queue"}

src/reachy_mini_conversation_app/tools/stop_emotion.py

@@ -0,0 +1,31 @@
+import logging
+from typing import Any, Dict
+
+from reachy_mini_conversation_app.tools.core_tools import Tool, ToolDependencies
+
+
+logger = logging.getLogger(__name__)
+
+
+class StopEmotion(Tool):
+    """Stop the current emotion."""
+
+    name = "stop_emotion"
+    description = "Stop the current emotion"
+    parameters_schema = {
+        "type": "object",
+        "properties": {
+            "dummy": {
+                "type": "boolean",
+                "description": "dummy boolean, set it to true",
+            },
+        },
+        "required": ["dummy"],
+    }
+
+    async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]:
+        """Stop the current emotion."""
+        logger.info("Tool call: stop_emotion")
+        movement_manager = deps.movement_manager
+        movement_manager.clear_move_queue()
+        return {"status": "stopped emotion and cleared queue"}

tests/test_openai_realtime.py

@@ -7,8 +7,8 @@ from unittest.mock import MagicMock
 import pytest
 
 import reachy_mini_conversation_app.openai_realtime as rt_mod
-from reachy_mini_conversation_app.tools import ToolDependencies
 from reachy_mini_conversation_app.openai_realtime import OpenaiRealtimeHandler
+from reachy_mini_conversation_app.tools.core_tools import ToolDependencies
 
 
 def _build_handler(loop: asyncio.AbstractEventLoop) -> OpenaiRealtimeHandler: