MCP 서버 개발

# pip 사용
pip install mcp

# uv 사용 (권장)
uv pip install mcp

# 또는 프로젝트 의존성에 추가
uv add mcp

#!/usr/bin/env python3
"""
간단한 MCP 서버 예제
"""

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 서버 인스턴스 생성
server = Server("my-mcp-server")

# 도구 등록
@server.tool()
async def hello(name: str = "World") -> str:
    """
    인사 메시지를 반환합니다.

    Args:
        name: 인사할 이름

    Returns:
        인사 메시지
    """
    return f"Hello, {name}!"

# 메인 함수
async def main():
    # stdio 전송으로 서버 실행
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

@server.tool()
async def add(a: float, b: float) -> float:
    """두 숫자를 더합니다."""
    return a + b

@server.tool()
async def multiply(a: float, b: float) -> float:
    """두 숫자를 곱합니다."""
    return a * b

from typing import List, Dict, Any
from mcp.types import TextContent, ImageContent, EmbeddedResource

@server.tool()
async def search_files(
    directory: str,
    pattern: str,
    max_results: int = 10
) -> List[Dict[str, Any]]:
    """
    디렉토리에서 파일을 검색합니다.

    Args:
        directory: 검색할 디렉토리
        pattern: 파일명 패턴 (glob)
        max_results: 최대 결과 수

    Returns:
        파일 정보 리스트
    """
    import os
    import glob

    search_path = os.path.join(directory, pattern)
    files = glob.glob(search_path, recursive=True)

    results = []
    for filepath in files[:max_results]:
        stat = os.stat(filepath)
        results.append({
            "path": filepath,
            "size": stat.st_size,
            "modified": stat.st_mtime
        })

    return results

import os
from pathlib import Path

@server.tool()
async def read_file(path: str) -> str:
    """파일 내용을 읽습니다."""
    with open(path, 'r', encoding='utf-8') as f:
        return f.read()

@server.tool()
async def write_file(path: str, content: str) -> str:
    """파일에 내용을 씁니다."""
    # 디렉토리가 없으면 생성
    os.makedirs(os.path.dirname(path), exist_ok=True)

    with open(path, 'w', encoding='utf-8') as f:
        f.write(content)

    return f"파일 작성 완료: {path}"

@server.tool()
async def list_directory(path: str) -> List[Dict[str, Any]]:
    """디렉토리 내용을 나열합니다."""
    items = []
    for entry in os.listdir(path):
        full_path = os.path.join(path, entry)
        is_dir = os.path.isdir(full_path)

        items.append({
            "name": entry,
            "path": full_path,
            "type": "directory" if is_dir else "file"
        })

    return items

from mcp.server import McpError

@server.tool()
async def safe_read_file(path: str) -> str:
    """안전하게 파일을 읽습니다."""
    # 경로 검증
    if ".." in path:
        raise McpError(
            code=-32602,
            message="Invalid path: path traversal not allowed"
        )

    # 파일 존재 확인
    if not os.path.exists(path):
        raise McpError(
            code=-32001,
            message="File not found",
            data={"path": path}
        )

    # 읽기 권한 확인
    if not os.access(path, os.R_OK):
        raise McpError(
            code=-32002,
            message="Permission denied",
            data={"path": path}
        )

    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        raise McpError(
            code=-32603,
            message=f"Failed to read file: {e}"
        )

from mcp.types import Resource, TextContent

# 1. 리소스 목록 핸들러
@server.list_resources()
async def list_resources() -> List[Resource]:
    """사용 가능한 리소스 목록을 반환합니다."""
    return [
        Resource(
            uri="file:///etc/hosts",
            name="System hosts file",
            description="시스템 호스트 파일",
            mimeType="text/plain"
        ),
        Resource(
            uri="config://app",
            name="Application config",
            description="애플리케이션 설정",
            mimeType="application/json"
        )
    ]

# 2. 리소스 읽기 핸들러
@server.read_resource()
async def read_resource(uri: str) -> str:
    """리소스 내용을 읽습니다."""
    if uri.startswith("file://"):
        path = uri[7:]  # "file://" 제거
        with open(path, 'r') as f:
            return f.read()

    elif uri == "config://app":
        import json
        config = {
            "version": "1.0.0",
            "debug": False
        }
        return json.dumps(config, indent=2)

    else:
        raise McpError(
            code=-32602,
            message=f"Unknown resource URI: {uri}"
        )

@server.list_resources()
async def list_project_files() -> List[Resource]:
    """프로젝트 파일을 동적으로 리소스로 제공합니다."""
    resources = []
    project_dir = "/path/to/project"

    for root, dirs, files in os.walk(project_dir):
        for filename in files:
            if filename.endswith(('.py', '.js', '.ts')):
                filepath = os.path.join(root, filename)
                uri = f"file://{filepath}"

                resources.append(Resource(
                    uri=uri,
                    name=filename,
                    description=f"Source file: {filepath}",
                    mimeType="text/plain"
                ))

    return resources

from mcp.types import Prompt, PromptMessage, TextContent

# 1. 프롬프트 목록
@server.list_prompts()
async def list_prompts() -> List[Prompt]:
    """사용 가능한 프롬프트 목록을 반환합니다."""
    return [
        Prompt(
            name="code-review",
            description="코드 리뷰 프롬프트",
            arguments=[
                {
                    "name": "language",
                    "description": "프로그래밍 언어",
                    "required": True
                },
                {
                    "name": "code",
                    "description": "리뷰할 코드",
                    "required": True
                }
            ]
        ),
        Prompt(
            name="translate",
            description="번역 프롬프트",
            arguments=[
                {
                    "name": "text",
                    "description": "번역할 텍스트",
                    "required": True
                },
                {
                    "name": "target_lang",
                    "description": "목표 언어",
                    "required": True
                }
            ]
        )
    ]

# 2. 프롬프트 가져오기
@server.get_prompt()
async def get_prompt(name: str, arguments: Dict[str, str]) -> List[PromptMessage]:
    """프롬프트를 생성하여 반환합니다."""
    if name == "code-review":
        language = arguments.get("language")
        code = arguments.get("code")

        prompt_text = f"""다음 {language} 코드를 리뷰해주세요:

```{language}
{code}
```

다음 관점에서 검토해주세요:
1. 코드 품질 및 가독성
2. 성능 최적화 가능성
3. 보안 취약점
4. 베스트 프랙티스 준수 여부
5. 개선 제안

각 항목에 대해 구체적인 피드백을 제공해주세요."""

        return [
            PromptMessage(
                role="user",
                content=TextContent(type="text", text=prompt_text)
            )
        ]

    elif name == "translate":
        text = arguments.get("text")
        target_lang = arguments.get("target_lang")

        prompt_text = f"""다음 텍스트를 {target_lang}로 번역해주세요:

{text}

번역 시 다음을 고려해주세요:
- 자연스러운 표현 사용
- 문화적 맥락 반영
- 전문 용어의 정확한 번역"""

        return [
            PromptMessage(
                role="user",
                content=TextContent(type="text", text=prompt_text)
            )
        ]

    else:
        raise McpError(
            code=-32602,
            message=f"Unknown prompt: {name}"
        )

#!/usr/bin/env python3
"""
완전한 MCP 파일시스템 서버
"""

import os
import json
from typing import List, Dict, Any
from pathlib import Path

from mcp.server import Server, McpError
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Prompt, PromptMessage, TextContent

# 서버 생성
server = Server("filesystem-server")

# 허용된 디렉토리 (보안)
ALLOWED_DIRS = [
    str(Path.home() / "Documents"),
    str(Path.home() / "Projects")
]

def is_path_allowed(path: str) -> bool:
    """경로가 허용된 디렉토리 내에 있는지 확인합니다."""
    abs_path = os.path.abspath(path)
    return any(abs_path.startswith(allowed) for allowed in ALLOWED_DIRS)

# ===== 도구 =====

@server.tool()
async def read_file(path: str) -> str:
    """파일 내용을 읽습니다."""
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        raise McpError(code=-32001, message="File not found")
    except Exception as e:
        raise McpError(code=-32603, message=str(e))

@server.tool()
async def write_file(path: str, content: str) -> str:
    """파일에 내용을 씁니다."""
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"Successfully wrote {len(content)} bytes to {path}"
    except Exception as e:
        raise McpError(code=-32603, message=str(e))

@server.tool()
async def list_directory(path: str) -> List[Dict[str, Any]]:
    """디렉토리 내용을 나열합니다."""
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    try:
        items = []
        for entry in os.listdir(path):
            full_path = os.path.join(path, entry)
            stat = os.stat(full_path)

            items.append({
                "name": entry,
                "path": full_path,
                "type": "directory" if os.path.isdir(full_path) else "file",
                "size": stat.st_size,
                "modified": stat.st_mtime
            })

        return items
    except Exception as e:
        raise McpError(code=-32603, message=str(e))

@server.tool()
async def search_files(
    directory: str,
    pattern: str,
    max_results: int = 50
) -> List[str]:
    """파일을 검색합니다."""
    if not is_path_allowed(directory):
        raise McpError(code=-32602, message="Path not allowed")

    import glob

    search_path = os.path.join(directory, "**", pattern)
    results = glob.glob(search_path, recursive=True)
    return results[:max_results]

# ===== 리소스 =====

@server.list_resources()
async def list_resources() -> List[Resource]:
    """허용된 디렉토리를 리소스로 나열합니다."""
    resources = []
    for dir_path in ALLOWED_DIRS:
        if os.path.exists(dir_path):
            resources.append(Resource(
                uri=f"file://{dir_path}",
                name=os.path.basename(dir_path),
                description=f"Directory: {dir_path}",
                mimeType="text/directory"
            ))
    return resources

@server.read_resource()
async def read_resource(uri: str) -> str:
    """리소스를 읽습니다."""
    if not uri.startswith("file://"):
        raise McpError(code=-32602, message="Only file:// URIs supported")

    path = uri[7:]
    if not is_path_allowed(path):
        raise McpError(code=-32602, message="Path not allowed")

    if os.path.isdir(path):
        # 디렉토리 내용을 JSON으로 반환
        items = await list_directory(path)
        return json.dumps(items, indent=2)
    else:
        # 파일 내용 반환
        return await read_file(path)

# ===== 메인 =====

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

# 프로젝트 초기화
mkdir my-mcp-server
cd my-mcp-server
npm init -y

# 의존성 설치
npm install @modelcontextprotocol/sdk
npm install -D typescript @types/node

# TypeScript 설정
npx tsc --init

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules"]
}

#!/usr/bin/env node
// src/index.ts

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  Tool
} from "@modelcontextprotocol/sdk/types.js";

// 서버 생성
const server = new Server(
  {
    name: "my-mcp-server",
    version: "1.0.0"
  },
  {
    capabilities: {
      tools: {}
    }
  }
);

// 도구 목록
const TOOLS: Tool[] = [
  {
    name: "hello",
    description: "인사 메시지를 반환합니다",
    inputSchema: {
      type: "object",
      properties: {
        name: {
          type: "string",
          description: "인사할 이름"
        }
      },
      required: []
    }
  }
];

// 도구 목록 핸들러
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: TOOLS
}));

// 도구 호출 핸들러
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "hello") {
    const userName = (args as { name?: string }).name || "World";
    return {
      content: [
        {
          type: "text",
          text: `Hello, ${userName}!`
        }
      ]
    };
  }

  throw new Error(`Unknown tool: ${name}`);
});

// 서버 실행
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);

import * as fs from "fs/promises";
import * as path from "path";

// 도구 정의
const TOOLS: Tool[] = [
  {
    name: "read_file",
    description: "파일 내용을 읽습니다",
    inputSchema: {
      type: "object",
      properties: {
        path: {
          type: "string",
          description: "파일 경로"
        }
      },
      required: ["path"]
    }
  },
  {
    name: "write_file",
    description: "파일에 내용을 씁니다",
    inputSchema: {
      type: "object",
      properties: {
        path: {
          type: "string",
          description: "파일 경로"
        },
        content: {
          type: "string",
          description: "파일 내용"
        }
      },
      required: ["path", "content"]
    }
  },
  {
    name: "list_directory",
    description: "디렉토리 내용을 나열합니다",
    inputSchema: {
      type: "object",
      properties: {
        path: {
          type: "string",
          description: "디렉토리 경로"
        }
      },
      required: ["path"]
    }
  }
];

// 도구 구현
async function readFile(filePath: string): Promise<string> {
  const content = await fs.readFile(filePath, "utf-8");
  return content;
}

async function writeFile(filePath: string, content: string): Promise<string> {
  await fs.mkdir(path.dirname(filePath), { recursive: true });
  await fs.writeFile(filePath, content, "utf-8");
  return `Successfully wrote ${content.length} bytes to ${filePath}`;
}

async function listDirectory(dirPath: string) {
  const entries = await fs.readdir(dirPath, { withFileTypes: true });

  const items = await Promise.all(
    entries.map(async (entry) => {
      const fullPath = path.join(dirPath, entry.name);
      const stats = await fs.stat(fullPath);

      return {
        name: entry.name,
        path: fullPath,
        type: entry.isDirectory() ? "directory" : "file",
        size: stats.size,
        modified: stats.mtime
      };
    })
  );

  return items;
}

// 도구 호출 핸들러
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  try {
    switch (name) {
      case "read_file": {
        const { path } = args as { path: string };
        const content = await readFile(path);
        return {
          content: [{ type: "text", text: content }]
        };
      }

      case "write_file": {
        const { path, content } = args as { path: string; content: string };
        const result = await writeFile(path, content);
        return {
          content: [{ type: "text", text: result }]
        };
      }

      case "list_directory": {
        const { path } = args as { path: string };
        const items = await listDirectory(path);
        return {
          content: [{ type: "text", text: JSON.stringify(items, null, 2) }]
        };
      }

      default:
        throw new Error(`Unknown tool: ${name}`);
    }
  } catch (error) {
    const errorMessage = error instanceof Error ? error.message : String(error);
    return {
      content: [{ type: "text", text: `Error: ${errorMessage}` }],
      isError: true
    };
  }
});

import {
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
  Resource
} from "@modelcontextprotocol/sdk/types.js";

// 리소스 목록 핸들러
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const resources: Resource[] = [
    {
      uri: "file:///etc/hosts",
      name: "System hosts",
      description: "시스템 호스트 파일",
      mimeType: "text/plain"
    },
    {
      uri: "config://app",
      name: "App config",
      description: "애플리케이션 설정",
      mimeType: "application/json"
    }
  ];

  return { resources };
});

// 리소스 읽기 핸들러
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;

  if (uri.startsWith("file://")) {
    const filePath = uri.slice(7);
    const content = await fs.readFile(filePath, "utf-8");

    return {
      contents: [
        {
          uri,
          mimeType: "text/plain",
          text: content
        }
      ]
    };
  }

  if (uri === "config://app") {
    const config = {
      version: "1.0.0",
      debug: false
    };

    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(config, null, 2)
        }
      ]
    };
  }

  throw new Error(`Unknown resource: ${uri}`);
});

#!/usr/bin/env node
// src/index.ts - 완전한 파일시스템 MCP 서버

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
  Tool,
  Resource
} from "@modelcontextprotocol/sdk/types.js";
import * as fs from "fs/promises";
import * as path from "path";
import { glob } from "glob";

// 허용된 디렉토리
const ALLOWED_DIRS = [
  path.join(process.env.HOME || "", "Documents"),
  path.join(process.env.HOME || "", "Projects")
];

function isPathAllowed(filePath: string): boolean {
  const absolute = path.resolve(filePath);
  return ALLOWED_DIRS.some((dir) => absolute.startsWith(dir));
}

// 서버 생성
const server = new Server(
  {
    name: "filesystem-server",
    version: "1.0.0"
  },
  {
    capabilities: {
      tools: {},
      resources: {}
    }
  }
);

// 도구 정의
const TOOLS: Tool[] = [
  {
    name: "read_file",
    description: "파일 내용을 읽습니다",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "파일 경로" }
      },
      required: ["path"]
    }
  },
  {
    name: "write_file",
    description: "파일에 내용을 씁니다",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "파일 경로" },
        content: { type: "string", description: "파일 내용" }
      },
      required: ["path", "content"]
    }
  },
  {
    name: "list_directory",
    description: "디렉토리 내용을 나열합니다",
    inputSchema: {
      type: "object",
      properties: {
        path: { type: "string", description: "디렉토리 경로" }
      },
      required: ["path"]
    }
  },
  {
    name: "search_files",
    description: "파일을 검색합니다",
    inputSchema: {
      type: "object",
      properties: {
        directory: { type: "string", description: "검색 디렉토리" },
        pattern: { type: "string", description: "파일명 패턴 (glob)" }
      },
      required: ["directory", "pattern"]
    }
  }
];

// 핸들러 등록
server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: TOOLS }));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  try {
    switch (name) {
      case "read_file": {
        const { path: filePath } = args as { path: string };
        if (!isPathAllowed(filePath)) {
          throw new Error("Path not allowed");
        }
        const content = await fs.readFile(filePath, "utf-8");
        return { content: [{ type: "text", text: content }] };
      }

      case "write_file": {
        const { path: filePath, content } = args as { path: string; content: string };
        if (!isPathAllowed(filePath)) {
          throw new Error("Path not allowed");
        }
        await fs.mkdir(path.dirname(filePath), { recursive: true });
        await fs.writeFile(filePath, content, "utf-8");
        return { content: [{ type: "text", text: `Wrote to ${filePath}` }] };
      }

      case "list_directory": {
        const { path: dirPath } = args as { path: string };
        if (!isPathAllowed(dirPath)) {
          throw new Error("Path not allowed");
        }
        const entries = await fs.readdir(dirPath, { withFileTypes: true });
        const items = entries.map((e) => ({
          name: e.name,
          type: e.isDirectory() ? "directory" : "file"
        }));
        return { content: [{ type: "text", text: JSON.stringify(items, null, 2) }] };
      }

      case "search_files": {
        const { directory, pattern } = args as { directory: string; pattern: string };
        if (!isPathAllowed(directory)) {
          throw new Error("Path not allowed");
        }
        const files = await glob(pattern, { cwd: directory });
        return { content: [{ type: "text", text: JSON.stringify(files, null, 2) }] };
      }

      default:
        throw new Error(`Unknown tool: ${name}`);
    }
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${error}` }],
      isError: true
    };
  }
});

// 리소스 핸들러
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const resources: Resource[] = ALLOWED_DIRS
    .filter((dir) => fs.access(dir).then(() => true).catch(() => false))
    .map((dir) => ({
      uri: `file://${dir}`,
      name: path.basename(dir),
      description: `Directory: ${dir}`,
      mimeType: "text/directory"
    }));

  return { resources };
});

server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;

  if (!uri.startsWith("file://")) {
    throw new Error("Only file:// URIs supported");
  }

  const filePath = uri.slice(7);
  if (!isPathAllowed(filePath)) {
    throw new Error("Path not allowed");
  }

  const content = await fs.readFile(filePath, "utf-8");

  return {
    contents: [{
      uri,
      mimeType: "text/plain",
      text: content
    }]
  };
});

// 서버 실행
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);

[project]
name = "mcp-server-filesystem"
version = "1.0.0"
description = "MCP filesystem server"
authors = [{ name = "Your Name", email = "you@example.com" }]
requires-python = ">=3.10"
dependencies = [
    "mcp>=0.1.0"
]

[project.scripts]
mcp-server-filesystem = "mcp_server_filesystem:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# 개발 모드 설치
uv pip install -e .

# 또는 pip
pip install -e .

# 서버 실행
mcp-server-filesystem

{
  "name": "@myorg/mcp-server-filesystem",
  "version": "1.0.0",
  "description": "MCP filesystem server",
  "type": "module",
  "bin": {
    "mcp-server-filesystem": "./dist/index.js"
  },
  "scripts": {
    "build": "tsc",
    "prepare": "npm run build"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^0.1.0"
  },
  "devDependencies": {
    "typescript": "^5.3.0",
    "@types/node": "^20.0.0"
  }
}

# 빌드
npm run build

# 글로벌 설치
npm install -g .

# 서버 실행
mcp-server-filesystem

# Python (PyPI)
uv build
uv publish

# TypeScript (npm)
npm publish

# MCP Inspector 사용
npx @modelcontextprotocol/inspector python server.py

# 또는 TypeScript 서버
npx @modelcontextprotocol/inspector node dist/index.js

import pytest
from mcp.client import Client
from mcp.client.stdio import StdioClientTransport

@pytest.mark.asyncio
async def test_server():
    client = Client({"name": "test", "version": "1.0.0"}, {"capabilities": {"tools": {}}})

    transport = StdioClientTransport({
        "command": "python",
        "args": ["server.py"]
    })

    await client.connect(transport)

    # 도구 목록 테스트
    tools = await client.list_tools()
    assert len(tools) > 0

    # 도구 호출 테스트
    result = await client.call_tool("hello", {"name": "Test"})
    assert "Hello, Test!" in result

    await client.close()