MCP Tool
- AI 모델이 Python 함수를 직접 호출하여 계산, 데이터 처리 등 다양한 작업을 수행할 수 있게 해주는 기능.
- MCP 서버는 AI 모델과 Python 코드 사이의 중간 인터페이스 역할을 하며, AI 모델이 필요에 따라 적절한 함수를 호출할 수 있도록 함.
- 주요 구성 요소
- Server: MCP 서버 인스턴스
- Tool: AI가 호출할 수 있는 도구 정의
- Prompt: 사용자와 AI 간의 상호작용을 위한 프롬프트 정의
- 비즈니스 로직: 실제 작업을 수행하는 함수들
간단한 MCP 서버 구현 예
덧셈과 뺄셈 기능을 제공하는 간단한 MCP 서버 예시:
import asyncio
import sys
import traceback
from typing import Annotated, Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.shared.exceptions import McpError
from mcp.types import (
INTERNAL_ERROR,
INVALID_PARAMS,
ErrorData,
GetPromptResult,
Prompt,
PromptArgument,
PromptMessage,
TextContent,
Tool,
)
from pydantic import BaseModel, Field
class CalculatorParams(BaseModel):
"""계산기 작업에 필요한 파라미터"""
a: Annotated[float, Field(description="첫 번째 숫자")]
b: Annotated[float, Field(description="두 번째 숫자")]
class Calculator:
"""기본 계산 작업을 처리하는 클래스"""
async def add(self, params: CalculatorParams) -> str:
"""두 숫자를 더하는 함수"""
try:
result = params.a + params.b
return f"{params.a} + {params.b} = {result}"
except Exception as e:
traceback.print_exc(file=sys.stderr)
raise McpError(
ErrorData(
code=INTERNAL_ERROR,
message=f"덧셈 중 예상치 못한 오류 발생: {str(e)}",
)
)
async def subtract(self, params: CalculatorParams) -> str:
"""첫 번째 숫자에서 두 번째 숫자를 빼는 함수"""
try:
result = params.a - params.b
return f"{params.a} - {params.b} = {result}"
except Exception as e:
traceback.print_exc(file=sys.stderr)
raise McpError(
ErrorData(
code=INTERNAL_ERROR,
message=f"뺄셈 중 예상치 못한 오류 발생: {str(e)}",
)
)
async def serve() -> None:
"""계산기 MCP 서버 실행"""
server = Server("mcp-calculator")
calculator = Calculator()
@server.list_tools()
async def list_tools() -> list[Tool]:
"""AI가 사용할 수 있는 도구 목록 정의"""
return [
Tool(
name="add",
description="두 숫자를 더합니다.",
inputSchema=CalculatorParams.model_json_schema(),
),
Tool(
name="subtract",
description="첫 번째 숫자에서 두 번째 숫자를 뺍니다.",
inputSchema=CalculatorParams.model_json_schema(),
),
]
@server.list_prompts()
async def list_prompts() -> list[Prompt]:
"""사용자가 선택할 수 있는 프롬프트 목록 정의"""
return [
Prompt(
name="add",
description="두 숫자의 합 계산",
arguments=[
PromptArgument(
name="a", description="첫 번째 숫자", required=True
),
PromptArgument(
name="b", description="두 번째 숫자", required=True
),
],
),
Prompt(
name="subtract",
description="두 숫자의 차 계산",
arguments=[
PromptArgument(
name="a", description="첫 번째 숫자", required=True
),
PromptArgument(
name="b", description="두 번째 숫자", required=True
),
],
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""AI가 호출한 도구를 실행하는 함수"""
try:
params = CalculatorParams(**arguments)
if name == "add":
result = await calculator.add(params)
return [TextContent(type="text", text=result)]
elif name == "subtract":
result = await calculator.subtract(params)
return [TextContent(type="text", text=result)]
else:
raise McpError(
ErrorData(code=INVALID_PARAMS, message=f"알 수 없는 도구: {name}")
)
except ValueError as e:
raise McpError(ErrorData(code=INVALID_PARAMS, message=str(e)))
@server.get_prompt()
async def get_prompt(name: str, arguments: dict | None) -> GetPromptResult:
"""프롬프트 호출 처리"""
if not arguments or not all(k in arguments for k in ["a", "b"]):
raise McpError(
ErrorData(
code=INVALID_PARAMS, message="두 숫자가 모두 필요합니다"
)
)
try:
params = CalculatorParams(**arguments)
if name == "add":
result = await calculator.add(params)
return GetPromptResult(
description=f"덧셈 완료: {params.a} + {params.b}",
messages=[
PromptMessage(
role="user", content=TextContent(type="text", text=result)
)
],
)
elif name == "subtract":
result = await calculator.subtract(params)
return GetPromptResult(
description=f"뺄셈 완료: {params.a} - {params.b}",
messages=[
PromptMessage(
role="user", content=TextContent(type="text", text=result)
)
],
)
else:
raise McpError(
ErrorData(code=INVALID_PARAMS, message=f"알 수 없는 프롬프트: {name}")
)
except Exception as e:
return GetPromptResult(
description="계산 실패",
messages=[
PromptMessage(
role="user", content=TextContent(type="text", text=str(e))
)
],
)
options = server.create_initialization_options()
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, options, raise_exceptions=True)
def main():
"""계산기 MCP 서버 실행 함수"""
import argparse
parser = argparse.ArgumentParser(description="기본 계산 기능을 제공하는 MCP 서버")
args = parser.parse_args()
asyncio.run(serve())
if __name__ == "__main__":
main()MCP 서버 구성요소
기본 클래스 및 타입
CalculatorParams: 계산에 필요한 매개변수 정의Calculator: 실제 계산 로직이 구현된 클래스
서버 설정 및 함수
list_tools(): 호출할 수 있는 도구 목록list_prompts(): 사용자가 선택할 수 있는 프롬프트 목록call_tool(): Claude가 도구를 호출할 때 실행되는 함수get_prompt(): 사용자가 프롬프트를 요청할 때 실행되는 함수
실행 프로세스
- 서버 인스턴스 생성
- 도구 및 프롬프트 등록
- stdin/stdout을 통한 통신
- 비동기 이벤트 루프 실행