forked from ci4s/ai4mats-mcp-tools
88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
import os
|
||
import json
|
||
import httpx
|
||
from mcp.server.fastmcp import FastMCP
|
||
|
||
# ============================================
|
||
# 配置
|
||
# ============================================
|
||
def load_config():
|
||
config_path = os.path.join(os.path.dirname(__file__), "../config/config.json")
|
||
default_config = {
|
||
"BAYESIAN_API_BASE_URL": "http://218.77.58.19:22222"
|
||
}
|
||
|
||
try:
|
||
if os.path.exists(config_path):
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
default_config.update(json.load(f))
|
||
except Exception:
|
||
pass
|
||
|
||
return default_config
|
||
|
||
|
||
config = load_config()
|
||
API_BASE_URL = config["BAYESIAN_API_BASE_URL"]
|
||
|
||
# ============================================
|
||
# MCP 服务定义
|
||
# ============================================
|
||
mcp = FastMCP("TEM-Analysis")
|
||
|
||
|
||
# ============================================
|
||
# 工具：TEM 图像分析
|
||
# ============================================
|
||
@mcp.tool()
|
||
async def analyze_tem_image(
|
||
filename: str,
|
||
content_base64: str,
|
||
model: str = "/detection/predict"
|
||
) -> str:
|
||
"""
|
||
调用 TEM（透射电子显微镜）图像分析模型。
|
||
|
||
参数说明：
|
||
- filename: 图片文件名（例如："sample_001.jpg"）
|
||
- content_base64: 图片 Base64 内容（不含 data URI 前缀）
|
||
- model: 模型路径（默认："/detection/predict"）
|
||
|
||
返回：
|
||
TEM 分析结果（通常为检测到的缺陷、结构或分类信息）
|
||
"""
|
||
|
||
api_url = f"{API_BASE_URL}/svc/tem/analysis"
|
||
|
||
payload = {
|
||
"model": model,
|
||
"filename": filename,
|
||
"content_base64": content_base64
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=120.0) as client:
|
||
response = await client.post(api_url, json=payload)
|
||
response.raise_for_status()
|
||
|
||
try:
|
||
return json.dumps(
|
||
response.json(),
|
||
indent=2,
|
||
ensure_ascii=False
|
||
)
|
||
except:
|
||
return response.text
|
||
|
||
except httpx.HTTPStatusError as e:
|
||
return f"❌ TEM 分析接口错误 ({e.response.status_code})"
|
||
except httpx.RequestError:
|
||
return "❌ 无法连接 TEM 分析服务"
|
||
except Exception as e:
|
||
return f"❌ 调用失败：{e}"
|
||
|
||
|
||
# ============================================
|
||
# ⚠️ 不写 mcp.run()
|
||
# 由主网关统一启动
|
||
# ============================================ |