Python으로 MCP 서버 구축 및 Claude 연동 가이드

Python으로 MCP 서버 구축 및 Claude 연동 가이드

안녕하세요! 이 블로그에서는 Python을 사용하여 MCP(Message Communication Protocol) 서버를 구축하고, 이를 Claude AI와 연동하여 사용하는 방법에 대해 상세히 알아보겠습니다.

목차

1. MCP 서버란?
2. 환경 설정
3. Python으로 MCP 서버 구현하기
4. Claude API 준비하기
5. MCP 서버와 Claude 연동하기
6. 서비스 배포 및 실행하기
7. 응용 및 확장 방법
8. 문제 해결 및 FAQ

MCP 서버란?

MCP(Message Communication Protocol) 서버는 클라이언트와 AI 모델 사이의 메시지 교환을 관리하는 중개 서버입니다. 이 서버를 통해 사용자의 요청을 처리하고, Claude와 같은 AI 모델에 전달하여 응답을 받아올 수 있습니다.

MCP 서버의 주요 기능:

• 메시지 포맷팅 및 전처리
• API 요청 관리 및 에러 핸들링
• 응답 후처리 및 클라이언트 전달
• 대화 기록 관리 및 컨텍스트 유지

환경 설정

먼저 필요한 환경을 설정해보겠습니다.

필수 요구사항

• Python 3.8 이상
• pip (Python 패키지 관리자)
• 가상환경 (권장: venv 또는 conda)

가상환경 설정 및 패키지 설치

# 가상환경 생성

python -m venv mcp_env

 

# 가상환경 활성화 (Windows)

mcp_env\Scripts\activate

 

# 가상환경 활성화 (macOS/Linux)

source mcp_env/bin/activate

 

# 필요한 패키지 설치

pip install fastapi uvicorn anthropic python-dotenv pydantic

Python으로 MCP 서버 구현하기

이제 FastAPI를 사용하여 MCP 서버를 구현해보겠습니다.

1. 프로젝트 구조 설정

mcp_server/

├── .env                  # 환경 변수 파일 (API 키 등)

├── main.py               # FastAPI 애플리케이션 진입점

├── requirements.txt      # 의존성 패키지 목록

├── app/

│   ├── __init__.py

│   ├── api/

│   │   ├── __init__.py

│   │   └── endpoints.py  # API 엔드포인트 정의

│   ├── core/

│   │   ├── __init__.py

│   │   ├── config.py     # 설정 관리

│   │   └── security.py   # 보안 관련 기능

│   ├── models/

│   │   ├── __init__.py

│   │   └── message.py    # 데이터 모델 정의

│   └── services/

│       ├── __init__.py

│       └── claude.py     # Claude API 통신 서비스

└── tests/                # 테스트 코드

2. 환경 변수 설정 (.env)

# .env 파일

CLAUDE_API_KEY=your_claude_api_key_here

ENVIRONMENT=development

3. 설정 관리 (config.py)

# app/core/config.py

from pydantic import BaseSettings

from dotenv import load_dotenv

import os

 

load_dotenv()

 

class Settings(BaseSettings):

    API_V1_STR: str = "/api/v1"

    PROJECT_NAME: str = "MCP Server"

    ENVIRONMENT: str = os.getenv("ENVIRONMENT", "development")

    CLAUDE_API_KEY: str = os.getenv("CLAUDE_API_KEY")

   

    class Config:

        case_sensitive = True

 

settings = Settings()

4. 데이터 모델 정의 (message.py)

# app/models/message.py

from pydantic import BaseModel

from typing import List, Optional

 

class Message(BaseModel):

    role: str

    content: str

 

class ConversationRequest(BaseModel):

    messages: List[Message]

    max_tokens: Optional[int] = 1000

    temperature: Optional[float] = 0.7

 

class ConversationResponse(BaseModel):

    message: Message

    conversation_id: str

5. Claude 서비스 구현 (claude.py)

# app/services/claude.py

import anthropic

import uuid

from app.core.config import settings

from app.models.message import Message, ConversationRequest, ConversationResponse

 

class ClaudeService:

    def __init__(self):

        self.client = anthropic.Client(api_key=settings.CLAUDE_API_KEY)

        self.model = "claude-3-7-sonnet-20250219"  # 최신 모델 버전으로 업데이트 필요할 수 있음

   

    async def generate_response(self, request: ConversationRequest) -> ConversationResponse:

        try:

            # 메시지 형식 변환

            messages = [{"role": msg.role, "content": msg.content} for msg in request.messages]

           

            # Claude API 호출

            response = self.client.messages.create(

                model=self.model,

                messages=messages,

                max_tokens=request.max_tokens,

                temperature=request.temperature

            )

           

            # 응답 처리

            response_message = Message(

                role="assistant",

                content=response.content[0].text

            )

           

            # 대화 ID 생성 (실제 구현에서는 데이터베이스와 연동 필요)

            conversation_id = str(uuid.uuid4())

           

            return ConversationResponse(

                message=response_message,

                conversation_id=conversation_id

            )

           

        except Exception as e:

            # 에러 처리

            print(f"Error communicating with Claude API: {str(e)}")

            raise

6. API 엔드포인트 정의 (endpoints.py)

# app/api/endpoints.py

from fastapi import APIRouter, Depends, HTTPException

from app.models.message import ConversationRequest, ConversationResponse

from app.services.claude import ClaudeService

 

router = APIRouter()

claude_service = ClaudeService()

 

@router.post("/chat", response_model=ConversationResponse)

async def chat(request: ConversationRequest):

    """

    채팅 메시지를 Claude에 전송하고 응답을 받아옵니다.

    """

    try:

        response = await claude_service.generate_response(request)

        return response

    except Exception as e:

        raise HTTPException(status_code=500, detail=str(e))

 

@router.get("/health")

async def health_check():

    """

    서버 상태 확인용 엔드포인트

    """

    return {"status": "healthy"}

7. FastAPI 애플리케이션 설정 (main.py)

# main.py

from fastapi import FastAPI

from fastapi.middleware.cors import CORSMiddleware

from app.api.endpoints import router as api_router

from app.core.config import settings

 

app = FastAPI(

    title=settings.PROJECT_NAME,

    openapi_url=f"{settings.API_V1_STR}/openapi.json"

)

 

# CORS 설정

app.add_middleware(

    CORSMiddleware,

    allow_origins=["*"],  # 프로덕션에서는 특정 도메인으로 제한해야 합니다

    allow_credentials=True,

    allow_methods=["*"],

    allow_headers=["*"],

)

 

# API 라우터 등록

app.include_router(api_router, prefix=settings.API_V1_STR)

 

if __name__ == "__main__":

    import uvicorn

    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

8. requirements.txt 생성

fastapi>=0.95.0

uvicorn>=0.21.1

anthropic>=0.5.0

python-dotenv>=1.0.0

pydantic>=1.10.7

Claude API 준비하기

Claude와 통신하기 위해서는 Anthropic의 API 키가 필요합니다.

1. Anthropic 웹사이트에 가입합니다.
2. API 섹션에서 API 키를 생성합니다.
3. 생성된 API 키를 .env 파일의 CLAUDE_API_KEY 변수에 설정합니다.
4. 
   

MCP 서버와 Claude 연동하기

이제 MCP 서버가 Claude API와 어떻게 연동되는지 살펴보겠습니다.

데이터 흐름

1. 클라이언트가 /api/v1/chat 엔드포인트로 POST 요청을 보냅니다.
2. 요청은 FastAPI 라우터를 통해 처리됩니다.
3. ClaudeService가 요청을 처리하여 Anthropic API로 전달합니다.
4. Claude의 응답을 받아 클라이언트에게 반환합니다.
5. 
   

클라이언트 예제 코드 (Python)

import requests

import json

 

# MCP 서버 URL

MCP_SERVER_URL = "http://localhost:8000/api/v1/chat"

 

# 요청 데이터

request_data = {

    "messages": [

        {"role": "user", "content": "안녕하세요, Claude!"}

    ],

    "max_tokens": 1000,

    "temperature": 0.7

}

 

# API 호출

response = requests.post(

    MCP_SERVER_URL,

    headers={"Content-Type": "application/json"},

    data=json.dumps(request_data)

)

 

# 응답 처리

if response.status_code == 200:

    result = response.json()

    print(f"Claude: {result['message']['content']}")

    print(f"Conversation ID: {result['conversation_id']}")

else:

    print(f"Error: {response.status_code} - {response.text}")

서비스 배포 및 실행하기

로컬에서 테스트

# 가상환경 활성화

source mcp_env/bin/activate  # Linux/macOS

# 또는

mcp_env\Scripts\activate  # Windows

 

# 서버 실행

python main.py

서버는 기본적으로 http://localhost:8000에서 실행됩니다.

Docker로 배포하기

# Dockerfile

FROM python:3.9-slim

 

WORKDIR /app

 

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

 

COPY . .

 

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker 빌드 및 실행:

# Docker 이미지 빌드

docker build -t mcp-server .

 

# Docker 컨테이너 실행

docker run -d -p 8000:8000 --env-file .env --name mcp-container mcp-server

클라우드 서비스에 배포

AWS, GCP, Azure 등의 클라우드 서비스에 배포할 수 있습니다. 예를 들어, AWS Elastic Beanstalk를 사용한 배포 과정은 다음과 같습니다:

1. AWS Elastic Beanstalk CLI 설치
2. 애플리케이션 초기화 및 환경 설정
3. 배포 수행

# Elastic Beanstalk CLI 설치

pip install awsebcli

 

# 애플리케이션 초기화

eb init -p python-3.8 mcp-server --region us-east-1

 

# 환경 생성 및 배포

eb create mcp-environment

응용 및 확장 방법

1. 대화 기록 관리

데이터베이스를 추가하여 대화 기록을 저장하고 관리할 수 있습니다:

# app/services/conversation.py

from pymongo import MongoClient

from app.core.config import settings

from datetime import datetime

 

class ConversationService:

    def __init__(self):

        self.client = MongoClient(settings.MONGODB_URI)

        self.db = self.client.mcp_database

        self.conversations = self.db.conversations

   

    async def create_conversation(self, user_id):

        conversation = {

            "user_id": user_id,

            "messages": [],

            "created_at": datetime.utcnow(),

            "updated_at": datetime.utcnow()

        }

        result = self.conversations.insert_one(conversation)

        return str(result.inserted_id)

   

    async def add_message(self, conversation_id, message):

        self.conversations.update_one(

            {"_id": conversation_id},

            {

                "$push": {"messages": message},

                "$set": {"updated_at": datetime.utcnow()}

            }

        )

   

    async def get_conversation(self, conversation_id):

        return self.conversations.find_one({"_id": conversation_id})

2. 사용자 인증 추가

JWT를 사용한 사용자 인증 시스템을 추가할 수 있습니다:

# app/core/security.py

from datetime import datetime, timedelta

from jose import JWTError, jwt

from fastapi import Depends, HTTPException, status

from fastapi.security import OAuth2PasswordBearer

 

SECRET_KEY = "your-secret-key"  # 실제 구현시 안전하게 관리해야 함

ALGORITHM = "HS256"

ACCESS_TOKEN_EXPIRE_MINUTES = 30

 

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

 

def create_access_token(data: dict):

    to_encode = data.copy()

    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)

    to_encode.update({"exp": expire})

    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    return encoded_jwt

 

def verify_token(token: str = Depends(oauth2_scheme)):

    credentials_exception = HTTPException(

        status_code=status.HTTP_401_UNAUTHORIZED,

        detail="Could not validate credentials",

        headers={"WWW-Authenticate": "Bearer"},

    )

    try:

        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        username: str = payload.get("sub")

        if username is None:

            raise credentials_exception

        return {"username": username}

    except JWTError:

        raise credentials_exception

3. 로드 밸런싱 및 스케일링

큰 규모의 시스템에서는 다음과 같은 방법으로 확장할 수 있습니다:

• Nginx와 같은 리버스 프록시 사용
• Docker Swarm이나 Kubernetes를 사용한 컨테이너 오케스트레이션
• 여러 인스턴스에 걸친 로드 밸런싱
• 
  

문제 해결 및 FAQ

Q: Claude API 연결이 안 되는 경우는 어떻게 해결하나요?

A: 다음 사항을 확인해보세요:

• API 키가 올바르게 설정되었는지 확인
• 네트워크 연결 상태 확인
• API 요청 제한(rate limit)에 도달했는지 확인
• Anthropic 서비스 상태 확인

Q: 서버 성능을 개선하려면 어떻게 해야 하나요?

A: 다음과 같은 방법을 고려해보세요:

• 응답 캐싱 구현
• 비동기 처리 최적화
• 데이터베이스 인덱싱 최적화
• 수평적 확장을 위한 아키텍처 구성

Q: 대화 컨텍스트를 유지하는 방법은?

A: Claude API에 이전 메시지를 포함하여 컨텍스트를 유지할 수 있습니다. 데이터베이스에 대화 기록을 저장하고, 새 요청 시 이전 메시지들을 포함시키는 방식으로 구현할 수 있습니다.

결론

이 블로그에서는 Python과 FastAPI를 사용하여 MCP 서버를 구축하고 Claude API와 연동하는 방법을 살펴보았습니다. 이러한 서버는 AI 모델과의 통신을 관리하고, 추가 기능을 구현하는 데 유용하게 사용될 수 있습니다.

실제 프로덕션 환경에서는 보안, 인증, 데이터베이스 연동 등의 추가 작업이 필요하며, 사용 사례에 맞게 코드를 수정하고 최적화하는 것이 중요합니다.

 

또 다른 방법

소스코드를 찾아서 다운로드 받습니다.

https://github.com/modelcontextprotocol

https://github.com/modelcontextprotocol/servers/tree/main/src

Claude 에게 소스들을 첨부하고 공부를 시킵니다.

프롬프트를 주고 코드 생성을 요청합니다.

'''

새로운 서버를 만들거야. 서버의 이름은 counting-chars

역할은 주어진 문자열 안에 특정 문자나 단어가 몇번 반복되는지를 세는 역할이야.

재료는 두개가 주어져. 첫번째는 문자열, 두번째는 문자열 안에 반복될것으로 여겨지는 문자나 문자열.

응답은 이게 몇번 반복되는지에 대한 숫자.

'''

코드가 생성되면 폴더에 넣습니다.

node.js를 설치합니다.

해당 폴더에서 CLI를 입력합니다.

C:\>npm i; npm run build

dist 폴더가 생성되고 index.js 파일이 생성됩니다.

Claude MCP 설정 파일에 등록한다. claude_desktop_config.json

{

    "mcpServers": {

        "counting-chars":{

            "command": "node",

            "args": [

                "C:\\servers-main\\src\\counting-chars dist\\index.js"

            ]

        }

    }

}

이제 Claude desktop에서 테스트할 수 있습니다.

MCP 서버 동작 예제

• Time - Time and timezone conversion capabilities
• Claude Desktop에서 파일 관리와 유튜브 요약 기능 설정하기