MCP 서버 직접 만들기 — 내 서비스를 Claude에 연동하는 법

MCP(Model Context Protocol)는 Claude가 외부 서비스랑 직접 대화하게 해주는 표준 프로토콜이에요.

기존에는 이랬어요.

나: "우리 DB에서 오늘 주문 건수 알려줘"
Claude: "저는 DB에 접근할 수 없어요. 직접 확인해보세요."

MCP 서버 만들면 이렇게 돼요.

나: "우리 DB에서 오늘 주문 건수 알려줘"
Claude: (MCP로 DB 조회) → "오늘 주문 247건, 어제보다 23% 증가했어요."

---

MCP 서버가 어떻게 동작하나

Claude ←→ MCP 서버 ←→ 내 서비스(DB, API, 파일 등)

1. Claude가 MCP 서버에 "이 툴 써줘" 요청
2. MCP 서버가 실제 서비스 호출
3. 결과를 Claude에게 반환
4. Claude가 결과 보고 답변

MCP 서버는 툴(Tool) 을 정의해요. 툴은 Claude가 호출할 수 있는 함수예요.

---

설치

pip install mcp

끝이에요. 의존성 하나예요.

---

기본 구조 — 50줄짜리 MCP 서버

# my_mcp_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio

# 서버 생성
app = Server("my-service")

# 툴 목록 정의 (Claude가 어떤 툴 쓸 수 있는지 알려줌)
@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="get_today_orders",
            description="오늘 주문 건수와 매출을 조회한다",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        )
    ]

# 툴 실행 (Claude가 툴 호출 시 실행됨)
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_today_orders":
        # 여기서 실제 DB 조회
        result = {
            "count": 247,
            "revenue": 3840000,
            "vs_yesterday": "+23%"
        }
        return [types.TextContent(type="text", text=str(result))]

# 서버 실행
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

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

이게 MCP 서버 기본 구조예요. 실제 서비스 연동은 call_tool 안에 코드 추가하면 돼요.

---

실전 예시 1 — PostgreSQL 연동

# postgres_mcp.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import asyncpg
import os

app = Server("postgres-mcp")

# DB 연결
async def get_db():
    return await asyncpg.connect(os.environ["DATABASE_URL"])

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="query_db",
            description="PostgreSQL 데이터베이스에 SELECT 쿼리를 실행한다",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "실행할 SELECT SQL 쿼리"
                    }
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="get_table_schema",
            description="테이블 스키마를 조회한다",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {
                        "type": "string",
                        "description": "조회할 테이블 이름"
                    }
                },
                "required": ["table_name"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = await get_db()

    try:
        if name == "query_db":
            query = arguments["query"]

            # SELECT만 허용 (보안)
            if not query.strip().upper().startswith("SELECT"):
                return [types.TextContent(
                    type="text",
                    text="❌ SELECT 쿼리만 허용됩니다."
                )]

            rows = await conn.fetch(query)
            result = [dict(row) for row in rows]
            return [types.TextContent(type="text", text=str(result))]

        elif name == "get_table_schema":
            table = arguments["table_name"]
            rows = await conn.fetch("""
                SELECT column_name, data_type, is_nullable
                FROM information_schema.columns
                WHERE table_name = $1
                ORDER BY ordinal_position
            """, table)
            schema = [dict(row) for row in rows]
            return [types.TextContent(type="text", text=str(schema))]

    finally:
        await conn.close()

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

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

Claude에 등록:

claude mcp add postgres-mcp python /path/to/postgres_mcp.py

실제 사용:

나: "users 테이블 스키마 알려줘"
Claude: (get_table_schema 호출)
→ "users 테이블:
   - id: integer, NOT NULL
   - email: varchar, NOT NULL
   - created_at: timestamp, NOT NULL
   - deleted_at: timestamp, NULL"

나: "오늘 가입한 유저 몇 명이야?"
Claude: (query_db 호출)
→ SELECT COUNT(*) FROM users WHERE created_at >= CURRENT_DATE
→ "오늘 가입한 유저: 47명"

---

실전 예시 2 — REST API 연동

사내 API나 외부 서비스 연동해요.

# api_mcp.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import httpx
import os

app = Server("internal-api-mcp")
BASE_URL = os.environ.get("API_BASE_URL", "https://api.myservice.com")
API_KEY = os.environ["API_KEY"]

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="get_user",
            description="유저 ID로 유저 정보를 조회한다",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "integer", "description": "유저 ID"}
                },
                "required": ["user_id"]
            }
        ),
        types.Tool(
            name="search_orders",
            description="조건으로 주문을 검색한다",
            inputSchema={
                "type": "object",
                "properties": {
                    "status": {
                        "type": "string",
                        "enum": ["pending", "completed", "cancelled"],
                        "description": "주문 상태"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "최대 조회 건수 (기본 10)"
                    }
                },
                "required": []
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    headers = {"Authorization": f"Bearer {API_KEY}"}

    async with httpx.AsyncClient() as client:
        if name == "get_user":
            user_id = arguments["user_id"]
            resp = await client.get(
                f"{BASE_URL}/users/{user_id}",
                headers=headers
            )
            resp.raise_for_status()
            return [types.TextContent(type="text", text=resp.text)]

        elif name == "search_orders":
            params = {
                "status": arguments.get("status"),
                "limit": arguments.get("limit", 10)
            }
            # None 값 제거
            params = {k: v for k, v in params.items() if v is not None}

            resp = await client.get(
                f"{BASE_URL}/orders",
                headers=headers,
                params=params
            )
            resp.raise_for_status()
            return [types.TextContent(type="text", text=resp.text)]

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

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

---

실전 예시 3 — Slack 알림 보내기

# slack_mcp.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import httpx
import os

app = Server("slack-mcp")
SLACK_TOKEN = os.environ["SLACK_BOT_TOKEN"]

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="send_slack_message",
            description="Slack 채널에 메시지를 보낸다",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {
                        "type": "string",
                        "description": "채널명 (예: #dev-alert)"
                    },
                    "message": {
                        "type": "string",
                        "description": "보낼 메시지"
                    }
                },
                "required": ["channel", "message"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "send_slack_message":
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://slack.com/api/chat.postMessage",
                headers={"Authorization": f"Bearer {SLACK_TOKEN}"},
                json={
                    "channel": arguments["channel"],
                    "text": arguments["message"]
                }
            )
            result = resp.json()
            if result["ok"]:
                return [types.TextContent(type="text", text="✅ 메시지 전송 완료")]
            else:
                return [types.TextContent(type="text", text=f"❌ 전송 실패: {result['error']}")]

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

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

실제 사용:

나: "배포 완료됐어. #dev-alert에 알림 보내줘"
Claude: (send_slack_message 호출)
→ channel: #dev-alert
→ message: "🚀 배포 완료 - v2.3.1 (2026-04-13 15:30)"
→ ✅ 메시지 전송 완료

---

Claude에 MCP 서버 등록하기

방법 1. 명령어로 바로 추가

# 등록
claude mcp add my-postgres python /path/to/postgres_mcp.py

# 환경변수 필요한 경우
claude mcp add my-api \
  --env DATABASE_URL=postgresql://... \
  --env API_KEY=sk-... \
  python /path/to/api_mcp.py

# 등록된 서버 목록
claude mcp list

# 제거
claude mcp remove my-postgres

 

방법 2. 설정 파일로 추가

프로젝트 루트에 .claude/mcp.json 만들어요.

{
  "mcpServers": {
    "my-postgres": {
      "command": "python",
      "args": ["/path/to/postgres_mcp.py"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost/mydb"
      }
    },
    "my-slack": {
      "command": "python",
      "args": ["/path/to/slack_mcp.py"],
      "env": {
        "SLACK_BOT_TOKEN": "xoxb-..."
      }
    }
  }
}

이 파일 있으면 claude 실행 시 자동으로 연결돼요.

 

방법 3. 전역 설정 (모든 프로젝트에서 쓰고 싶을 때)

# ~/.claude/config.json 에 저장됨
claude mcp add --global my-postgres python /path/to/postgres_mcp.py

---

팀 프로젝트 권장 방식:

프로젝트 루트/
├── .claude/
│   ├── mcp.json       ← MCP 서버 설정 (git에 올림)
│   └── commands/
└── CLAUDE.md

mcp.json은 git에 올리되 API 키같은 민감 정보는 .env에서 읽게 해요.

{
  "mcpServers": {
    "my-postgres": {
      "command": "python",
      "args": ["./mcp/postgres_mcp.py"],
      "env": {
        "DATABASE_URL": "${DATABASE_URL}"
      }
    }
  }
}

이렇게 하면 팀원이 클론 받으면 MCP 설정도 같이 받아요. 

---

툴 설계 팁

Claude가 툴을 잘 쓰게 하려면 description을 명확하게 써야 해요.

# 나쁜 예
types.Tool(
    name="get_data",
    description="데이터를 가져온다",
    ...
)

# 좋은 예
types.Tool(
    name="get_order_stats",
    description="""주문 통계를 조회한다.
    특정 기간의 주문 건수, 총 매출, 평균 주문금액을 반환한다.
    날짜 미입력 시 오늘 기준으로 조회한다.""",
    ...
)

description이 좋을수록 Claude가 언제 이 툴을 써야 하는지 정확하게 판단해요.

---

전후 비교

MCP 없이:

나: "이번 주 매출 어때?"
Claude: "저는 DB에 접근할 수 없어요."
나: (DBeaver 열고 SQL 직접 실행)
나: (결과 복사해서 Claude한테 붙여넣기)
나: "이 데이터 분석해줘"
Claude: "매출이 전주 대비 15% 증가했네요..."
(총 5분)

MCP 있을 때:

나: "이번 주 매출 어때?"
Claude: (query_db 자동 호출)
→ "이번 주 매출 4,820만원, 전주 대비 15% 증가.
   월요일이 가장 높고(920만원), 목요일이 가장 낮아요(540만원)."
(총 5초)

 

---

 

📌 관련 글

MCP 서버 팀 배포 가이드

MCP 서버 보안 설정 완전 가이드 — 인증, 권한 제한, 위험 차단

MCP 보안 설정 완전 가이드

MCP 서버 팀 배포 가이드 — 로컬에서 서버로 올려서 팀 전체가 쓰기