MCP Voice Notification

Instructions

Implementation Reference

• stdio_server.py:25-37 (handler)

  Primary MCP tool handler for voice_notification. Decorated with @app.tool() to register it as an MCP tool. Calls the helper function to generate and play the voice.

  @app.tool()
  def voice_notification(text: str = "Done!") -> str:
      """
      Generate a voice notification using Grok Voice API and play it.
  
      Args:
          text: The text to convert to speech (default: "Done!")
  
      Returns:
          str: Confirmation message
      """
      success, message = generate_and_play_voice(text, GROK_API_KEY)
      return message

• server.py:28-40 (handler)

  Secondary MCP tool handler (HTTP server variant) for voice_notification. Identical logic to stdio version.

  @app.tool()
  def voice_notification(text: str = "Done!") -> str:
      """
      Generate a voice notification using Grok Voice API and play it.
  
      Args:
          text: The text to convert to speech (default: "Done!")
  
      Returns:
          str: Confirmation message
      """
      success, message = generate_and_play_voice(text, GROK_API_KEY)
      return message

• voice.py:16-129 (helper)

  Supporting utility that performs the core voice generation via WebSocket to Grok's realtime API (wss://api.x.ai/v1/realtime), collects PCM audio, saves to WAV, and plays using system audio players (afplay, aplay, etc.).

  def generate_and_play_voice(text: str, api_key: str) -> Tuple[bool, str]:
      """
      Generate a voice notification using Grok Voice API and play it.
  
      Args:
          text: The text to convert to speech
          api_key: Grok API key
  
      Returns:
          Tuple of (success: bool, message: str)
      """
      try:
          uri = "wss://api.x.ai/v1/realtime"
          headers = {
              "Authorization": f"Bearer {api_key}"
          }
  
          ws = websocket.create_connection(uri, header=headers)
          ws.settimeout(5)
  
          # Send session configuration
          session_config = {
              "type": "session.update",
              "session": {
                  "modalities": ["text", "audio"],
                  "instructions": "You are a text-to-speech system. Your only job is to speak exactly what the user provides, word for word. Do not add any greeting, introduction, or response. Just speak the exact text given.",
                  "voice": "alloy",
                  "input_audio_format": "pcm16",
                  "output_audio_format": "pcm16",
                  "input_audio_transcription": {"model": "whisper-1"},
                  "turn_detection": {"type": "server_vad"}
              }
          }
          ws.send(json.dumps(session_config))
  
          # Send a text message
          message = {
              "type": "conversation.item.create",
              "item": {
                  "type": "message",
                  "role": "user",
                  "content": [{"type": "input_text", "text": f"Speak this exactly: {text}"}]
              }
          }
          ws.send(json.dumps(message))
  
          # Trigger response
          ws.send(json.dumps({"type": "response.create"}))
  
          # Collect audio data
          audio_data = b''
          start = time.time()
          while time.time() - start < 10:  # timeout after 10 seconds
              try:
                  msg = ws.recv()
                  data = json.loads(msg)
                  if data.get("type") == "response.output_audio.delta":
                      audio_data += base64.b64decode(data['delta'])
                  elif data.get("type") == "response.done":
                      break
              except websocket.WebSocketTimeoutException:
                  break
              except websocket.WebSocketConnectionClosedException:
                  break
              except json.JSONDecodeError:
                  continue
              except Exception:
                  break
  
          ws.close()
  
          if not audio_data:
              return False, "No audio data received from API"
  
          # Create temporary WAV file
          with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_file:
              temp_wav_path = temp_file.name
  
          with wave.open(temp_wav_path, 'wb') as wf:
              wf.setnchannels(1)
              wf.setsampwidth(2)  # 16 bits
              wf.setframerate(24000)
              wf.writeframes(audio_data)
  
          # Play the audio
          played_successfully = False
          try:
              subprocess.run(["afplay", temp_wav_path], check=True, capture_output=True)
              played_successfully = True
          except subprocess.CalledProcessError:
              pass
          except FileNotFoundError:
              # afplay not available (non-macOS), try other players
              for player in ["aplay", "paplay", "play"]:
                  try:
                      subprocess.run([player, temp_wav_path], check=True, capture_output=True)
                      played_successfully = True
                      break
                  except (subprocess.CalledProcessError, FileNotFoundError):
                      continue
  
          # Clean up temp file
          try:
              os.unlink(temp_wav_path)
          except OSError:
              pass
  
          if played_successfully:
              return True, f"Voice notification played: '{text}'"
          else:
              return False, f"Voice notification generated but failed to play: '{text}'"
  
      except Exception as e:
          return False, f"Error: {e}"

• stdio_server.py:18-18 (registration)

  Creates the FastMCP application instance named "voice-notification".

  app = FastMCP("voice-notification")

• server.py:21-21 (registration)

  Creates the FastMCP application instance named "voice-notification" for HTTP server.

  app = FastMCP("voice-notification")