wehub-resource-sync
21 KiB
21 KiB
Explorare Detaliată a Funcțiilor Protocolului MCP
Acest ghid explorează funcții avansate ale protocolului MCP care depășesc gestionarea de bază a uneltelor și resurselor. Înțelegerea acestor funcții te ajută să construiești servere MCP mai robuste, ușor de utilizat și gata pentru producție.
Funcții Acoperite
- Notificări de Progres - Raportarea progresului pentru operațiuni de durată lungă
- Anularea Cererilor - Permite clienților să anuleze cererile în curs
- Șabloane de Resurse - URI-uri dinamice de resurse cu parametri
- Evenimente în Ciclu de Viață al Serverului - Inițializare și închidere corectă
- Controlul Logării - Configurarea logării pe partea de server
- Modele de Gestionare a Erorilor - Răspunsuri consistente la erori
1. Notificări de Progres
Pentru operațiuni care necesită timp (procesarea datelor, descărcări de fișiere, apeluri API), notificările de progres țin utilizatorii informați.
Cum Funcționează
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (operație îndelungată)
Server-->>Client: notificare: progres 10%
Server-->>Client: notificare: progres 50%
Server-->>Client: notificare: progres 90%
Server->>Client: rezultat (finalizat)
Implementare 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."""
# Obține dimensiunea fișierului pentru calculul progresului
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Procesează fragmentul
await process_chunk(chunk)
processed += len(chunk)
# Trimite notificare de progres
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)
# Raportează progresul după fiecare element
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"
Implementare 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);
// Trimite notificare de progres
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) }] };
}
});
Gestionarea pe Client (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Înregistrează handler
session.on_notification("notifications/progress", handle_progress)
# Apelează instrumentul (actualizările de progres vor fi primite prin handler)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Anularea Cererilor
Permite clienților să anuleze cererile care nu mai sunt necesare sau care durează prea mult.
Implementare 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): # Caută prin mai multe pagini
# Verifică dacă a fost solicitată anularea
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Simulează căutarea pe pagină
page_results = await search_page(query, page)
results.extend(page_results)
# O întârziere mică permite verificări de anulare
await asyncio.sleep(0.1)
except CancelledError:
# Returnează rezultate parțiale
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"
Implementarea Contextului de Anulare
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 # Timeout normal, continuă
Anularea pe Partea de Client
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:
# Cerere de anulare
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Șabloane de Resurse
Șabloanele de resurse permit construirea dinamică a URI-urilor cu parametri, utile pentru API-uri și baze de date.
Definirea Șabloanelor
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."""
# Analizează URI-ul pentru a extrage parametrii
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}")
Implementare 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;
// Analizează URI-ul problemei 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. Evenimente în Ciclu de Viață al Serverului
Gestionarea corectă a inițializării și închiderii asigură o administrare curată a resurselor.
Managementul Ciclu de Viață în Python
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Stare partajată
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Pornire
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # Serverul rulează aici
# Oprire
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)
Ciclu de Viață în 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() {
// Inițializează resursele
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Pornește serverul
await this.server.connect(transport);
}
async stop() {
// Curăță resursele
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) => {
// Folosește this.dbConnection în siguranță
// ...
});
}
}
// Utilizare cu oprire controlată
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Controlul Logării
MCP suportă niveluri de logare pe partea de server pe care clienții le pot controla.
Implementarea Nivelurilor de Logare
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Mapează nivelurile MCP la nivelurile de logare 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
Trimiterea Mesajelor de Log către Client
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Trimite notificare de jurnal către client
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Efectuează lucrarea...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Modele de Gestionare a Erorilor
Gestionarea consistentă a erorilor îmbunătățește depanarea și experiența utilizatorului.
Coduri de Eroare 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)
Răspunsuri Structurate la Erori
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Validează intrarea
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:
# Verifică permisiunile
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Efectuează operația
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:
# Înregistrează erorile neașteptate
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Gestionarea Erorilor în 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"
);
}
// Mai multe validări...
}
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; // Eroare MCP deja existentă
}
// Convertește alte erori
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Eroare necunoscută
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Funcții Experimentale (MCP 2025-11-25)
Aceste funcții sunt marcate ca experimentale în specificație:
Task-uri (Operațiuni de Durată Lungă)
# Sarcinile permit urmărirea operațiunilor de lungă durată cu stare
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Raportați începutul sarcinii
await ctx.report_status("running", "Initializing training...")
# Bucla de antrenament
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"
Anotări pentru Unelte
# Anotările oferă metadate despre comportamentul uneltei
@app.tool(
annotations={
"destructive": False, # Nu modifică datele
"idempotent": True, # Sigur de încercat din nou
"timeout_seconds": 30, # Durata maximă așteptată
"requires_approval": False # Nu este necesară aprobarea utilizatorului
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
Ce Urmează
Resurse Suplimentare
Declinare a responsabilității:
Acest document a fost tradus folosind serviciul de traducere AI Co-op Translator. Deși ne străduim pentru acuratețe, vă rugăm să rețineți că traducerile automate pot conține erori sau inexactități. Documentul original, în limba sa nativă, trebuie considerat sursa autorizată. Pentru informații critice, se recomandă traducerea profesională realizată de un specialist. Nu ne asumăm răspunderea pentru eventualele neînțelegeri sau interpretări greșite ce pot apărea în urma utilizării acestei traduceri.