"""Levelang MCP Server -- tools and resources."""
from __future__ import annotations
import httpx
from mcp.server.fastmcp import FastMCP
from .client import LevelangClient
from .config import get_settings
from .formatting import format_language_detail, format_language_list, format_translation
def _sanitize_text(text: str) -> str:
"""Light sanitization of input text before sending to the backend.
Strips leading/trailing whitespace only. Does NOT collapse internal
newlines -- multi-line input (poems, paragraphs) is legitimate and
should be preserved. The backend is responsible for handling newlines
safely when constructing LLM prompts and parsing responses.
"""
return text.strip()
settings = get_settings()
mcp = FastMCP(
"Levelang",
instructions="AI-powered translation at learner-appropriate proficiency levels. "
"Use the translate tool to translate text at a specific proficiency level, "
"and list_languages to discover available languages, levels, and moods.",
port=settings.mcp_port,
)
levelang = LevelangClient()
# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------
@mcp.tool()
async def translate(
text: str,
target_language: str,
level: str,
source_language: str = "eng",
mood: str = "casual",
) -> str:
"""Translate text to a target language at a specific proficiency level.
Unlike standard translators that produce native-speaker complexity,
this translates at the learner's level -- beginner translations use
simple grammar, intermediate uses more complex structures, etc.
Args:
text: The text to translate (any length, any source language)
target_language: Target language code -- use list_languages to see
available codes (e.g. fra, deu, cmn, yue, ita)
level: Proficiency level -- proficiency levels available for the target language (e.g. beginner, intermediate, advanced, and/or fluent)
source_language: Source language code (default: eng for English)
mood: Tone -- tones available for the target language
Returns:
The translated text with metadata about the translation.
"""
try:
result = await levelang.translate(
text=_sanitize_text(text),
source_language_code=source_language,
target_language_code=target_language,
level=level,
mood=mood,
)
return format_translation(result)
except httpx.HTTPStatusError as e:
if e.response.status_code == 422:
detail = e.response.json().get("detail", "Validation error")
return f"Invalid request: {detail}"
elif e.response.status_code == 429:
return "Rate limit reached. Please wait a moment and try again."
elif e.response.status_code >= 500:
return "Translation service is temporarily unavailable. Please try again."
else:
return f"Backend error (HTTP {e.response.status_code}): {e.response.text}"
except httpx.TimeoutException:
return "Translation request timed out. The backend may be under heavy load."
except httpx.ConnectError:
return "Cannot reach the Levelang backend. Check that the service is running."
except Exception as e:
return f"Unexpected error: {e}"
@mcp.tool()
async def list_languages() -> str:
"""List all languages supported by Levelang with their available levels and moods.
Use this to discover valid language codes, proficiency levels, and mood
options before calling the translate tool.
Returns:
Formatted list of supported languages and their configurations.
"""
try:
result = await levelang.get_languages()
return format_language_list(result)
except httpx.HTTPStatusError as e:
if e.response.status_code >= 500:
return "Language service is temporarily unavailable. Please try again."
return f"Backend error (HTTP {e.response.status_code}): {e.response.text}"
except httpx.TimeoutException:
return "Request timed out while fetching languages."
except httpx.ConnectError:
return "Cannot reach the Levelang backend. Check that the service is running."
except Exception as e:
return f"Unexpected error: {e}"
# ---------------------------------------------------------------------------
# Resources
# ---------------------------------------------------------------------------
@mcp.resource("levelang://languages")
async def languages_resource() -> str:
"""List of all supported languages with their levels and moods."""
try:
result = await levelang.get_languages()
return format_language_list(result)
except Exception:
return "Unable to fetch language list from the backend."
@mcp.resource("levelang://languages/{language_code}")
async def language_detail_resource(language_code: str) -> str:
"""Detailed configuration for a specific language including level descriptions."""
try:
result = await levelang.get_language(language_code)
return format_language_detail(result)
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
return f"Language '{language_code}' not found."
return f"Error fetching language details: {e.response.text}"
except Exception:
return f"Unable to fetch details for language '{language_code}'."
