import contextlib
import logging
import os
from typing import Any, Dict, Callable
from fastapi import FastAPI
from fastapi.responses import JSONResponse, Response
import mcp.types as types
from mcp.server import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.shared.exceptions import McpError
# api/index.py의 최신 15개 Tool 정의를 재사용하여 동기화
from api.index import TOOLS as API_TOOLS, TOOL_HANDLERS as API_TOOL_HANDLERS
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)
logger = logging.getLogger("semiprocess")
MCP_SPEC_VERSION = "2026-01-14"
server = Server("SemiProcess", version=MCP_SPEC_VERSION)
# API Tool -> MCP Tool 변환 (inputSchema -> schema, sync -> async TextContent)
TOOLS: Dict[str, Dict[str, Any]] = {}
def _wrap_handler(fn: Callable[..., str]) -> Callable[..., types.TextContent]:
async def _handler(**kwargs: Any) -> types.TextContent:
result = fn(**kwargs)
return types.TextContent(type="text/markdown", text=result)
return _handler
for tool in API_TOOLS:
name = tool["name"]
TOOLS[name] = {
"description": tool.get("description", ""),
"schema": tool.get("inputSchema", {}),
"handler": _wrap_handler(API_TOOL_HANDLERS[name]),
}
@server.list_tools()
async def list_tools(request: types.ListToolsRequest) -> types.ListToolsResult:
return types.ListToolsResult(
tools=[
types.Tool.model_validate(
{
"name": name,
"description": tool["description"],
"inputSchema": tool["schema"],
}
)
for name, tool in TOOLS.items()
]
)
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any] | None) -> list[types.Content]:
tool = TOOLS.get(name)
if not tool:
raise McpError(f"Unknown tool '{name}'")
handler = tool["handler"]
args = arguments or {}
try:
result = await handler(**args)
except TypeError as exc:
raise McpError(f"Invalid arguments for tool '{name}': {exc}") from exc
return [result]
def _create_session_manager() -> StreamableHTTPSessionManager:
# 요청마다 새 인스턴스를 생성해 run() 중복 호출 문제를 피한다.
return StreamableHTTPSessionManager(
server,
json_response=True, # allow JSON responses (PlayMCP 호환)
stateless=True, # stateless per MCP spec
)
# lifespan 용 기본 매니저 (현재는 사용하지 않지만 초기화 에러 방지)
session_manager = _create_session_manager()
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
# Vercel에서는 lifespan이 보장되지 않으므로 no-op
yield
app = FastAPI(
title="SemiProcess MCP Server",
version=MCP_SPEC_VERSION,
description="MCP server for semiconductor process management",
lifespan=lifespan,
)
# Disable trailing-slash redirects to avoid 307 on /mcp
app.router.redirect_slashes = False
# Disable automatic slash redirects to avoid 307 on /mcp
app.router.redirect_slashes = False
@app.middleware("http")
async def restore_original_path(request, call_next):
"""
Vercel rewrite 시 원본 경로를 복원한다. /api/index.py 접두어 제거 + 헤더 기반 복원.
"""
path = request.scope.get("path", "")
original_path = path
# 1) vercel python runtime가 /api/index.py 접두어를 붙인 경우 제거
for prefix in ("/api/index.py",):
if path.startswith(prefix):
path = path[len(prefix) :] or "/"
break
# 2) 헤더에 실린 원본 경로가 있으면 사용
headers = request.headers
original = headers.get("x-original-pathname") or headers.get("x-vercel-original-pathname") or headers.get("x-matched-path")
if original:
path = original
# FastAPI 내부가 path 기반으로 라우팅하므로 scope의 path만 수정
request.scope["path"] = path
logger.info("path_restore original=%s restored=%s host=%s", original_path, path, headers.get("host"))
return await call_next(request)
@app.get("/health")
async def health() -> dict:
return {"status": "ok"}
@app.get("/favicon.ico")
async def favicon_ico():
# Return empty response to avoid 404 on favicon requests
return Response(status_code=204)
@app.get("/favicon.png")
async def favicon_png():
return Response(status_code=204)
async def mcp_asgi(scope, receive, send):
"""ASGI entrypoint for Streamable HTTP/SSE transport."""
logger.info("mcp_asgi scope path=%s method=%s headers=%s", scope.get("path"), scope.get("method"), scope.get("headers"))
if scope.get("type") != "http":
return await JSONResponse({"detail": "Unsupported scope"}, status_code=400)(scope, receive, send)
try:
# Vercel serverless에서 lifespan 훅이 호출되지 않을 수 있으므로
# 요청 시점마다 새 manager를 만들고 run()을 한 번만 호출한다.
headers = [(k, v) for k, v in (scope.get("headers") or []) if k != b"accept"]
# 강제로 JSON 우선 Accept를 세팅해 StreamableHTTP의 406을 방지
headers.append((b"accept", b"application/json, text/event-stream, */*"))
new_scope = dict(scope)
new_scope["headers"] = headers
session_manager = _create_session_manager()
async with session_manager.run():
await session_manager.handle_request(new_scope, receive, send)
except Exception as exc: # noqa: BLE001
logger.exception("mcp_asgi internal error")
return await JSONResponse({"detail": "internal error", "error": str(exc)}, status_code=500)(scope, receive, send)
app.mount("/mcp", mcp_asgi)
app.mount("/mcp/", mcp_asgi)
@app.get("/")
async def root():
return {
"service": "SemiProcess MCP",
"spec": MCP_SPEC_VERSION,
"health": "/health",
"mcp": "/mcp",
}
if __name__ == "__main__":
port = int(os.getenv("PORT", "8000"))
import uvicorn
uvicorn.run("src.server:app", host="0.0.0.0", port=port, reload=False)
