#!/usr/bin/env python3
|
"""Standalone translation MCP. Only exposes translate. Does not call ConnectionHandler.chat."""
|
|
from __future__ import annotations
|
|
import os
|
from pathlib import Path
|
|
import httpx
|
import yaml
|
from mcp.server.fastmcp import FastMCP
|
|
ROOT = Path(__file__).resolve().parents[2]
|
TRANSLATOR_PROMPT = (
|
"你是一台翻译机。只输出译文，不要解释、不要拼音课、不要追问、不要闲聊。"
|
"未指定语言时：中文译成英文，英文译成中文；其它语言默认译成中文。"
|
)
|
|
mcp = FastMCP("english-translate")
|
|
|
def _read_yaml(path: Path) -> dict:
|
if not path.is_file():
|
return {}
|
with path.open("r", encoding="utf-8") as f:
|
data = yaml.safe_load(f) or {}
|
return data if isinstance(data, dict) else {}
|
|
|
def _load_llm_config() -> dict:
|
default_cfg = _read_yaml(ROOT / "config.yaml")
|
custom_cfg = _read_yaml(ROOT / "data" / ".config.yaml")
|
|
def merge(base, overlay):
|
out = dict(base)
|
for k, v in overlay.items():
|
if isinstance(v, dict) and isinstance(out.get(k), dict):
|
out[k] = merge(out[k], v)
|
else:
|
out[k] = v
|
return out
|
|
cfg = merge(default_cfg, custom_cfg)
|
selected = (cfg.get("selected_module") or {}).get("LLM")
|
llm_map = cfg.get("LLM") or {}
|
if selected and selected in llm_map:
|
item = dict(llm_map[selected])
|
item["_name"] = selected
|
return item
|
if llm_map:
|
name, item = next(iter(llm_map.items()))
|
data = dict(item)
|
data["_name"] = name
|
return data
|
return {}
|
|
|
def _chat_complete(text: str, source_lang: str, target_lang: str) -> str:
|
llm = _load_llm_config()
|
api_key = llm.get("api_key") or os.environ.get("OPENAI_API_KEY") or ""
|
if api_key and ("你的" in str(api_key) or "key" == str(api_key).lower()):
|
api_key = os.environ.get("OPENAI_API_KEY") or api_key
|
base_url = llm.get("base_url") or llm.get("url") or "https://api.openai.com/v1"
|
model = llm.get("model_name") or llm.get("model") or "gpt-4o-mini"
|
if not api_key or "你的" in str(api_key):
|
raise RuntimeError("翻译 MCP 未配置可用的 LLM api_key")
|
|
extra = ""
|
if source_lang or target_lang:
|
extra = f"\n源语言:{source_lang or 'auto'} 目标语言:{target_lang or 'auto'}"
|
user = f"翻译下面内容，只输出译文：{extra}\n{text}"
|
|
url = str(base_url).rstrip("/")
|
if not url.endswith("/chat/completions"):
|
if url.endswith("/v1"):
|
url = url + "/chat/completions"
|
else:
|
url = url + "/v1/chat/completions"
|
|
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
payload = {
|
"model": model,
|
"temperature": 0.1,
|
"messages": [
|
{"role": "system", "content": TRANSLATOR_PROMPT},
|
{"role": "user", "content": user},
|
],
|
}
|
with httpx.Client(timeout=60.0) as client:
|
resp = client.post(url, headers=headers, json=payload)
|
resp.raise_for_status()
|
data = resp.json()
|
choices = data.get("choices") or []
|
if not choices:
|
raise RuntimeError("翻译接口没有返回内容")
|
msg = (choices[0].get("message") or {}).get("content") or ""
|
return str(msg).strip()
|
|
|
@mcp.tool()
|
def translate(text: str, source_lang: str = "", target_lang: str = "") -> str:
|
"""Translate text. Chinese->English, English->Chinese unless languages are set. Return translation only."""
|
if not text or not str(text).strip():
|
return ""
|
return _chat_complete(str(text).strip(), source_lang or "", target_lang or "")
|
|
|
if __name__ == "__main__":
|
os.chdir(ROOT)
|
mcp.run(transport="stdio")
|
