import logging
from typing import Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from .tools import ToolNotFoundError, execute_tool, list_tool_definitions
logger = logging.getLogger("mcp_server")
class MCPRequest(BaseModel):
# JSON-RPC 2.0 요청 스키마
jsonrpc: str = "2.0"
id: Optional[str] = None
method: str
params: Optional[dict] = Field(default_factory=dict)
class ToolSchema(BaseModel):
# 도구의 입력/출력 스키마 정의
type: str
properties: dict
required: list[str] = Field(default_factory=list)
class ToolModel(BaseModel):
# 도구 메타데이터 모델
name: str
description: str
inputSchema: ToolSchema
outputSchema: ToolSchema
class ToolExecuteRequest(BaseModel):
# 도구 실행 요청 스키마
tool: str
input: dict = Field(default_factory=dict)
app = FastAPI(title="MCP Server (Python)")
def _build_tools_list() -> list[dict]:
return [
{
"name": tool_def.name,
"description": tool_def.description,
"inputSchema": {
"type": tool_def.input_schema.tool_type,
"properties": tool_def.input_schema.properties,
"required": tool_def.input_schema.required,
},
}
for tool_def in list_tool_definitions()
]
def _build_tool_models() -> list[ToolModel]:
return [
ToolModel(
name=tool_def.name,
description=tool_def.description,
inputSchema=ToolSchema(
type=tool_def.input_schema.tool_type,
properties=tool_def.input_schema.properties,
required=tool_def.input_schema.required,
),
outputSchema=ToolSchema(
type=tool_def.output_schema.tool_type,
properties=tool_def.output_schema.properties,
required=tool_def.output_schema.required,
),
)
for tool_def in list_tool_definitions()
]
@app.on_event("startup")
async def on_startup():
# 서버 시작 시 로그 출력
logger.info("[MCP] Server starting up")
@app.get("/health")
async def health():
# 상태 확인 요청 처리
logger.info("[MCP] Health check requested")
return {"status": "ok"}
@app.post("/mcp")
async def mcp(req: MCPRequest):
# JSON-RPC 2.0 요청 수신/처리
logger.info("[MCP] /mcp endpoint reached")
logger.info("[MCP] Request body: %s", req.model_dump())
logger.info("[MCP] Request received: id=%s, method=%s", req.id, req.method)
try:
if req.method == "initialize":
logger.info("[MCP] Initialize requested")
protocol_version = req.params.get("protocolVersion", "2024-11-05")
return {
"jsonrpc": "2.0",
"id": req.id,
"result": {
"protocolVersion": protocol_version,
"capabilities": {
"tools": {}
},
"serverInfo": {
"name": "mcp-server-py",
"version": "0.1.0"
}
}
}
elif req.method == "notifications/initialized":
logger.info("[MCP] Client initialized notification received")
# Notification은 응답이 필요 없음 (id가 None)
if req.id is None:
return {"jsonrpc": "2.0"}
else:
return {"jsonrpc": "2.0", "id": req.id, "result": {}}
elif req.method == "tools/list":
logger.info("[MCP] Copilot이 도구 목록을 요청했습니다.")
return {
"jsonrpc": "2.0",
"id": req.id,
"result": {"tools": _build_tools_list()}
}
elif req.method == "tools/call":
tool_name = req.params.get("name")
tool_arguments = req.params.get("arguments", {})
logger.info("[MCP] Tool execute requested: tool=%s", tool_name)
try:
result = execute_tool(tool_name, tool_arguments)
logger.info("[MCP] Tool executed: tool=%s", tool_name)
return {
"jsonrpc": "2.0",
"id": req.id,
"result": result
}
except ToolNotFoundError as e:
logger.warning("[MCP] Tool not found: tool=%s", tool_name)
return {
"jsonrpc": "2.0",
"id": req.id,
"error": {
"code": -32601,
"message": "Tool not found",
"data": {"tool": tool_name}
}
}
else:
logger.warning("[MCP] Unknown method: %s", req.method)
return {
"jsonrpc": "2.0",
"id": req.id,
"error": {
"code": -32601,
"message": "Method not found",
"data": {"method": req.method}
}
}
except Exception as e:
logger.error("[MCP] Error processing request: %s", str(e))
return {
"jsonrpc": "2.0",
"id": req.id,
"error": {
"code": -32603,
"message": "Internal error",
"data": {"error": str(e)}
}
}
@app.get("/mcp/tools")
async def list_tools():
# 도구 목록 샘플 제공
logger.info("[MCP] Tool list requested")
tools = _build_tool_models()
logger.info("[MCP] Tool list returned: count=%d", len(tools))
return {"tools": tools}
@app.post("/mcp/tool/execute")
async def execute_tool_handler(req: ToolExecuteRequest):
# 도구 실행 요청 처리
logger.info("[MCP] Tool execute requested: tool=%s", req.tool)
try:
result = execute_tool(req.tool, req.input)
except ToolNotFoundError:
logger.warning("[MCP] Tool not found: tool=%s", req.tool)
raise HTTPException(status_code=404, detail={"error": "tool_not_found", "tool": req.tool})
logger.info("[MCP] Tool executed: tool=%s", req.tool)
return {"tool": req.tool, "output": result}
