wehub-resource-sync
20 KiB
20 KiB
Mendalam Fitur Protokol MCP
Panduan ini mengeksplorasi fitur protokol MCP tingkat lanjut yang melampaui penanganan alat dan sumber daya dasar. Memahami fitur-fitur ini membantu Anda membangun server MCP yang lebih kuat, ramah pengguna, dan siap produksi.
Fitur yang Dibahas
- Notifikasi Progres - Melaporkan progres untuk operasi yang berjalan lama
- Pembatalan Permintaan - Memungkinkan klien membatalkan permintaan yang sedang berlangsung
- Template Sumber Daya - URI sumber daya dinamis dengan parameter
- Peristiwa Siklus Hidup Server - Inisialisasi dan penghentian yang tepat
- Kontrol Logging - Konfigurasi pencatatan sisi server
- Polapola Penanganan Kesalahan - Respon kesalahan yang konsisten
1. Notifikasi Progres
Untuk operasi yang memakan waktu (pemrosesan data, unduhan file, panggilan API), notifikasi progres menjaga pengguna tetap mendapat informasi.
Cara Kerjanya
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (operasi panjang)
Server-->>Client: notifikasi: kemajuan 10%
Server-->>Client: notifikasi: kemajuan 50%
Server-->>Client: notifikasi: kemajuan 90%
Server->>Client: hasil (selesai)
Implementasi Python
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."""
# Dapatkan ukuran file untuk perhitungan kemajuan
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Proses potongan
await process_chunk(chunk)
processed += len(chunk)
# Kirim notifikasi kemajuan
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)
# Laporkan kemajuan setelah setiap item
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"
Implementasi TypeScript
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);
// Kirim notifikasi kemajuan
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) }] };
}
});
Penanganan di Klien (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Daftarkan penangan
session.on_notification("notifications/progress", handle_progress)
# Panggil alat (pembaruan kemajuan akan tiba melalui penangan)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Pembatalan Permintaan
Memungkinkan klien membatalkan permintaan yang tidak lagi diperlukan atau memakan waktu terlalu lama.
Implementasi Python
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): # Cari melalui banyak halaman
# Periksa apakah pembatalan diminta
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Simulasikan pencarian halaman
page_results = await search_page(query, page)
results.extend(page_results)
# Penundaan kecil memungkinkan pemeriksaan pembatalan
await asyncio.sleep(0.1)
except CancelledError:
# Kembalikan hasil sebagian
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"
Mengimplementasikan Konteks Pembatalan
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 # Waktu habis normal, lanjutkan
Pembatalan di Sisi Klien
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:
# Permintaan pembatalan
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Template Sumber Daya
Template sumber daya memungkinkan pembuatan URI dinamis dengan parameter, berguna untuk API dan basis data.
Mendefinisikan Template
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."""
# Mengurai URI untuk mengambil parameter
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}")
Implementasi TypeScript
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;
// Mengurai URI masalah GitHub
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. Peristiwa Siklus Hidup Server
Penanganan inisialisasi dan penghentian yang tepat memastikan pengelolaan sumber daya yang bersih.
Manajemen Siklus Hidup Python
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Status bersama
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Memulai
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # Server berjalan di sini
# Mematikan
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)
Siklus Hidup TypeScript
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() {
// Inisialisasi sumber daya
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Mulai server
await this.server.connect(transport);
}
async stop() {
// Bersihkan sumber daya
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) => {
// Gunakan this.dbConnection dengan aman
// ...
});
}
}
// Penggunaan dengan penghentian secara lembut
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Kontrol Logging
MCP mendukung level logging sisi server yang dapat dikontrol oleh klien.
Mengimplementasikan Level Logging
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Pemetaan level MCP ke level logging Python
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
Mengirim Pesan Log ke Klien
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Kirim notifikasi log ke klien
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Melakukan pekerjaan...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Polapola Penanganan Kesalahan
Penanganan kesalahan yang konsisten meningkatkan debugging dan pengalaman pengguna.
Kode Kesalahan MCP
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)
Respon Kesalahan Terstruktur
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Validasi masukan
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:
# Periksa izin
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Lakukan operasi
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:
# Catat kesalahan tak terduga
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Penanganan Kesalahan di TypeScript
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"
);
}
// Validasi lebih lanjut...
}
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; // Sudah merupakan kesalahan MCP
}
// Konversi kesalahan lain
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Kesalahan tidak diketahui
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Fitur Eksperimental (MCP 2025-11-25)
Fitur-fitur ini ditandai sebagai eksperimental dalam spesifikasi:
Tugas (Operasi Berjalan Lama)
# Tugas memungkinkan pelacakan operasi jangka panjang dengan status
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Laporkan tugas dimulai
await ctx.report_status("running", "Initializing training...")
# Loop pelatihan
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"
Anotasi Alat
# Anotasi memberikan metadata tentang perilaku alat
@app.tool(
annotations={
"destructive": False, # Tidak memodifikasi data
"idempotent": True, # Aman untuk mencoba ulang
"timeout_seconds": 30, # Durasi maksimum yang diharapkan
"requires_approval": False # Tidak perlu persetujuan pengguna
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
Selanjutnya
Sumber Daya Tambahan
Penafian:
Dokumen ini telah diterjemahkan menggunakan layanan terjemahan AI Co-op Translator. Meskipun kami berusaha untuk memberikan terjemahan yang akurat, harap diingat bahwa terjemahan otomatis mungkin mengandung kesalahan atau ketidakakuratan. Dokumen asli dalam bahasa aslinya harus dianggap sebagai sumber yang sah dan terpercaya. Untuk informasi penting, disarankan menggunakan jasa penerjemah manusia profesional. Kami tidak bertanggung jawab atas kesalahpahaman atau tafsir yang keliru yang timbul dari penggunaan terjemahan ini.