Kokoro Text to Speech MCP Server

Instructions

Implementation Reference

• mcp-tts.py:495-525 (handler)

  The MCP tool handler for 'text_to_speech', registered via @mcp.tool() decorator. It prepares request data from arguments and delegates to the server's process_tts_request method for execution.
  @mcp.tool() async def text_to_speech(text: str, voice: str = os.environ.get('TTS_VOICE', 'af_heart'), speed: float = float(os.environ.get('TTS_SPEED', 1.0)), lang: str = os.environ.get('TTS_LANGUAGE', 'en-us'), filename: str = None, upload_to_s3: bool = os.environ.get('S3_ENABLED', 'true').lower() == 'true') -> dict: """ Convert text to speech using the Kokoro TTS service. Args: text: The text to convert to speech voice: Voice ID to use (default: af_heart) speed: Speech speed (default: 1.0) lang: Language code (default: en-us) filename: Optional filename for the MP3 (default: auto-generated UUID) upload_to_s3: Whether to upload to S3 if enabled (default: True) Returns: A dictionary with information about the generated audio file """ request_data = { "text": text, "voice": voice, "speed": speed, "lang": lang, "filename": filename, "upload_to_s3": upload_to_s3 } return await mcp_tts_server.process_tts_request(request_data)
• mcp-tts.py:495-495 (registration)

  Registration of the 'text_to_speech' tool using the FastMCP @tool() decorator.
  @mcp.tool()
• mcp-tts.py:286-414 (helper)

  Helper method in MCPTTSServer class that implements the core TTS logic: parameter extraction, audio generation with Kokoro TTS, local file saving, optional S3 upload, and response formatting.
  async def process_tts_request(self, request_data): """Process a TTS request and return a JSON response.""" try: if not TTS_AVAILABLE: return { "success": False, "error": "TTS service is not available. Missing required modules." } text = request_data.get('text', '') voice = request_data.get('voice', os.environ.get('TTS_VOICE', 'af_heart')) speed = float(request_data.get('speed', 1.0)) lang = request_data.get('lang', 'en-us') filename = request_data.get('filename', None) upload_to_s3_flag = request_data.get('upload_to_s3', True) if not text: return {"success": False, "error": "No text provided"} if not filename: filename = str(uuid.uuid4()) if not filename.endswith('.mp3'): filename += '.mp3' filename = secure_filename(filename) os.makedirs(MP3_FOLDER, exist_ok=True) mp3_path = os.path.join(MP3_FOLDER, filename) mp3_filename = os.path.basename(mp3_path) print(f"Generating audio for: {text[:50]}{'...' if len(text) > 50 else ''}") print(f"Using voice: {voice}, speed: {speed}, language: {lang}") print(f"Output file: {mp3_path}") loop = asyncio.get_running_loop() try: # Attempt primary parameter format result = await loop.run_in_executor( None, lambda: tts_service.generate_audio( text=text, output_file=mp3_path, voice=voice, speed=speed, lang=lang ) ) if isinstance(result, dict) and not result.get('success', True): print(f"TTS service returned an error: {result}") # Log the result for debugging return { "success": False, "error": result.get('error', 'Unknown TTS generation error'), "tts_result": result, # Include full TTS service response "request_params": { "text": text, "voice": voice, "speed": speed, "lang": lang, "filename": filename }, "timestamp": datetime.datetime.now().isoformat() } except TypeError as e: print(f"TypeError in TTS service call: {e}") print("Trying alternative parameter format...") result = await loop.run_in_executor( None, lambda: tts_service.generate_audio( text, mp3_path, voice=voice, speed=speed ) ) if not os.path.exists(mp3_path): return { "success": False, "error": "Failed to generate audio file" } file_size = os.path.getsize(mp3_path) print(f"Audio generated successfully. File size: {file_size} bytes") response_data = { "success": True, "message": "Audio generated successfully", "filename": mp3_filename, "file_size": file_size, "path": mp3_path, "s3_uploaded": False } if upload_to_s3_flag: print(f"Uploading {mp3_filename} to S3...") s3_url = self.upload_to_s3(mp3_path, mp3_filename) if s3_url: response_data["s3_uploaded"] = True response_data["s3_url"] = s3_url # Delete local file if configured to do so if os.environ.get('DELETE_LOCAL_AFTER_S3_UPLOAD', '').lower() in ('true', '1', 'yes'): try: print(f"Removing local file {mp3_path} after successful S3 upload") os.remove(mp3_path) response_data["local_file_kept"] = False except Exception as e: print(f"Error removing local file after S3 upload: {e}") response_data["local_file_kept"] = True else: response_data["local_file_kept"] = True else: response_data["s3_uploaded"] = False response_data["s3_error"] = "S3 upload failed" return response_data except Exception as e: print(f"Error processing TTS request: {str(e)}") import traceback traceback.print_exc() return { "success": False, "error": str(e) }
• mcp-tts.py:496-514 (schema)

  Input schema defined by function parameters with type hints and defaults, plus detailed docstring describing args and return type.
  async def text_to_speech(text: str, voice: str = os.environ.get('TTS_VOICE', 'af_heart'), speed: float = float(os.environ.get('TTS_SPEED', 1.0)), lang: str = os.environ.get('TTS_LANGUAGE', 'en-us'), filename: str = None, upload_to_s3: bool = os.environ.get('S3_ENABLED', 'true').lower() == 'true') -> dict: """ Convert text to speech using the Kokoro TTS service. Args: text: The text to convert to speech voice: Voice ID to use (default: af_heart) speed: Speech speed (default: 1.0) lang: Language code (default: en-us) filename: Optional filename for the MP3 (default: auto-generated UUID) upload_to_s3: Whether to upload to S3 if enabled (default: True) Returns: A dictionary with information about the generated audio file """