From 332c88f99dea72bc5e6f6e21b503bdea0707b377 Mon Sep 17 00:00:00 2001 From: admin Date: Sat, 20 Jun 2026 01:19:35 +0000 Subject: [PATCH] Update MCP server to expose clear JSON tool schema - Use Pydantic models and explicit operation_ids. - No multipart; accept audio_base64, audio_url, or audio_path via JSON. - Ensure OpenAPI spec is fully self-describing for MCP tool generation. --- scraibe/mcp_server.py | 136 +++++++++++++++++++++++++++++------------- 1 file changed, 94 insertions(+), 42 deletions(-) diff --git a/scraibe/mcp_server.py b/scraibe/mcp_server.py index 0b8d6be..263a8dd 100644 --- a/scraibe/mcp_server.py +++ b/scraibe/mcp_server.py @@ -2,7 +2,7 @@ MCP-style HTTP server for ScrAIbe. - Exposes an OpenAPI-compliant endpoint for external LLMs to: - - Upload audio (as base64 or via URL) in a JSON body. + - Submit audio (as base64, URL, or internal path) via JSON. - Receive transcript JSON (no summary). - WebUI remains always enabled; this is additive. @@ -18,13 +18,12 @@ import os import time import uuid import base64 -import tempfile import logging from typing import Optional import httpx from fastapi import FastAPI, HTTPException -from pydantic import BaseModel +from pydantic import BaseModel, Field from .autotranscript import Scraibe @@ -35,8 +34,11 @@ app = FastAPI( version="0.1.0", description=( "MCP-style HTTP API for ScrAIbe. " - "Allows external LLMs to upload audio and receive transcript JSON." + "Allows external LLMs to submit audio and receive transcript JSON." ), + openapi_tags=[ + {"name": "transcription", "description": "Transcription endpoints"} + ], ) # In-memory job store for MCP (simple; can be replaced with Redis later) @@ -44,11 +46,49 @@ _mcp_jobs: dict = {} class TranscribeRequest(BaseModel): - audio_base64: Optional[str] = None - audio_url: Optional[str] = None - audio_path: Optional[str] = None - language: Optional[str] = None - num_speakers: Optional[int] = None + """ + Input for transcription. + + Exactly one of audio_base64, audio_url, or audio_path must be provided. + """ + audio_base64: Optional[str] = Field( + None, + description="Base64-encoded audio file content." + ) + audio_url: Optional[str] = Field( + None, + description="Public URL to the audio file." + ) + audio_path: Optional[str] = Field( + None, + description="Internal file path on the server (for internal use only)." + ) + language: Optional[str] = Field( + None, + description="Optional language hint (e.g., 'english', 'german')." + ) + num_speakers: Optional[int] = Field( + None, + description="Optional number of speakers for diarization." + ) + + +class TranscribeResponse(BaseModel): + job_id: str + status: str + message: str + + +class JobStatusResponse(BaseModel): + job_id: str + status: str + message: str + + +class TranscriptJSONResponse(BaseModel): + job_id: str + transcript: str + segments: list def _job_id() -> str: @@ -64,7 +104,6 @@ def _save_audio_from_request(req: TranscribeRequest) -> str: os.makedirs(upload_dir, exist_ok=True) if req.audio_base64: - # base64 upload try: data = base64.b64decode(req.audio_base64) except Exception as e: @@ -78,7 +117,6 @@ def _save_audio_from_request(req: TranscribeRequest) -> str: return file_path if req.audio_url: - # download from URL try: with httpx.stream("GET", req.audio_url, timeout=60) as resp: if resp.status_code != 200: @@ -99,7 +137,6 @@ def _save_audio_from_request(req: TranscribeRequest) -> str: raise HTTPException(status_code=400, detail=f"Error downloading audio: {e}") if req.audio_path: - # use provided path (for internal use) path = req.audio_path if not os.path.isfile(path): raise HTTPException(status_code=400, detail="audio_path does not exist") @@ -107,16 +144,21 @@ def _save_audio_from_request(req: TranscribeRequest) -> str: raise HTTPException( status_code=400, - detail="Provide one of: audio_base64, audio_url, or audio_path", + detail="Provide exactly one of: audio_base64, audio_url, or audio_path", ) -@app.get("/health") +@app.get("/health", tags=["transcription"]) async def health(): return {"status": "ok"} -@app.post("/transcribe") +@app.post( + "/transcribe", + tags=["transcription"], + operation_id="transcribe", + response_model=TranscribeResponse, +) async def transcribe(req: TranscribeRequest): """ Submit an audio file for transcription. @@ -172,22 +214,22 @@ async def transcribe(req: TranscribeRequest): "message": f"Error enqueuing job: {e}", "file_path": file_path, } - return { - "job_id": job_id, - "status": "error", - "message": _mcp_jobs[job_id]["message"], - } + return TranscribeResponse( + job_id=job_id, + status="error", + message=_mcp_jobs[job_id]["message"], + ) _mcp_jobs[job_id] = { "status": "queued", "message": "Job queued for processing.", "file_path": file_path, } - return { - "job_id": job_id, - "status": "queued", - "message": _mcp_jobs[job_id]["message"], - } + return TranscribeResponse( + job_id=job_id, + status="queued", + message=_mcp_jobs[job_id]["message"], + ) # Synchronous path _mcp_jobs[job_id] = { @@ -221,26 +263,36 @@ async def transcribe(req: TranscribeRequest): t = threading.Thread(target=_run_sync, daemon=True) t.start() - return { - "job_id": job_id, - "status": "processing", - "message": _mcp_jobs[job_id]["message"], - } + return TranscribeResponse( + job_id=job_id, + status="processing", + message=_mcp_jobs[job_id]["message"], + ) -@app.get("/transcribe/{job_id}/status") +@app.get( + "/transcribe/{job_id}/status", + tags=["transcription"], + operation_id="get_status", + response_model=JobStatusResponse, +) async def get_status(job_id: str): job = _mcp_jobs.get(job_id) if not job: raise HTTPException(status_code=404, detail="Job not found") - return { - "job_id": job_id, - "status": job["status"], - "message": job.get("message", ""), - } + return JobStatusResponse( + job_id=job_id, + status=job["status"], + message=job.get("message", ""), + ) -@app.get("/transcribe/{job_id}/json") +@app.get( + "/transcribe/{job_id}/json", + tags=["transcription"], + operation_id="get_json", + response_model=TranscriptJSONResponse, +) async def get_json(job_id: str): job = _mcp_jobs.get(job_id) if not job: @@ -255,8 +307,8 @@ async def get_json(job_id: str): transcript_text = job.get("transcript", "") segments = job.get("segments", []) - return { - "job_id": job_id, - "transcript": transcript_text, - "segments": segments, - } + return TranscriptJSONResponse( + job_id=job_id, + transcript=transcript_text, + segments=segments, + )