converted to mcp server
This commit is contained in:
66
src/main.py
66
src/main.py
@@ -32,12 +32,12 @@ async def lifespan(app: FastAPI):
|
|||||||
|
|
||||||
app = FastAPI(title="Transcriptor", lifespan=lifespan)
|
app = FastAPI(title="Transcriptor", lifespan=lifespan)
|
||||||
mcp = FastApiMCP(app,
|
mcp = FastApiMCP(app,
|
||||||
name="Transcriptor MCP",
|
name="agent-transcriptor",
|
||||||
description="user uploads audio files (in mp3 format) and transcriptor uses AI STT model to transcribe text from files",
|
description="user uploads audio files (in mp3 format) and transcriptor uses AI STT model to transcribe text from files",
|
||||||
base_url=os.environ["MCP_BASE_URL"],
|
base_url=os.environ["MCP_BASE_URL"],
|
||||||
exclude_operations=["get_upload_page"],
|
exclude_operations=["get_upload_page"],
|
||||||
describe_all_responses=True,
|
describe_all_responses=True,
|
||||||
describe_full_response_schema=True
|
describe_full_response_schema=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -60,10 +60,14 @@ async def get_upload_page(request: Request):
|
|||||||
@app.post("/upload", operation_id="upload_single_audio_file")
|
@app.post("/upload", operation_id="upload_single_audio_file")
|
||||||
async def upload_file(file: UploadFile = File(...)):
|
async def upload_file(file: UploadFile = File(...)):
|
||||||
"""
|
"""
|
||||||
API endpoint to handle file uploads
|
Upload a single MP3/WAV audio file.
|
||||||
- Validates that the file is mp3 or wav
|
|
||||||
- Saves it to the local filesystem
|
Example curl command:
|
||||||
- Checks for duplicate filenames
|
curl -X 'POST' \
|
||||||
|
'http://0.0.0.0:33754/upload' \
|
||||||
|
-H 'accept: application/json' \
|
||||||
|
-H 'Content-Type: multipart/form-data' \
|
||||||
|
-F 'file=@How To Parallel Charge LiPo Batteries Without Burning Down Your House.mp3;type=audio/mpeg'
|
||||||
"""
|
"""
|
||||||
# Validate file extension
|
# Validate file extension
|
||||||
if not is_valid_file(file.filename):
|
if not is_valid_file(file.filename):
|
||||||
@@ -90,9 +94,12 @@ async def upload_file(file: UploadFile = File(...)):
|
|||||||
@app.post("/upload-multiple", operation_id="upload_multiple_audio_files")
|
@app.post("/upload-multiple", operation_id="upload_multiple_audio_files")
|
||||||
async def upload_multiple_files(files: List[UploadFile] = File(...)):
|
async def upload_multiple_files(files: List[UploadFile] = File(...)):
|
||||||
"""
|
"""
|
||||||
API endpoint to handle multiple file uploads
|
Upload multiple audio files simultaneously.
|
||||||
- Processes each file individually
|
|
||||||
- Returns a summary of the upload results
|
Parameters:
|
||||||
|
- files: List of audio files (required)
|
||||||
|
|
||||||
|
Returns: Summary of upload results for each file
|
||||||
"""
|
"""
|
||||||
results = []
|
results = []
|
||||||
|
|
||||||
@@ -152,7 +159,12 @@ async def upload_multiple_files(files: List[UploadFile] = File(...)):
|
|||||||
@app.post("/process-multiple", operation_id="transcribe_multiple_files_in_batch")
|
@app.post("/process-multiple", operation_id="transcribe_multiple_files_in_batch")
|
||||||
async def process_multiple_files(filenames: List[str] = Form(...)):
|
async def process_multiple_files(filenames: List[str] = Form(...)):
|
||||||
"""
|
"""
|
||||||
API endpoint to process multiple files at once
|
Batch transcribe multiple audio files by filename.
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
- filenames: List of audio file names to process
|
||||||
|
|
||||||
|
Returns: Batch transcription results
|
||||||
"""
|
"""
|
||||||
results = []
|
results = []
|
||||||
|
|
||||||
@@ -205,7 +217,12 @@ async def process_multiple_files(filenames: List[str] = Form(...)):
|
|||||||
|
|
||||||
|
|
||||||
def get_file_list():
|
def get_file_list():
|
||||||
"""Helper function to get file list with metadata and status"""
|
"""
|
||||||
|
Retrieve audio files with metadata and processing status.
|
||||||
|
|
||||||
|
Returns: List of file objects containing name, size, creation time,
|
||||||
|
processing status, and transcript availability
|
||||||
|
"""
|
||||||
files = []
|
files = []
|
||||||
for file_path in UPLOAD_DIR.iterdir():
|
for file_path in UPLOAD_DIR.iterdir():
|
||||||
if file_path.is_file() and file_path.name != '.gitkeep': # Skip .gitkeep file
|
if file_path.is_file() and file_path.name != '.gitkeep': # Skip .gitkeep file
|
||||||
@@ -230,15 +247,28 @@ def get_file_list():
|
|||||||
return files
|
return files
|
||||||
|
|
||||||
|
|
||||||
@app.get("/files", operation_id="list_already_uploaded_files")
|
@app.get("/files", operation_id="list_uploaded_files")
|
||||||
async def list_files():
|
async def list_files():
|
||||||
"""API endpoint to list all uploaded files"""
|
"""
|
||||||
|
Retrieve list of all uploaded audio files with metadata.
|
||||||
|
|
||||||
|
Returns: Object containing files array with status information
|
||||||
|
"""
|
||||||
return {"files": get_file_list()}
|
return {"files": get_file_list()}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/process/{filename}", operation_id="transcribe_single_file")
|
@app.post("/process/{filename}", operation_id="transcribe_single_file")
|
||||||
async def process_file(filename: str):
|
async def process_file(filename: str):
|
||||||
"""API endpoint to manually trigger file processing"""
|
"""
|
||||||
|
Transcribe a single audio file by filename.
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
- filename: Name of the audio file to process
|
||||||
|
|
||||||
|
Returns: Processing status information
|
||||||
|
|
||||||
|
Raises: 404 (file not found), 500 (processing failed)
|
||||||
|
"""
|
||||||
file_path = UPLOAD_DIR / filename
|
file_path = UPLOAD_DIR / filename
|
||||||
|
|
||||||
# Check if file exists
|
# Check if file exists
|
||||||
@@ -259,7 +289,11 @@ async def process_file(filename: str):
|
|||||||
@app.get("/download-transcripts", operation_id="download_all_transcripts")
|
@app.get("/download-transcripts", operation_id="download_all_transcripts")
|
||||||
async def download_transcripts():
|
async def download_transcripts():
|
||||||
"""
|
"""
|
||||||
API endpoint to download all transcript files as a single ZIP
|
Download all available transcripts as a ZIP archive.
|
||||||
|
|
||||||
|
Returns: ZIP file stream containing all transcript files
|
||||||
|
|
||||||
|
Raises: 404 (no transcripts available)
|
||||||
"""
|
"""
|
||||||
# Check if there are any transcripts
|
# Check if there are any transcripts
|
||||||
transcript_files = list(TRANSCRIPT_DIR.glob("*.txt"))
|
transcript_files = list(TRANSCRIPT_DIR.glob("*.txt"))
|
||||||
@@ -319,4 +353,4 @@ async def broadcast_file_list():
|
|||||||
logger.error(f"Error broadcasting to a client: {e}")
|
logger.error(f"Error broadcasting to a client: {e}")
|
||||||
|
|
||||||
|
|
||||||
# Run with: uvicorn src.main:app --reload
|
mcp.setup_server()
|
||||||
|
|||||||
Reference in New Issue
Block a user