Model Context Protocol(MCP) 완전 가이드: AI 도구 표준화와 Claude 통합까지

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/tmp/example.txt"
    }
  }
}

[Host: Claude Desktop]
    └── [Client] ──── stdio/SSE ──── [MCP Server: filesystem]
    └── [Client] ──── stdio/SSE ──── [MCP Server: github]
    └── [Client] ──── stdio/SSE ──── [MCP Server: database]

@mcp.prompt()
def code_review_prompt(language: str, code: str) -> str:
    return f"{language} 코드를 검토하고 개선점을 알려주세요:\n\n{code}"

pip install fastmcp

from fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

@mcp.tool()
def read_file(path: str) -> str:
    """파일 내용을 읽어 반환합니다."""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str) -> str:
    """파일에 내용을 씁니다."""
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"{path} 파일 저장 완료"

@mcp.resource("file://{path}")
def get_file_resource(path: str) -> str:
    """URI를 통해 파일 리소스를 제공합니다."""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

if __name__ == "__main__":
    mcp.run()

import os
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

@mcp.tool()
def list_directory(path: str = ".") -> list[str]:
    """디렉토리의 파일 목록을 반환합니다."""
    p = Path(path)
    if not p.is_dir():
        raise ValueError(f"{path}는 디렉토리가 아닙니다")
    return [str(item) for item in p.iterdir()]

@mcp.tool()
def search_files(directory: str, pattern: str) -> list[str]:
    """디렉토리에서 패턴에 맞는 파일을 검색합니다."""
    p = Path(directory)
    return [str(f) for f in p.rglob(pattern)]

if __name__ == "__main__":
    mcp.run(transport="stdio")

import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { z } from 'zod'

const server = new McpServer({
  name: 'database-server',
  version: '1.0.0',
})

// 리소스: DB 레코드를 URI로 노출
server.resource(
  'user',
  new ResourceTemplate('db://users/{id}', { list: undefined }),
  async (uri, { id }) => {
    const user = await fetchUserById(id)
    return {
      contents: [
        {
          uri: uri.href,
          text: JSON.stringify(user, null, 2),
          mimeType: 'application/json',
        },
      ],
    }
  }
)

// 도구: SQL 쿼리 실행
server.tool('query_db', { sql: z.string().describe('실행할 SQL 쿼리') }, async ({ sql }) => {
  const results = await executeQuery(sql)
  return {
    content: [
      {
        type: 'text',
        text: JSON.stringify(results, null, 2),
      },
    ],
  }
})

async function main() {
  const transport = new StdioServerTransport()
  await server.connect(transport)
}

main()

# SSE 서버 실행 예시
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["/path/to/filesystem_server.py"],
      "env": {
        "ALLOWED_DIRS": "/home/user/documents,/tmp"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_your_token_here"
      }
    },
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "/path/to/database.db"]
    }
  }
}

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 초기화
            await session.initialize()

            # 사용 가능한 도구 목록
            tools = await session.list_tools()
            print("도구 목록:", [t.name for t in tools.tools])

            # 도구 호출
            result = await session.call_tool(
                "read_file",
                arguments={"path": "/tmp/test.txt"}
            )
            print("결과:", result.content[0].text)

asyncio.run(main())

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def create_mcp_agent():
    model = ChatAnthropic(model="claude-3-5-sonnet-20241022")

    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # MCP 도구를 LangChain 도구로 변환
            tools = await load_mcp_tools(session)

            # ReAct 에이전트 생성
            agent = create_react_agent(model, tools)

            result = await agent.ainvoke({
                "messages": [{"role": "user", "content": "/tmp 디렉토리의 파일 목록을 보여주세요"}]
            })
            return result

asyncio.run(create_mcp_agent())

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("safe-server")

DANGEROUS_PATHS = ["/etc", "/sys", "/proc", "/root"]

@mcp.tool()
def delete_file(path: str, confirmed: bool = False) -> str:
    """
    파일을 삭제합니다. 삭제 전 반드시 confirmed=True를 명시해야 합니다.

    Args:
        path: 삭제할 파일 경로
        confirmed: 삭제 의사를 명시적으로 확인 (기본값: False)
    """
    import os
    from pathlib import Path

    abs_path = str(Path(path).resolve())

    # 위험 경로 차단
    for danger in DANGEROUS_PATHS:
        if abs_path.startswith(danger):
            raise ToolError(f"{danger} 하위 경로는 삭제할 수 없습니다")

    # 확인 없이 호출된 경우 사용자에게 확인 요청
    if not confirmed:
        return (
            f"경고: {abs_path} 파일을 영구 삭제하려고 합니다. "
            f"confirmed=True를 설정하여 다시 호출하면 삭제됩니다."
        )

    os.remove(abs_path)
    return f"{abs_path} 삭제 완료"

import os
from pathlib import Path

ALLOWED_DIRS = os.getenv("ALLOWED_DIRS", "/tmp").split(",")

def is_path_allowed(path: str) -> bool:
    abs_path = Path(path).resolve()
    return any(
        str(abs_path).startswith(allowed.strip())
        for allowed in ALLOWED_DIRS
    )

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
# 루트가 아닌 전용 사용자로 실행
RUN useradd -m mcpuser
USER mcpuser
CMD ["python", "server.py"]

from pydantic import BaseModel, validator

class FileReadParams(BaseModel):
    path: str

    @validator("path")
    def validate_path(cls, v):
        p = Path(v).resolve()
        if ".." in str(p):
            raise ValueError("경로 트래버설 시도 감지")
        if not is_path_allowed(str(p)):
            raise ValueError("허용되지 않은 경로입니다")
        return str(p)