from .utils import catch_error_and_return, base64_decode
from dotenv import load_dotenv
from fastmcp import FastMCP
import requests
import tempfile
import os
load_dotenv()
transcribe_mcp = FastMCP(name="Transcribe server")
@transcribe_mcp.tool
@catch_error_and_return
def transcribe_audio(
file_base64: str,
language: str = "ru",
diarize: bool = False,
diarize_speakers_num: int = 1,
audio_format: str = ".mp3",
):
"""
Transcribe audio
:param file_base64: fill base64 string for audio
:param audio_format: audio format (optional)
:param diarize: Dialogue diarization (optional)
:param diarize_speakers_num: number of speakers (optional)
:param language: audio language (optional)
:return: message about starting transcribe
"""
url = os.getenv("TRANSCRIBE_UPLOAD_AUDIO")
api_key = os.getenv("TRANSCRIBE_API_KEY")
headers = {
"authorization": f"Bearer {api_key}",
}
data = {
"model": "large-v3",
"language": language,
"diarize": diarize,
"diarize_speakers_num": diarize_speakers_num,
}
fp = tempfile.TemporaryFile(mode="wb", suffix=audio_format, delete_on_close=False)
fp.write(base64_decode(file_base64))
fp.close()
f = open(fp.name, "rb")
file_send = {"file": f}
result = requests.post(
url,
data=data,
files=file_send,
headers=headers,
)
f.close()
os.remove(fp.name)
return result.json()
@transcribe_mcp.tool
@catch_error_and_return
def transcribe_get_status(task_id: str) -> dict:
"""
Get transcribe status
:param task_id: Task id
:return: transcribe status
"""
url = os.getenv("TRANSCRIBE_GET_STATUS").replace("{task_id}", task_id)
api_key = os.getenv("TRANSCRIBE_API_KEY")
headers = {
"authorization": f"Bearer {api_key}",
}
result = requests.get(url=url, headers=headers).json()
return result
@transcribe_mcp.tool
@catch_error_and_return
def transcribe_get_result(task_id: str):
"""
Get transcribe result
:param task_id: Task id
:return: transcribe result
"""
url = os.getenv("TRANSCRIBE_GET_TRANSCRIBE").replace("{task_id}", task_id)
api_key = os.getenv("TRANSCRIBE_API_KEY")
headers = {
"authorization": f"Bearer {api_key}",
}
result = requests.get(url=url, headers=headers).json()
return result
@transcribe_mcp.tool
def health_check():
"""
health check for server
:return: server status
"""
url = os.getenv("TRANSCRIBE_HEALTH_URL")
api_key = os.getenv("TRANSCRIBE_API_KEY")
headers = {
"Authorization": f"Bearer {api_key}",
}
try:
result = requests.get(url, headers=headers)
result.raise_for_status()
return {"status": "ok"}
except Exception as ex:
return {"status": f"Error {ex}"}
