Files

T

wehub-resource-sync c275ba2868 chore: import upstream snapshot with attribution

2026-07-13 13:31:35 +08:00

20 KiB

Raw Permalink Blame History

MCP Protokol Özellikleri Derinlemesine İnceleme

Bu rehber, temel araç ve kaynak yönetiminin ötesine geçen gelişmiş MCP protokol özelliklerini inceler. Bu özellikleri anlamak, daha sağlam, kullanıcı dostu ve üretim hazır MCP sunucuları oluşturmanıza yardımcı olur.

Kapsanan Özellikler

İlerleme Bildirimleri - Uzun süren işlemler için ilerlemeyi bildirin
İstek İptali - İstemcilerin devam eden istekleri iptal etmesine izin verin
Kaynak Şablonları - Parametreli dinamik kaynak URI'ları
Sunucu Yaşam Döngüsü Olayları - Doğru başlatma ve kapatma
Günlük Kaydı Kontrolü - Sunucu tarafı günlük yapılandırması
Hata İşleme Kalıpları - Tutarlı hata yanıtları

1. İlerleme Bildirimleri

Zaman alan işlemler (veri işleme, dosya indirme, API çağrıları) için ilerleme bildirimleri kullanıcıları bilgilendirir.

Nasıl Çalışır

sequenceDiagram
    participant Client
    participant Server
    
    Client->>Server: tools/call (uzun işlem)
    Server-->>Client: bildirim: ilerleme %10
    Server-->>Client: bildirim: ilerleme %50
    Server-->>Client: bildirim: ilerleme %90
    Server->>Client: sonuç (tamamlandı)

Python Uygulaması

from mcp.server import Server, NotificationOptions
from mcp.types import ProgressNotification
import asyncio

app = Server("progress-server")

@app.tool()
async def process_large_file(file_path: str, ctx) -> str:
    """Process a large file with progress updates."""
    
    # İlerleme hesaplaması için dosya boyutunu al
    file_size = os.path.getsize(file_path)
    processed = 0
    
    with open(file_path, 'rb') as f:
        while chunk := f.read(8192):
            # Parçayı işle
            await process_chunk(chunk)
            processed += len(chunk)
            
            # İlerleme bildirimini gönder
            progress = (processed / file_size) * 100
            await ctx.send_notification(
                ProgressNotification(
                    progressToken=ctx.request_id,
                    progress=progress,
                    total=100,
                    message=f"Processing: {progress:.1f}%"
                )
            )
    
    return f"Processed {file_size} bytes"

@app.tool()
async def batch_operation(items: list[str], ctx) -> str:
    """Process multiple items with progress."""
    
    results = []
    total = len(items)
    
    for i, item in enumerate(items):
        result = await process_item(item)
        results.append(result)
        
        # Her öğeden sonra ilerlemeyi bildir
        await ctx.send_notification(
            ProgressNotification(
                progressToken=ctx.request_id,
                progress=i + 1,
                total=total,
                message=f"Processed {i + 1}/{total}: {item}"
            )
        )
    
    return f"Completed {total} items"

TypeScript Uygulaması

import { Server } from "@modelcontextprotocol/sdk/server/index.js";

server.setRequestHandler(CallToolSchema, async (request, extra) => {
  const { name, arguments: args } = request.params;
  
  if (name === "process_data") {
    const items = args.items as string[];
    const results = [];
    
    for (let i = 0; i < items.length; i++) {
      const result = await processItem(items[i]);
      results.push(result);
      
      // İlerleme bildirimi gönder
      await extra.sendNotification({
        method: "notifications/progress",
        params: {
          progressToken: request.id,
          progress: i + 1,
          total: items.length,
          message: `Processing item ${i + 1}/${items.length}`
        }
      });
    }
    
    return { content: [{ type: "text", text: JSON.stringify(results) }] };
  }
});

İstemci İşleme (Python)

async def handle_progress(notification):
    """Handle progress notifications from server."""
    params = notification.params
    print(f"Progress: {params.progress}/{params.total} - {params.message}")

# İşleyici kaydet
session.on_notification("notifications/progress", handle_progress)

# Aracı çağır (ilerleme güncellemeleri işleyici aracılığıyla gelecektir)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})

2. İstek İptali

İstemcilerin artık gerekmeyen veya çok uzun süren istekleri iptal etmelerine izin verin.

Python Uygulaması

from mcp.server import Server
from mcp.types import CancelledError
import asyncio

app = Server("cancellable-server")

@app.tool()
async def long_running_search(query: str, ctx) -> str:
    """Search that can be cancelled."""
    
    results = []
    
    try:
        for page in range(100):  # Birçok sayfada ara
            # İptal isteğinin olup olmadığını kontrol et
            if ctx.is_cancelled:
                raise CancelledError("Search cancelled by user")
            
            # Sayfa aramasını simüle et
            page_results = await search_page(query, page)
            results.extend(page_results)
            
            # Küçük gecikme iptal kontrollerine olanak tanır
            await asyncio.sleep(0.1)
            
    except CancelledError:
        # Kısmi sonuçları döndür
        return f"Cancelled. Found {len(results)} results before cancellation."
    
    return f"Found {len(results)} total results"

@app.tool()
async def download_file(url: str, ctx) -> str:
    """Download with cancellation support."""
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            total_size = int(response.headers.get('content-length', 0))
            downloaded = 0
            chunks = []
            
            async for chunk in response.content.iter_chunked(8192):
                if ctx.is_cancelled:
                    return f"Download cancelled at {downloaded}/{total_size} bytes"
                
                chunks.append(chunk)
                downloaded += len(chunk)
            
            return f"Downloaded {downloaded} bytes"

İptal Bağlamının Uygulanması

class CancellableContext:
    """Context object that tracks cancellation state."""
    
    def __init__(self, request_id: str):
        self.request_id = request_id
        self._cancelled = asyncio.Event()
        self._cancel_reason = None
    
    @property
    def is_cancelled(self) -> bool:
        return self._cancelled.is_set()
    
    def cancel(self, reason: str = "Cancelled"):
        self._cancel_reason = reason
        self._cancelled.set()
    
    async def check_cancelled(self):
        """Raise if cancelled, otherwise continue."""
        if self.is_cancelled:
            raise CancelledError(self._cancel_reason)
    
    async def sleep_or_cancel(self, seconds: float):
        """Sleep that can be interrupted by cancellation."""
        try:
            await asyncio.wait_for(
                self._cancelled.wait(),
                timeout=seconds
            )
            raise CancelledError(self._cancel_reason)
        except asyncio.TimeoutError:
            pass  # Normal zaman aşımı, devam et

İstemci Tarafı İptal

import asyncio

async def search_with_timeout(session, query, timeout=30):
    """Search with automatic cancellation on timeout."""
    
    task = asyncio.create_task(
        session.call_tool("long_running_search", {"query": query})
    )
    
    try:
        result = await asyncio.wait_for(task, timeout=timeout)
        return result
    except asyncio.TimeoutError:
        # İptal talebi
        await session.send_notification({
            "method": "notifications/cancelled",
            "params": {"requestId": task.request_id, "reason": "Timeout"}
        })
        return "Search timed out"

3. Kaynak Şablonları

Kaynak şablonları, API ve veritabanları için yararlı parametrelerle dinamik URI yapımına izin verir.

Şablonların Tanımlanması

from mcp.server import Server
from mcp.types import ResourceTemplate

app = Server("template-server")

@app.list_resource_templates()
async def list_templates() -> list[ResourceTemplate]:
    """Return available resource templates."""
    return [
        ResourceTemplate(
            uriTemplate="db://users/{user_id}",
            name="User Profile",
            description="Fetch user profile by ID",
            mimeType="application/json"
        ),
        ResourceTemplate(
            uriTemplate="api://weather/{city}/{date}",
            name="Weather Data",
            description="Historical weather for city and date",
            mimeType="application/json"
        ),
        ResourceTemplate(
            uriTemplate="file://{path}",
            name="File Content",
            description="Read file at given path",
            mimeType="text/plain"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """Read resource, expanding template parameters."""
    
    # Parametreleri çıkarmak için URI'yi ayrıştır
    if uri.startswith("db://users/"):
        user_id = uri.split("/")[-1]
        return await fetch_user(user_id)
    
    elif uri.startswith("api://weather/"):
        parts = uri.replace("api://weather/", "").split("/")
        city, date = parts[0], parts[1]
        return await fetch_weather(city, date)
    
    elif uri.startswith("file://"):
        path = uri.replace("file://", "")
        return await read_file(path)
    
    raise ValueError(f"Unknown resource URI: {uri}")

TypeScript Uygulaması

server.setRequestHandler(ListResourceTemplatesSchema, async () => {
  return {
    resourceTemplates: [
      {
        uriTemplate: "github://repos/{owner}/{repo}/issues/{issue_number}",
        name: "GitHub Issue",
        description: "Fetch a specific GitHub issue",
        mimeType: "application/json"
      },
      {
        uriTemplate: "db://tables/{table}/rows/{id}",
        name: "Database Row",
        description: "Fetch a row from a database table",
        mimeType: "application/json"
      }
    ]
  };
});

server.setRequestHandler(ReadResourceSchema, async (request) => {
  const uri = request.params.uri;
  
  // GitHub sorun URI'sini ayrıştırın
  const githubMatch = uri.match(/^github:\/\/repos\/([^/]+)\/([^/]+)\/issues\/(\d+)$/);
  if (githubMatch) {
    const [_, owner, repo, issueNumber] = githubMatch;
    const issue = await fetchGitHubIssue(owner, repo, parseInt(issueNumber));
    return {
      contents: [{
        uri,
        mimeType: "application/json",
        text: JSON.stringify(issue, null, 2)
      }]
    };
  }
  
  throw new Error(`Unknown resource URI: ${uri}`);
});

4. Sunucu Yaşam Döngüsü Olayları

Doğru başlatma ve kapatma işlemleri, kaynakların düzgün yönetilmesini sağlar.

Python Yaşam Döngüsü Yönetimi

from mcp.server import Server
from contextlib import asynccontextmanager

app = Server("lifecycle-server")

# Paylaşılan durum
db_connection = None
cache = None

@asynccontextmanager
async def lifespan(server: Server):
    """Manage server lifecycle."""
    global db_connection, cache
    
    # Başlangıç
    print("🚀 Server starting...")
    db_connection = await create_database_connection()
    cache = await create_cache_client()
    print("✅ Resources initialized")
    
    yield  # Sunucu burada çalışır
    
    # Kapanış
    print("🛑 Server shutting down...")
    await db_connection.close()
    await cache.close()
    print("✅ Resources cleaned up")

app = Server("lifecycle-server", lifespan=lifespan)

@app.tool()
async def query_database(sql: str) -> str:
    """Use the shared database connection."""
    result = await db_connection.execute(sql)
    return str(result)

TypeScript Yaşam Döngüsü

import { Server } from "@modelcontextprotocol/sdk/server/index.js";

class ManagedServer {
  private server: Server;
  private dbConnection: DatabaseConnection | null = null;
  
  constructor() {
    this.server = new Server({
      name: "lifecycle-server",
      version: "1.0.0"
    });
    
    this.setupHandlers();
  }
  
  async start() {
    // Kaynakları başlat
    console.log("🚀 Server starting...");
    this.dbConnection = await createDatabaseConnection();
    console.log("✅ Database connected");
    
    // Sunucuyu başlat
    await this.server.connect(transport);
  }
  
  async stop() {
    // Kaynakları temizle
    console.log("🛑 Server shutting down...");
    if (this.dbConnection) {
      await this.dbConnection.close();
    }
    await this.server.close();
    console.log("✅ Cleanup complete");
  }
  
  private setupHandlers() {
    this.server.setRequestHandler(CallToolSchema, async (request) => {
      // this.dbConnection öğesini güvenli kullan
      // ...
    });
  }
}

// Nazik kapatma ile kullanım
const server = new ManagedServer();

process.on('SIGINT', async () => {
  await server.stop();
  process.exit(0);
});

await server.start();

5. Günlük Kaydı Kontrolü

MCP, istemcilerin kontrol edebileceği sunucu tarafı günlük kayıt düzeylerini destekler.

Günlük Düzeylerinin Uygulanması

from mcp.server import Server
from mcp.types import LoggingLevel
import logging

app = Server("logging-server")

# MCP seviyelerini Python logging seviyelerine eşleştir
LEVEL_MAP = {
    LoggingLevel.DEBUG: logging.DEBUG,
    LoggingLevel.INFO: logging.INFO,
    LoggingLevel.WARNING: logging.WARNING,
    LoggingLevel.ERROR: logging.ERROR,
}

logger = logging.getLogger("mcp-server")

@app.set_logging_level()
async def set_logging_level(level: LoggingLevel) -> None:
    """Handle client request to change logging level."""
    python_level = LEVEL_MAP.get(level, logging.INFO)
    logger.setLevel(python_level)
    logger.info(f"Logging level set to {level}")

@app.tool()
async def debug_operation(data: str) -> str:
    """Tool with various logging levels."""
    logger.debug(f"Processing data: {data}")
    
    try:
        result = process(data)
        logger.info(f"Successfully processed: {result}")
        return result
    except Exception as e:
        logger.error(f"Processing failed: {e}")
        raise

İstemciye Günlük Mesajları Gönderme

@app.tool()
async def complex_operation(input: str, ctx) -> str:
    """Operation that logs to client."""
    
    # Günlük bildirimini istemciye gönder
    await ctx.send_log(
        level="info",
        message=f"Starting complex operation with input: {input}"
    )
    
    # İş yap...
    result = await do_work(input)
    
    await ctx.send_log(
        level="debug",
        message=f"Operation complete, result size: {len(result)}"
    )
    
    return result

6. Hata İşleme Kalıpları

Tutarlı hata işleme, hata ayıklamayı ve kullanıcı deneyimini iyileştirir.

MCP Hata Kodları

from mcp.types import McpError, ErrorCode

class ToolError(McpError):
    """Base class for tool errors."""
    pass

class ValidationError(ToolError):
    """Invalid input parameters."""
    def __init__(self, message: str):
        super().__init__(ErrorCode.INVALID_PARAMS, message)

class NotFoundError(ToolError):
    """Requested resource not found."""
    def __init__(self, resource: str):
        super().__init__(ErrorCode.INVALID_REQUEST, f"Not found: {resource}")

class PermissionError(ToolError):
    """Access denied."""
    def __init__(self, action: str):
        super().__init__(ErrorCode.INVALID_REQUEST, f"Permission denied: {action}")

class InternalError(ToolError):
    """Internal server error."""
    def __init__(self, message: str):
        super().__init__(ErrorCode.INTERNAL_ERROR, message)

Yapılandırılmış Hata Yanıtları

@app.tool()
async def safe_operation(input: str) -> str:
    """Tool with comprehensive error handling."""
    
    # Girişi doğrula
    if not input:
        raise ValidationError("Input cannot be empty")
    
    if len(input) > 10000:
        raise ValidationError(f"Input too large: {len(input)} chars (max 10000)")
    
    try:
        # İzinleri kontrol et
        if not await check_permission(input):
            raise PermissionError(f"read {input}")
        
        # İşlemi gerçekleştir
        result = await perform_operation(input)
        
        if result is None:
            raise NotFoundError(input)
        
        return result
        
    except ConnectionError as e:
        raise InternalError(f"Database connection failed: {e}")
    except TimeoutError as e:
        raise InternalError(f"Operation timed out: {e}")
    except Exception as e:
        # Beklenmeyen hataları kaydet
        logger.exception(f"Unexpected error in safe_operation")
        raise InternalError(f"Unexpected error: {type(e).__name__}")

TypeScript'te Hata İşleme

import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";

function validateInput(data: unknown): asserts data is ValidInput {
  if (typeof data !== "object" || data === null) {
    throw new McpError(
      ErrorCode.InvalidParams,
      "Input must be an object"
    );
  }
  // Daha fazla doğrulama...
}

server.setRequestHandler(CallToolSchema, async (request) => {
  try {
    validateInput(request.params.arguments);
    
    const result = await performOperation(request.params.arguments);
    
    return {
      content: [{ type: "text", text: JSON.stringify(result) }]
    };
    
  } catch (error) {
    if (error instanceof McpError) {
      throw error;  // Zaten bir MCP hatası
    }
    
    // Diğer hataları dönüştür
    if (error instanceof NotFoundError) {
      throw new McpError(ErrorCode.InvalidRequest, error.message);
    }
    
    // Bilinmeyen hata
    console.error("Unexpected error:", error);
    throw new McpError(
      ErrorCode.InternalError,
      "An unexpected error occurred"
    );
  }
});

Deneysel Özellikler (MCP 2025-11-25)

Bu özellikler spesifikasyonda deneysel olarak işaretlenmiştir:

Görevler (Uzun Süreçli İşlemler)

# Görevler, durumu ile uzun süre çalışan işlemleri takip etmeyi sağlar
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
    """Long-running ML training task."""
    
    # Görev başlatıldı bildirimi
    await ctx.report_status("running", "Initializing training...")
    
    # Eğitim döngüsü
    for epoch in range(100):
        await train_epoch(model_id, data_path, epoch)
        await ctx.report_status(
            "running",
            f"Training epoch {epoch + 1}/100",
            progress=epoch + 1,
            total=100
        )
    
    await ctx.report_status("completed", "Training finished")
    return f"Model {model_id} trained successfully"

Araç Açıklamaları

# Açıklamalar, araç davranışı hakkında meta veri sağlar
@app.tool(
    annotations={
        "destructive": False,      # Veriyi değiştirmez
        "idempotent": True,        # Tekrar denemek güvenlidir
        "timeout_seconds": 30,     # Beklenen maksimum süre
        "requires_approval": False # Kullanıcı onayı gerekmez
    }
)
async def safe_query(query: str) -> str:
    """A read-only database query tool."""
    return await execute_read_query(query)

Sonraki Adımlar

Ek Kaynaklar

Feragatname:
Bu belge, AI çeviri servisi Co-op Translator kullanılarak çevrilmiştir. Doğruluk için çaba gösterilse de, otomatik çevirilerin hata veya yanlışlık içerebileceğini lütfen unutmayınız. Orijinal belge, kendi ana dilinde yetkili kaynak olarak kabul edilmelidir. Kritik bilgiler için profesyonel insan çevirisi önerilmektedir. Bu çevirinin kullanılmasıyla ortaya çıkabilecek herhangi bir yanlış anlama veya yanlış yorumdan dolayı sorumluluk kabul edilmemektedir.

20 KiB Raw Permalink Blame History Unescape Escape

MCP Protokol Özellikleri Derinlemesine İnceleme

Kapsanan Özellikler

1. İlerleme Bildirimleri

Nasıl Çalışır

Python Uygulaması

TypeScript Uygulaması

İstemci İşleme (Python)

2. İstek İptali

Python Uygulaması

İptal Bağlamının Uygulanması

İstemci Tarafı İptal

3. Kaynak Şablonları

Şablonların Tanımlanması

TypeScript Uygulaması

4. Sunucu Yaşam Döngüsü Olayları

Python Yaşam Döngüsü Yönetimi

TypeScript Yaşam Döngüsü

5. Günlük Kaydı Kontrolü

Günlük Düzeylerinin Uygulanması

İstemciye Günlük Mesajları Gönderme

6. Hata İşleme Kalıpları

MCP Hata Kodları

Yapılandırılmış Hata Yanıtları

TypeScript'te Hata İşleme

Deneysel Özellikler (MCP 2025-11-25)

Görevler (Uzun Süreçli İşlemler)

Araç Açıklamaları

Sonraki Adımlar

Ek Kaynaklar

20 KiB

Raw Permalink Blame History