Spaces:
Running
Running
| from fastapi import FastAPI, Request | |
| import src.Paraphrase as Paraphrase | |
| import src.Translate as Translate | |
| from typing import Optional | |
| from fastapi_mcp import FastApiMCP | |
| 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'} | |
| 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''' | |
| <html> | |
| <head> | |
| <title>FastAPI with MCP</title> | |
| </head> | |
| <body> | |
| <h2>FastAPI URLS</h2> | |
| <p><a href="{host_url}" target="_blank">Host URL:</a> {host_url}</p> | |
| <p><a href="{host_url}/docs" target="_blank">DOCS</a></p> | |
| <p><a href="{host_url}/redoc" target="_blank">REDOC</a></p> | |
| <p><a href="{host_url}/openapi.json" target="_blank">openapi.json</a></p> | |
| <p><a href="{host_url}/mcp" target="_blank">MCP</a></p> | |
| <p>MCP configuration: {mcp_config}</a></p> | |
| <p>MODELS: {list(MODELS.values())}"</p> | |
| </body> | |
| </html> | |
| ''' | |
| 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} | |
| def paraphrase(text: str, model: str = MODELS['paraphrase']): | |
| resultValue, exception = Paraphrase.paraphraseParaphraseMethod(text, model) | |
| return {"input": text, "result": resultValue, "exception": exception} | |
| 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} | |
| def bergamot(input_text: str, sl: str = 'de', tl: str = 'en', model: Optional[str] = 'bergamot'): | |
| try: | |
| import bergamot | |
| config = bergamot.ServiceConfig(numWorkers=4) | |
| service = bergamot.Service(config) | |
| models = service.modelFromConfigPath("./deen/bergamot.config.yml") | |
| options = bergamot.ResponseOptions(alignment=False, qualityScores=False, HTML=False) | |
| response = service.translate(models, bergamot.VectorString([input_text]), options) | |
| print(response, type(response), len(response), [r.target.text for r in response], next(iter(response)).target.text) | |
| response = [r.target.text for r in response] if len(response) > 1 else next(iter(response)).target.text | |
| # response = [r.target.text for r in model_response][0] if isinstance(response, bergamot._bergamot.VectorResponse) else next(iter(response)).target.text | |
| # bergamot._bergamot.VectorResponse is an iterable of bergamot._bergamot.Response | |
| except Exception as error: | |
| response = error | |
| return {"input": input_text, "result": response, "model": str(model)} | |
| # 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"] | |
| ) | |
| # Mount the MCP server directly to the FASTAPI app | |
| mcp.mount() |