main.py•11.7 kB
#!/usr/bin/env python3
"""
Sample MCP Server using FastMCP
간단한 명언/인용구 조회 서버 예제
"""
import asyncio
import sys
import os
import logging
from fastmcp import FastMCP
from fastapi import FastAPI
from pydantic import BaseModel, Field
from .tools import (
get_random_quote,
search_quotes,
get_quote_by_id
)
from typing import Optional
from dotenv import load_dotenv
from contextlib import contextmanager
# .env 파일 로드 (로컬 개발용 - 우선순위 2순위)
load_dotenv()
# FastAPI / FastMCP 앱 구성
api = FastAPI()
mcp_logger = logging.getLogger("sample-mcp")
level = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").upper(), logging.INFO)
mcp_logger.setLevel(level)
if not mcp_logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
mcp_logger.addHandler(handler)
mcp_logger.propagate = True
mcp = FastMCP()
# Pydantic 모델 정의
class QuoteSearchRequest(BaseModel):
keyword: str = Field(..., description="검색할 키워드")
limit: int = Field(10, description="결과 개수", ge=1, le=50)
class QuoteByIdRequest(BaseModel):
quote_id: str = Field(..., description="인용구 ID")
# 실제 구현 함수들
async def get_random_quote_impl(arguments: Optional[dict] = None):
"""랜덤 명언 조회 구현"""
try:
if arguments is None:
arguments = {}
return await asyncio.to_thread(get_random_quote, arguments)
except Exception as e:
return {"error": f"명언 조회 중 오류가 발생했습니다: {str(e)}"}
async def search_quotes_impl(req: QuoteSearchRequest, arguments: Optional[dict] = None):
"""명언 검색 구현"""
try:
if arguments is None:
arguments = {}
return await asyncio.to_thread(search_quotes, req.keyword, req.limit, arguments)
except Exception as e:
return {"error": f"명언 검색 중 오류가 발생했습니다: {str(e)}"}
async def get_quote_by_id_impl(req: QuoteByIdRequest, arguments: Optional[dict] = None):
"""ID로 명언 조회 구현"""
try:
if arguments is None:
arguments = {}
return await asyncio.to_thread(get_quote_by_id, req.quote_id, arguments)
except Exception as e:
return {"error": f"명언 조회 중 오류가 발생했습니다: {str(e)}"}
async def health_impl():
"""서비스 상태 확인 구현"""
api_key = os.environ.get("QUOTE_API_KEY", "")
return {
"status": "ok",
"environment": {
"quote_api_key": "설정됨" if api_key else "설정되지 않음 (선택사항)"
}
}
# 일시 환경 변수 적용용 컨텍스트 매니저
@contextmanager
def temporary_env(overrides: dict):
saved_values = {}
try:
for key, value in (overrides or {}).items():
saved_values[key] = os.environ.get(key)
if value is not None:
os.environ[key] = str(value)
yield
finally:
for key, original in saved_values.items():
if original is None:
os.environ.pop(key, None)
else:
os.environ[key] = original
# HTTP 엔드포인트
@api.get("/health")
async def health_check_get():
"""HTTP GET 엔드포인트: 서비스 상태 확인"""
return await health_impl()
@api.post("/health")
async def health_check_post():
"""HTTP POST 엔드포인트: 서비스 상태 확인"""
return await health_impl()
# HTTP 엔드포인트: 도구 목록 조회
@api.get("/tools")
async def get_tools_http():
"""HTTP 엔드포인트: 사용 가능한 도구 목록 조회"""
try:
tools_list = []
server = getattr(mcp, 'server', None)
if server and hasattr(server, 'tools'):
tools = getattr(server, 'tools', {})
for tool_name, tool in tools.items():
tool_info = {
"name": tool_name,
"description": getattr(tool, 'description', '') or '',
}
if hasattr(tool, 'parameters'):
tool_info["parameters"] = getattr(tool, 'parameters', {})
else:
tool_info["parameters"] = {}
tools_list.append(tool_info)
if not tools_list:
mcp_logger.warning("FastMCP tools not accessible, returning hardcoded tool list")
tools_list = [
{
"name": "get_random_quote_tool",
"description": "랜덤 명언을 조회합니다.",
"parameters": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "search_quotes_tool",
"description": "키워드로 명언을 검색합니다.",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "검색할 키워드"},
"limit": {"type": "integer", "description": "결과 개수 (기본값: 10)"}
},
"required": ["keyword"]
}
},
{
"name": "get_quote_by_id_tool",
"description": "ID로 특정 명언을 조회합니다.",
"parameters": {
"type": "object",
"properties": {
"quote_id": {"type": "string", "description": "인용구 ID"}
},
"required": ["quote_id"]
}
}
]
return tools_list
except Exception as e:
mcp_logger.exception("Error getting tools list: %s", str(e))
return []
# HTTP 엔드포인트: 도구 호출
@api.post("/tools/{tool_name}")
async def call_tool_http(tool_name: str, request_data: dict):
mcp_logger.debug("HTTP call_tool | tool=%s request=%s", tool_name, request_data)
env = request_data.get("env", {}) if isinstance(request_data, dict) else {}
async def run_sync(func, *args, **kwargs):
return await asyncio.to_thread(func, *args, **kwargs)
# 공통 타입 변환 함수들
def convert_float_to_int(data: dict, keys: list):
"""지정된 키의 float 값을 int로 변환"""
for key in keys:
if key in data and isinstance(data[key], float):
data[key] = int(data[key])
def convert_to_str(data: dict, keys: list):
"""지정된 키의 값을 문자열로 변환"""
for key in keys:
if key in data and data[key] is not None and not isinstance(data[key], str):
data[key] = str(data[key])
try:
# 크레덴셜 추출
creds = {}
if isinstance(env, dict):
if "QUOTE_API_KEY" in env:
creds["QUOTE_API_KEY"] = env["QUOTE_API_KEY"]
if creds:
masked = dict(creds)
for key in masked:
if masked[key]:
masked[key] = masked[key][:6] + "***"
mcp_logger.debug("Applying temp env | %s", masked)
async def run_with_env(coro_func):
with temporary_env(creds):
return await coro_func
if tool_name == "health":
return await health_impl()
if tool_name == "get_random_quote_tool":
return await run_with_env(
run_sync(get_random_quote, arguments=request_data)
)
if tool_name == "search_quotes_tool":
keyword = request_data.get("keyword")
if not keyword:
return {"error": "Missing required parameter: keyword"}
limit = request_data.get("limit", 10)
if isinstance(limit, float):
limit = int(limit)
return await run_with_env(
run_sync(search_quotes, keyword, limit, arguments=request_data)
)
if tool_name == "get_quote_by_id_tool":
quote_id = request_data.get("quote_id")
if not quote_id:
return {"error": "Missing required parameter: quote_id"}
convert_to_str(request_data, ["quote_id"])
return await run_with_env(
run_sync(get_quote_by_id, quote_id, arguments=request_data)
)
return {"error": "Tool not found"}
except Exception as e:
mcp_logger.exception("Error in call_tool_http: %s", str(e))
return {"error": f"Error calling tool: {str(e)}"}
# MCP 도구 정의
@mcp.tool()
async def health():
"""서비스 상태 확인"""
return await health_impl()
@mcp.tool()
async def get_random_quote_tool():
"""
랜덤 명언을 조회합니다.
무료 Quote API를 사용하여 랜덤 명언을 가져옵니다.
Returns:
명언 정보 딕셔너리:
{
"quote": "명언 내용",
"author": "작가명",
"id": "인용구 ID"
}
또는 {"error": "오류 메시지"} 형식
"""
return await get_random_quote_impl(None)
@mcp.tool()
async def search_quotes_tool(keyword: str, limit: int = 10):
"""
키워드로 명언을 검색합니다.
Args:
keyword: 검색할 키워드 (필수, 예: "success", "life", "wisdom")
limit: 결과 개수 (기본값: 10, 최대: 50)
Returns:
검색 결과 딕셔너리:
{
"total": 검색된 명언 수,
"quotes": [
{
"quote": "명언 내용",
"author": "작가명",
"id": "인용구 ID"
},
...
]
}
또는 {"error": "오류 메시지"} 형식
"""
req = QuoteSearchRequest(keyword=keyword, limit=limit)
return await search_quotes_impl(req, None)
@mcp.tool()
async def get_quote_by_id_tool(quote_id: str):
"""
ID로 특정 명언을 조회합니다.
Args:
quote_id: 인용구 ID (필수)
Returns:
명언 정보 딕셔너리:
{
"quote": "명언 내용",
"author": "작가명",
"id": "인용구 ID"
}
또는 {"error": "오류 메시지"} 형식
"""
req = QuoteByIdRequest(quote_id=quote_id)
return await get_quote_by_id_impl(req, None)
async def main():
"""MCP 서버를 실행합니다."""
print("MCP Sample Quote Server starting...", file=sys.stderr)
print("Server: sample-quote-service", file=sys.stderr)
print("Available tools: health, get_random_quote_tool, search_quotes_tool, get_quote_by_id_tool", file=sys.stderr)
try:
await mcp.run_stdio_async()
except Exception as e:
print(f"Server error: {e}", file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
raise
if __name__ == "__main__":
# MCP 서버로 실행 (stdio 모드)
# HTTP 서버로 실행하려면 환경 변수 HTTP_MODE=1 설정
if os.environ.get("HTTP_MODE") == "1":
import uvicorn
port = int(os.environ.get('PORT', 8098))
uvicorn.run("src.main:api", host="0.0.0.0", port=port, reload=False)
else:
# MCP stdio 모드
asyncio.run(main())