wehub-resource-sync
20 KiB
20 KiB
Poglobljen pogled na značilnosti protokola MCP
Ta vodič raziskuje napredne značilnosti protokola MCP, ki presegajo osnovno upravljanje orodij in virov. Razumevanje teh značilnosti vam pomaga zgraditi bolj robustne, prijazne do uporabnika in produkcijsko pripravljene strežnike MCP.
Pregled značilnosti
- Obvestila o napredku – Poročanje o napredku za dolgotrajne operacije
- Preklic zahtev – Dovoljevanje strankam preklic v teku zahtev
- Predloge virov – Dinamični URI-ji virov s parametri
- Dogodki življenjskega cikla strežnika – Pravilna inicializacija in zaustavitev
- Nadzor beleženja – Konfiguracija beleženja na strežniku
- Vzorec ravnanja z napakami – Dosledni odgovori na napake
1. Obvestila o napredku
Za operacije, ki trajajo dlje (obdelava podatkov, prenos datotek, klici API), obvestila o napredku uporabnike obveščajo o stanju.
Kako deluje
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (dolga operacija)
Server-->>Client: obvestilo: napredek 10%
Server-->>Client: obvestilo: napredek 50%
Server-->>Client: obvestilo: napredek 90%
Server->>Client: rezultat (zaključeno)
Implementacija v Pythonu
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."""
# Pridobi velikost datoteke za izračun napredka
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Obdelaj del
await process_chunk(chunk)
processed += len(chunk)
# Pošlji obvestilo o napredku
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)
# Poročaj o napredku po vsakem elementu
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"
Implementacija v TypeScript-u
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);
// Pošlji obvestilo o napredku
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) }] };
}
});
Ravnanje na strani odjemalca (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Registriraj upravljavca
session.on_notification("notifications/progress", handle_progress)
# Pokliči orodje (posodobitve napredka bodo prispele prek upravljavca)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Preklic zahtev
Dovoljevanje strankam, da prekličejo zahteve, ki niso več potrebne ali trajajo predolgo.
Implementacija v Pythonu
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): # Iskanje skozi veliko strani
# Preveri, če je bila zahtevana preklica
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Simuliraj iskanje po straneh
page_results = await search_page(query, page)
results.extend(page_results)
# Kratek zamik omogoča preverjanje preklica
await asyncio.sleep(0.1)
except CancelledError:
# Vrni delne rezultate
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"
Implementacija konteksta preklica
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 # Normalni časovni potek, nadaljuj
Preklic na strani odjemalca
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:
# Zahteva za preklic
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Predloge virov
Predloge virov omogočajo dinamično sestavo URI-jev s parametri, kar je uporabno za API-je in baze podatkov.
Določanje predlog
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."""
# Razčleni URI, da pridobiš parametre
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}")
Implementacija v TypeScript-u
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;
// Razčleni URI za GitHub zadevo
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. Dogodki življenjskega cikla strežnika
Pravilno upravljanje inicializacije in zaustavitve zagotavlja čisto upravljanje virov.
Upravljanje življenjskega cikla v Pythonu
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Deljeno stanje
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Zagon
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # Strežnik teče tukaj
# Zaustavitev
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)
Življenjski cikel v TypeScript-u
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() {
// Inicializiraj vire
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Začni strežnik
await this.server.connect(transport);
}
async stop() {
// Očisti vire
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) => {
// Varno uporabi this.dbConnection
// ...
});
}
}
// Uporaba z nežno zaustavitvijo
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Nadzor beleženja
MCP podpira ravni beleženja na strani strežnika, ki jih lahko nadzorujejo odjemalci.
Implementacija ravni beleženja
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Preslikajte raven MCP na ravni dnevnika v Pythonu
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
Pošiljanje zapisov odjemalcu
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Pošlji beležko o dogodku odjemalcu
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Opravi delo...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Vzorec ravnanja z napakami
Dosledno ravnanje z napakami izboljšuje odpravljanje napak in uporabniško izkušnjo.
Kode napak 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)
Strukturirani odgovori na napake
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Preveri vhodne podatke
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:
# Preveri dovoljenja
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Izvedi operacijo
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:
# Zabeleži nepričakovane napake
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Ravnanje z napakami v TypeScript-u
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"
);
}
// Več preverjanja ...
}
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; // Že MCP napaka
}
// Pretvori druge napake
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Neznana napaka
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Eksperimentalne značilnosti (MCP 2025-11-25)
Te značilnosti so v specifikaciji označene kot eksperimentalne:
Naloge (dolgotrajne operacije)
# Naloge omogočajo spremljanje dolgo trajajočih operacij s stanjem
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Poročilo o začetku naloge
await ctx.report_status("running", "Initializing training...")
# Zanka učenja
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"
Oznake orodij
# Oznake zagotavljajo metapodatke o vedenju orodja
@app.tool(
annotations={
"destructive": False, # Ne spreminja podatkov
"idempotent": True, # Varno za ponovni poskus
"timeout_seconds": 30, # Pričakovani največji čas trajanja
"requires_approval": False # Ni potrebnega uporabniškega odobritve
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
Kaj sledi
Dodatni viri
Opozorilo: Ta dokument je bil preveden z uporabo storitve za avtomatski prevod AI Co-op Translator. Čeprav si prizadevamo za natančnost, upoštevajte, da lahko avtomatizirani prevodi vsebujejo napake ali netočnosti. Izvirni dokument v njegovem prvotnem jeziku velja za uradni vir. Za pomembne informacije priporočamo strokovni človeški prevod. Ne odgovarjamo za morebitne nesporazume ali napačne interpretacije, ki izhajajo iz uporabe tega prevoda.