from fastapi import FastAPI, Request, Query import src.Paraphrase as Paraphrase import src.Translate as Translate from typing import Optional from fastapi_mcp import FastApiMCP from huggingface_hub import hf_hub_download, list_repo_files from sentence_transformers import SentenceTransformer app = FastAPI() # app = FastAPI(docs_url="/docs") MODELS = {'enro': 'BlackKakapo/opus-mt-en-ro', 'roen': 'BlackKakapo/opus-mt-ro-en', 'gemma': 'Gargaz/gemma-2b-romanian-better', 'paraphrase': 'tuner007/pegasus_paraphrase'} EMBEDDING_MODELS = {"all-MiniLM-L6-v2":384, "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2":384, "sentence-transformers/distiluse-base-multilingual-cased-v2":512, "sentence-transformers/stsb-xlm-r-multilingual":768, "sentence-transformers/use-cmlm-multilingual":768, "sentence-transformers/paraphrase-multilingual-mpnet-base-v2":768} EMBEDDING_MODEL = "sentence-transformers/distiluse-base-multilingual-cased-v2" @app.get("/") def index(request: Request): from fastapi.responses import HTMLResponse host_url = "https://" + request.url.netloc mcp_config = '''{"mcpServers": {"fastapi-mcp": {"url": "https://tiberiucristianleon-fastapimt.hf.space/mcp"}}}''' html_content = f'''
Host URL: {host_url}
MCP configuration: {mcp_config}
MODELS: {list(MODELS.values())}"
''' return HTMLResponse(content=html_content) # @app.get("/") # async def get_host_url(request: Request): # host_url = request.url.scheme + "s://" + request.url.netloc # return {"host_url": host_url, 'endpoints': ['/paraphrase', '/translate', f'{host_url}/docs', f'{host_url}/redoc', f'{host_url}/openapi.json'], 'models': MODELS} @app.get("/paraphrase", operation_id="get_paraphrase", description="Paraphrase text", tags=["paraphrase"], summary="Paraphrase text") def paraphrase(text: str, model: str = MODELS['paraphrase']): resultValue, exception = Paraphrase.paraphraseParaphraseMethod(text, model) return {"input": text, "result": resultValue, "exception": exception} @app.get("/translate", operation_id="get_translate", description="Translate text", tags=["translate"], summary="Translate text") def translate(text: str, model: Optional[str] = MODELS['enro']): if 'BlackKakapo' in model: translation = Translate.paraphraseTranslateMethod(text, model) else: translation: str = Translate.gemma_direct(text, model) return {"input": text, "result": translation, "model": model} # Keep track of installed (src, tgt) pairs installed_pairs = set() # https://tiberiucristianleon-fastapimt.hf.space/bergamot?input_text=das%20ist%20keine%20gute%20Frau&input_text=das%20ist%20eine%20gute%20Nachricht&sl=de&tl=en&model=bergamot @app.get("/bergamot", operation_id="get_bergamot", description="Translate text with Bergamot", tags=["bergamot"], summary="Translate text with Bergamot") def bergamot(input_text: list[str] = Query(description="Input list of strings"), sl: str = 'de', tl: str = 'en', model_name: Optional[str] = 'deen'): """ Translates the input text from the source language to the target language using a specified model. Parameters: input_text (str): The source text to be translated sl (str): The source language of the input text tl (str): The target language into which the input text is translated model_name (str): The selected translation model name Returns: dict: input_text(str): The input text in the source language translated_text(str): The input text translated to the selected target language message_text(str): A descriptive message summarizing the translation process. Example: "Translated from English to German with ende." Example: >>> bergamot("Hello world", "en", "de", "ende") {"input_text": "Hello world", "translated_text": "Hallo Welt", "message_text": "Translated from English to German with ende."} """ try: import bergamot # print(type(input_text)) # input_text = [input_text] if isinstance(input_text, str) else input_text config = bergamot.ServiceConfig(numWorkers=4) service = bergamot.Service(config) repo_id="TiberiuCristianLeon/Bergamot" branches = ['base', 'base-memory', 'tiny'] subfolder = f"{sl}{tl}" # List all files in the repo all_files = list_repo_files(repo_id, repo_type='model') print(all_files, len(all_files)) for branch in branches: branch_files = [f for f in all_files if f.startswith(branch)] model_files = [f for f in branch_files if f.startswith(model_name)] print(model_files) for file_path in model_files: if subfolder not in file_path: local_path = hf_hub_download(repo_id=repo_id, filename=file_path) print(f"Downloaded to: {local_path}") installed_pairs.add(subfolder) model = service.modelFromConfigPath(f"./{model_name}/config.yml") options = bergamot.ResponseOptions(alignment=False, qualityScores=False, HTML=False) rawresponse = service.translate(model, bergamot.VectorString(input_text), options) response: list|str = [r.target.text for r in rawresponse] if len(rawresponse) > 1 else next(iter(rawresponse)).target.text print(type(input_text), len(input_text), len(rawresponse), type(response), response) # response = [r.target.text for r in model_response][0] if isinstance(response, bergamot._bergamot.VectorResponse) else next(iter(response)).target.text # response is of type bergamot._bergamot.VectorResponse, an iterable of bergamot._bergamot.Response message_text = f"Translated from {sl} to {tl} with {model_name}." except Exception as error: response, message_text = error, error return {"input": input_text, "translated_text": response, "message_text": message_text} @app.get("/embed", operation_id="get_embeddings", description="Embed text", tags=["embed"], summary="Embed text") def embed(text: str, model: str = EMBEDDING_MODEL): model = SentenceTransformer(model) embeddings = model.encode(text) print(embeddings.shape, len(embeddings)) # similarities = model.similarity(embeddings, embeddings) return {"input": text, "embeddings": embeddings.tolist(), "shape": embeddings.shape} # Create an MCP server based on this app mcp = FastApiMCP( app, name="Translate and paraphrase FASTAPI MCP", description="MCP server to translate and paraphrase text", describe_all_responses=True, describe_full_response_schema=True, include_operations=["get_translate", "get_paraphrase"], include_tags=["paraphrase", "translate", "bergamot"] ) # Mount the MCP server directly to the FASTAPI app mcp.mount()