wehub-resource-sync
20 KiB
20 KiB
MCP Protokol Funktioner Dybt Dyk
Denne guide udforsker avancerede MCP protokol funktioner, der går ud over grundlæggende håndtering af værktøjer og ressourcer. Forståelse af disse funktioner hjælper dig med at bygge mere robuste, brugervenlige og produktionsklare MCP servere.
Dækkede Funktioner
- Fremdriftsmeddelelser - Rapportér fremdrift for langvarige operationer
- Annullering af Forespørgsler - Lad klienter annullere igangværende forespørgsler
- Ressource Skabeloner - Dynamiske ressource-URI'er med parametre
- Server Livscyklus Begivenheder - Korrekt initialisering og nedlukning
- Logningskontrol - Server-side logningskonfiguration
- Fejlhåndteringsmønstre - Konsistente fejlbeskeder
1. Fremdriftsmeddelelser
For operationer, der tager tid (databehandling, filoverførsler, API-opkald), holder fremdriftsmeddelelser brugerne informerede.
Sådan Fungerer Det
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (lang operation)
Server-->>Client: notifikation: fremgang 10%
Server-->>Client: notifikation: fremgang 50%
Server-->>Client: notifikation: fremgang 90%
Server->>Client: resultat (fuldført)
Python Implementering
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."""
# Hent filstørrelse til fremskridtsberegning
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Behandl stykke
await process_chunk(chunk)
processed += len(chunk)
# Send fremskridtsmeddelelse
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)
# Rapportér fremskridt efter hver genstand
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 Implementering
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);
// Send fremdriftsmeddelelse
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) }] };
}
});
Klient Håndtering (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Registrer handler
session.on_notification("notifications/progress", handle_progress)
# Kald værktøj (fremskridtsopdateringer vil ankomme via handler)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Annullering af Forespørgsler
Lad klienter annullere forespørgsler, der ikke længere er nødvendige eller tager for lang tid.
Python Implementering
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): # Søg gennem mange sider
# Tjek om annullering blev anmodet
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Simuler sidesøgning
page_results = await search_page(query, page)
results.extend(page_results)
# Lille forsinkelse tillader annulleringskontroller
await asyncio.sleep(0.1)
except CancelledError:
# Returner delvise resultater
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"
Implementering af Annulleringskontekst
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 timeout, fortsæt
Klient-Side Annullering
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:
# Anmod om annullering
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Ressource Skabeloner
Ressource skabeloner tillader dynamisk opbygning af URI'er med parametre, nyttigt for API'er og databaser.
Definere Skabeloner
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."""
# Parse URI'en for at udtrække 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}")
TypeScript Implementering
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;
// Parse GitHub problem URI
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. Server Livscyklus Begivenheder
Korrekt håndtering af initialisering og nedlukning sikrer ren ressourcehåndtering.
Python Livscyklus Håndtering
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Delt tilstand
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Opstart
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # Server kører her
# Nedlukning
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 Livscyklus
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() {
// Initialiser ressourcer
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Start server
await this.server.connect(transport);
}
async stop() {
// Ryd op i ressourcer
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) => {
// Brug this.dbConnection sikkert
// ...
});
}
}
// Brug med ordentlig nedlukning
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Logningskontrol
MCP understøtter server-side logningsniveauer, som klienter kan styre.
Implementering af Logningsniveauer
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Kortlæg MCP-niveauer til Python logging-niveauer
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
Afsendelse af Logbeskeder til Klient
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Send lognotifikation til klient
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Udfør arbejde...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Fejlhåndteringsmønstre
Konsistent fejlhåndtering forbedrer fejlfinding og brugeroplevelse.
MCP Fejlkoder
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)
Strukturerede Fejlsvar
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Valider input
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:
# Tjek tilladelser
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Udfør operation
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:
# Log uventede fejl
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Fejlhåndtering i 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"
);
}
// Mere validering...
}
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; // Allerede en MCP-fejl
}
// Konverter andre fejl
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Ukendt fejl
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Eksperimentelle Funktioner (MCP 2025-11-25)
Disse funktioner er markeret som eksperimentelle i specifikationen:
Opgaver (Langvarige Operationer)
# Opgaver tillader sporing af langvarige operationer med tilstand
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Rapporter opgave startet
await ctx.report_status("running", "Initializing training...")
# Træningsloop
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"
Værktøjsannotationer
# Annotationer giver metadata om værktøjets opførsel
@app.tool(
annotations={
"destructive": False, # Ændrer ikke data
"idempotent": True, # Sikkert at prøve igen
"timeout_seconds": 30, # Forventet maksimal varighed
"requires_approval": False # Ingen bruger-godkendelse nødvendig
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
Hvad Kommer Nu
Yderligere Ressourcer
Ansvarsfraskrivelse: Dette dokument er oversat ved hjælp af AI-oversættelsestjenesten Co-op Translator. Selvom vi bestræber os på nøjagtighed, skal du være opmærksom på, at automatiske oversættelser kan indeholde fejl eller unøjagtigheder. Det oprindelige dokument på dets modersmål skal betragtes som den autoritative kilde. For vigtig information anbefales professionel menneskelig oversættelse. Vi påtager os intet ansvar for misforståelser eller fejltolkninger, der opstår som følge af brugen af denne oversættelse.