mcp

Adım Adım Bir MCP Sunucusu Nasıl Oluşturulur

Model Bağlam Protokolü (MCP), yapay zeka modellerini gerçek dünyadaki araçlara ve hizmetlere bağlamak için yeni standart haline geliyor. Bir MCP sunucusu oluşturmak, verileri, eylemleri ve kaynakları basit, standartlaştırılmış bir arayüz aracılığıyla Claude gibi bir LLM’ye maruz bırakmanıza olanak tanır.

Bu kılavuzda, Python’da temel bir MCP sunucusunun nasıl kurulacağını, kaynakların ve araçların nasıl tanımlanacağını ve bir MCP istemcisine nasıl bağlanacağını adım adım öğreneceksiniz.

Önemli Çıkarımlar

  • MCP sunucuları, AI modellerinin standartlaştırılmış kaynaklar ve araçlar aracılığıyla harici sistemlerle etkileşime girmesine izin verir.
  • Resmi SDK’yı kullanarak Python’da bir MCP sunucusu oluşturabilirsiniz.
  • En az çalışan bir sunucu hem salt okunur verileri (kaynaklar) hem de yürütülebilir eylemleri (araçlar) kullanıma sunabilir.
  • Güvenlik ve hata işleme, üretim dağıtımları için kritik öneme sahiptir.

MCP Sunucusu Nedir?

Bir MCP sunucusu, bir LLM ile veritabanı, dosya depolama veya API gibi harici bir sistem arasında bir köprü görevi görür. Kaynakları (okunabilir veriler), araçları (eylemler) ve istemleri (talimatları) LLM’nin görevleri sırasında güvenle kullanabileceği şekilde tanımlar.

MCP, her model veya araç için özel bir entegrasyon yazmak yerine, protokol sürümü 0.1 (Nisan 2025 itibarıyla geçerli) ile çalışan evrensel bir standart sunar.

Başlamadan önce ihtiyacımız olanlar

  • Python 3.8 veya üzeri
  • Python komut dosyası oluşturma ile ilgili temel deneyim
  • Python için MCP SDK’sı (pip aracılığıyla kullanılabilir)
  • Claude Desktop veya Cursor gibi MCP uyumlu bir istemci (test için isteğe bağlı)
  • Sürüm denetimi için Git (önerilir)
  • Bir metin düzenleyicisi veya IDE (Visual Studio Code önerilir)

Çekirdek yapıyı anlama

MCP’de:

  • Server: LLM’ye kaynaklar ve araçlar sağlar.
  • Client: LLM’yi sunucunuza bağlar.
  • Protocol: İstemci ve sunucu arasındaki iletişimi yönetir.

İki önemli temel öğe tanımlayacaksınız:

  • Resource: LLM’nin okuyabileceği statik veya dinamik bilgiler.
  • Tool: LLM’nin yürütebileceği çağrılabilir bir işlev.

İletişim akışı şu şekilde çalışır:

  1. LLM (bir istemci aracılığıyla) sunucunuzdan veri veya eylemler talep eder
  2. Sunucunuz bu istekleri işler ve standartlaştırılmış yanıtlar döndürür
  3. LLM daha sonra bu bilgileri akıl yürütme ve yanıtlarında kullanabilir

1. Python projenizi kurun

Bir proje dizini ve bir Python sanal ortamı oluşturarak başlayın.

mkdir my_mcp_server
cd my_mcp_server
python -m venv venv
source venv/bin/activate  # Linux/Mac
venvScriptsactivate     # Windows

Temel bir proje yapısı oluşturun:

mkdir -p src/resources src/tools tests
touch src/__init__.py src/resources/__init__.py src/tools/__init__.py
touch requirements.txt README.md

requirements.txt sitenize aşağıdakileri ekleyin:

mcp-server>=0.1.0
pydantic>=2.0.0
pytest>=7.0.0

2. MCP SDK’sını yükleyin

Python ve diğer bağımlılıklar için MCP sunucusu SDK’sını yükleyin:

pip install -r requirements.txt

Resmi SDK henüz yayınlanmadıysa, bir GitHub deposundan yüklemeniz gerekebilir:

pip install git+https://github.com/anthropic/mcp-server-python.git

3. Temel bir MCP sunucusu oluşturun

src/server.py adlı bir dosya oluşturun:

from typing import Dict, Any
from mcp_server import MCPServer
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_server")

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Resources and tools will be added here
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Bu, uygun günlük kaydı ile temel bir MCP sunucusu kurar.

4. Bir kaynak tanımlayın

Kaynaklar, modelin okuyabileceği verileri kullanıma sunar. Bir dosya oluşturalım src/resources/user_profiles.py:

from typing import List, Dict, Any
from pydantic import BaseModel
import logging

logger = logging.getLogger("mcp_server.resources")

class UserProfile(BaseModel):
    """Data model for user profiles."""
    name: str
    role: str
    department: str = "General"
    years_experience: int = 0

def fetch_user_profiles() -> List[Dict[str, Any]]:
    """
    Fetch user profiles from the database.
    
    Returns:
        List[Dict[str, Any]]: A list of user profile dictionaries.
    """
    try:
        # In a real implementation, this would query a database
        # For this example, we'll return mock data
        users = [
            UserProfile(name="Alice", role="Engineer", department="Engineering", years_experience=5),
            UserProfile(name="Bob", role="Product Manager", department="Product", years_experience=3),
            UserProfile(name="Charlie", role="Designer", department="Design", years_experience=7)
        ]
        
        logger.info(f"Successfully fetched {len(users)} user profiles")
        return [user.model_dump() for user in users]
    except Exception as e:
        logger.error(f"Error fetching user profiles: {e}")
        # In production, you might want to return an empty list or raise
        # a specific exception depending on your error handling strategy
        return []

Şimdi src/server.py bu kaynağı içerecek şekilde güncelleştirin:

from typing import Dict, Any
from mcp_server import MCPServer, Resource
import logging
from src.resources.user_profiles import fetch_user_profiles

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        server.add_resource(user_profiles)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

LLM artık MCP istemcisi aracılığıyla user_profiles sorgulayabilir.

5. Aracı tanımlayın

Araçlar, LLM’nin bir eylemi yürütmesine izin verir. Bir dosya oluşturun src/tools/user_management.py:

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
import logging

logger = logging.getLogger("mcp_server.tools")

class CreateUserRequest(BaseModel):
    """Validation model for user creation requests."""
    name: str = Field(..., min_length=2, description="User's full name")
    role: str = Field(..., min_length=2, description="User's job role")
    department: Optional[str] = Field("General", description="User's department")
    years_experience: Optional[int] = Field(0, ge=0, description="Years of professional experience")

def create_user_profile(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Create a new user profile in the database.
    
    Args:
        request_data (Dict[str, Any]): User data containing name, role, etc.
        
    Returns:
        Dict[str, Any]: Response with status and user info
    """
    try:
        # Validate the input data
        user_data = CreateUserRequest(**request_data)
        
        # In a real implementation, this would insert into a database
        # For this example, we'll just log the action
        logger.info(f"Creating new user: {user_data.name} - {user_data.role} in {user_data.department}")
        
        # Return success response with created user data
        return {
            "status": "success",
            "message": f"User {user_data.name} created successfully",
            "user": user_data.model_dump()
        }
    except ValidationError as e:
        # Handle validation errors
        logger.error(f"Validation error: {e}")
        return {
            "status": "error",
            "message": "Invalid user data provided",
            "details": str(e)
        }
    except Exception as e:
        # Handle other errors
        logger.error(f"Error creating user: {e}")
        return {
            "status": "error",
            "message": "Failed to create user",
            "details": str(e)
        }

Şimdi src/server.py bu aracı içerecek şekilde güncelleyin:

from typing import Dict, Any
from mcp_server import MCPServer, Resource, Tool
import logging
from src.resources.user_profiles import fetch_user_profiles
from src.tools.user_management import create_user_profile

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        # Add the create user tool
        create_user = Tool(
            name="create_user_profile",
            description="Create a new user profile in the database.",
            parameters={
                "name": {"type": "string", "description": "User's full name"},
                "role": {"type": "string", "description": "User's job role"},
                "department": {"type": "string", "description": "User's department (optional)"},
                "years_experience": {"type": "integer", "description": "Years of experience (optional)"}
            },
            execute_fn=create_user_profile
        )
        
        server.add_resource(user_profiles)
        server.add_tool(create_user)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

6. Hata işleme ve doğrulama

Doğrulama mantığınızı merkezileştirmek için bir dosya src/utils/validation.py oluşturun:

from typing import Dict, Any, List, Optional, Type
from pydantic import BaseModel, ValidationError
import logging

logger = logging.getLogger("mcp_server.validation")

def validate_request(
    data: Dict[str, Any],
    model_class: Type[BaseModel]
) -> tuple[Optional[BaseModel], Optional[Dict[str, Any]]]:
    """
    Validate request data against a Pydantic model.
    
    Args:
        data: The input data to validate
        model_class: The Pydantic model class to use for validation
        
    Returns:
        tuple: (validated_model, error_dict)
            - If valid: (model instance, None)
            - If invalid: (None, error dictionary)
    """
    try:
        validated_data = model_class(**data)
        return validated_data, None
    except ValidationError as e:
        errors = e.errors()
        error_dict = {
            "status": "error",
            "message": "Validation failed",
            "errors": errors
        }
        logger.error(f"Validation error: {errors}")
        return None, error_dict

Bu yardımcı program işlevi, giriş verilerini tutarlı bir şekilde doğrulamak için tüm araçlarınızda kullanılabilir.

7. MCP sunucusunu çalıştırın ve test edin

Sunucunuzun çalıştığını doğrulamak için test_server.py basit bir test komut dosyası oluşturun:

import requests
import json
import time
import subprocess
import sys
from pathlib import Path

def test_server():
    """Simple test to verify the MCP server is running correctly."""
    # Start the server in a separate process
    server_process = subprocess.Popen([sys.executable, "src/server.py"])
    
    try:
        # Wait for server to start
        time.sleep(2)
        
        # Test the server using the MCP client
        # In a real test, you would use the MCP client SDK
        # For this example, we'll simulate a client using HTTP requests
        
        # Assuming the server is running on localhost:8000
        base_url = "http://localhost:8000"
        
        # Test fetching resources
        response = requests.get(f"{base_url}/resources/user_profiles")
        assert response.status_code == 200
        data = response.json()
        print("Resource response:", json.dumps(data, indent=2))
        
        # Test executing a tool
        tool_data = {
            "name": "Test User",
            "role": "Tester",
            "department": "QA"
        }
        response = requests.post(
            f"{base_url}/tools/create_user_profile",
            json=tool_data
        )
        assert response.status_code == 200
        data = response.json()
        print("Tool response:", json.dumps(data, indent=2))
        
        print("All tests passed!")
        
    finally:
        # Clean up: terminate the server process
        server_process.terminate()
        server_process.wait()

if __name__ == "__main__":
    test_server()

Sunucunuzu çalıştırın:

python src/server.py

Ayrı bir terminalde, sunucu çalışırken şuna benzer bir çıktı görebilirsiniz:

2025-04-28 10:15:23 - mcp_server - INFO - Starting MCP server...
2025-04-28 10:15:23 - mcp_server - INFO - Server listening on 0.0.0.0:8000
2025-04-28 10:15:30 - mcp_server.resources - INFO - Successfully fetched 3 user profiles
2025-04-28 10:15:35 - mcp_server.tools - INFO - Creating new user: Test User - Tester in QA

Ardından, sunucu URL’sini veya sunucuyu başlatma komutunu sağlayarak MCP istemcinizi (Claude Desktop gibi) yerel MCP sunucunuza bağlanacak şekilde yapılandırın.

Güvenlikle ilgili dikkat edilmesi gerekenler

Bir MCP sunucusu dağıtırken, şu en iyi güvenlik uygulamalarını göz önünde bulundurun:

  1. Kimlik doğrulaması: İstemcilerin kimliğini doğrulamak için API anahtarlarını veya OAuth’u uygulayın.
def authenticate_request(request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != os.environ.get("MCP_API_KEY"):
        raise ValueError("Invalid API key")
  1. Input Validation: Tüm girişleri her zaman Pydantic modellerini kullanarak doğrulayın.
  2. Rate Limiting: Kötüye kullanımı önlemek için hız sınırlaması uygulayın.
  3. HTTPS: Üretimde her zaman HTTPS kullanın.
  4. Restricted Actions: Araçların yapabileceklerine ilişkin net sınırlar tanımlayın.

Performans optimizasyonu

  1. Caching: Pahalı kaynak getirmelerini önbelleğe alın:
from functools import lru_cache

@lru_cache(maxsize=128, ttl=300)  # Cache for 5 minutes
def fetch_user_profiles():
    # Expensive database query
    pass
  1. Async Processing: G/Ç’ye bağlı işlemler için async kullanın:
async def fetch_user_profiles():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/users") as response:
            data = await response.json()
            return data
  1. Connection Pooling: Veritabanı erişimi için bağlantı havuzlarını kullanın.

Dağıtım

Yerel kalkınma

Yerel geliştirme için şunu çalıştırın:

python src/server.py

Docker dağıtımı

Bir Dockerfile oluşturun:

FROM python:3.10-slim

WORKDIR /app

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

COPY . .

EXPOSE 8000

CMD ["python", "src/server.py"]

Oluşturun ve çalıştırın:

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

Bulut dağıtımı (AWS)

  1. Bir EC2 bulut sunucusu oluşturun veya AWS App Runner’ı kullanın
  2. Docker kapsayıcınızı dağıtma
  3. Application Load Balancer ayarlama
  4. Erişimi kısıtlamak için güvenlik gruplarını yapılandırma

MCP sunucunuzu test etme

Bir test dosyası oluşturun tests/test_resources.py:

import pytest
from src.resources.user_profiles import fetch_user_profiles

def test_fetch_user_profiles():
    """Test that user profiles can be fetched successfully."""
    profiles = fetch_user_profiles()
    
    # Check structure
    assert isinstance(profiles, list)
    assert len(profiles) > 0
    
    # Check content
    first_profile = profiles[0]
    assert "name" in first_profile
    assert "role" in first_profile
    assert isinstance(first_profile["name"], str)

Tests dosyasını şununla çalıştırın:

pytest

Yaygın hatalar ve sorun giderme

Sorun Çözüm Örnek MCP sunucusuna bağlanılamıyor Sunucunuzun çalıştığını ve bağlantı noktasının doğru olduğunu kontrol edin netstat -tulpn | grep 8000 LLM kaynakları bulamıyor Ad ve açıklama alanlarının doğru ayarlandığını doğrulayın Kaynak başlatmanızı kontrol edin Araç yürütmedeki hatalar Giriş parametrelerinin beklenen türlerle eşleştiğini doğrulayın Doğrulama için Pydantic kullanın İstemci çıktıyı ayrıştıramıyor İşlevlerinizin JSON serileştirilebilir verileri döndürdüğünden emin olun Özel nesneler yerine .model_dump() kullanın Sunucu Başlangıçta çöküyor İçe aktarmalarınızı ve ortam değişkenlerinizi kontrol edin Ayrıntılı günlük kaydı için DEBUG=True olarak ayarlayın Araç zaman aşımı Harici API çağrıları için zaman aşımı işleme ekleme Zaman aşımı ile asyncio.wait_for() kullanın Kimlik doğrulama hataları API anahtarlarını ve izinleri doğrulayın İstek başlıklarını kontrol edin XML/JSON ayrıştırma hataları Uygun içerik türü başlıklarını kullanın İçerik türünü ayarlayın: application/json

Sonraki adımlar

Temel MCP sunucunuzu oluşturduktan sonra, şu gelişmiş uzantıları göz önünde bulundurun:

  1. Database Integration: PostgreSQL, MongoDB veya diğer veritabanlarına bağlanın.
  2. File Operations: Dosya okuma, yazma ve dönüştürme için araçlar ekleyin.
  3. External APIs: GitHub, Slack veya Google Drive gibi popüler hizmetlerle entegre edin.
  4. Webhooks: LLM’nin diğer sistemlerdeki olayları tetiklemesine izin verin.
  5. Streaming Resources: Büyük veri kümelerinin akışını destekleyin.
  6. Context-Aware Actions: LLM’nin mevcut bağlamını anlayan araçlar ekleyin.

Örnek: Veritabanı bağlantısı ekleme

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """Create a database connection context manager."""
    conn = None
    try:
        conn = psycopg2.connect(
            host=os.environ.get("DB_HOST"),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD")
        )
        yield conn
    finally:
        if conn is not None:
            conn.close()

def fetch_user_profiles_from_db():
    """Fetch user profiles from a PostgreSQL database."""
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT name, role, department FROM users")
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

Özetle

Python’da basit bir MCP sunucusu oluşturmak, LLM’leri çok daha güçlü hale getirmenin kapısını açar. Verileri ve eylemleri temiz, standartlaştırılmış bir protokol aracılığıyla açığa çıkararak, yapay zeka sistemlerinin harici hizmetlerle güvenli ve anlamlı bir şekilde etkileşim kurmasını kolaylaştırırsınız.

Küçük adımlarla başlayın, tek bir kaynağa ve tek bir araca odaklanın ve zaman içinde veritabanları, bulut depolama veya dahili API’ler gibi daha gelişmiş kullanım durumlarına genişletebilirsiniz.

MCP ekosistemi hızla büyüyor ve bu standartları şimdi uygulamak, uygulamalarınızı hem protokoldeki hem de onu kullanan LLM’lerdeki iyileştirmelerden yararlanacak şekilde konumlandıracaktır.

Yazının orijinalini buradan okuyabilirsiniz.