Spaces:
Sleeping
Sleeping
File size: 4,233 Bytes
c1dac64 9f0a178 f7cc9ad 8b05c02 d120873 e46dbee 75255bc d120873 c1dac64 4e5cd58 eaec419 d3a5771 4e5cd58 eaec419 4e5cd58 53869cd 1d71a28 a696db8 c3a14f0 d3a5771 75255bc 4e5cd58 1d71a28 75255bc d120873 42c14b6 75255bc 698fb3f 75255bc d120873 42c14b6 75255bc 7db8a97 c3a14f0 7db8a97 c3a14f0 75255bc 4e95015 678e936 bf15f3f eec15ee f873413 678e936 f873413 678e936 44b38c0 2359c6a fad77a2 0519180 fad77a2 eec15ee 272e2ae 4c7bc2b 4e95015 8b05c02 1d71a28 8b05c02 27ced65 c832cd2 53869cd 8b05c02 a696db8 4e95015 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
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'}
@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'''
<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}
@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}
@app.get("/bergamot", operation_id="get_bergamot", description="Translate text with Bergamot", tags=["bergamot"], summary="Translate text with Bergamot")
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() |