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")