wehub-resource-sync
20 KiB
20 KiB
MCP Protocol Features Deep Dive
Mwongozo huu unachunguza vipengele vya hali ya juu vya itifaki ya MCP vinavyozidi usindikaji wa zana na rasilimali za msingi. Kuelewa vipengele hivi kunakusaidia kujenga seva za MCP zenye nguvu zaidi, rahisi kwa watumiaji, na tayari kwa uzalishaji.
Features Covered
- Progress Notifications - Ripoti maendeleo kwa shughuli zinazochukua muda mrefu
- Request Cancellation - Ruhusu wateja kughairi maombi yanayoendelea
- Resource Templates - URI za rasilimali zinazobadilika na vigezo
- Server Lifecycle Events - Uanzishaji na kufunga kwa usahihi
- Logging Control - Usanidi wa kumbukumbu upande wa seva
- Error Handling Patterns - Majibu ya makosa kwa uwiano
1. Progress Notifications
Kwa shughuli zinazochukua muda (usindikaji data, kupakua faili, miito ya API), taarifa za maendeleo hufanya watumiaji wawe na taarifa.
How It Works
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (kazi ndefu)
Server-->>Client: taarifa: maendeleo 10%
Server-->>Client: taarifa: maendeleo 50%
Server-->>Client: taarifa: maendeleo 90%
Server->>Client: matokeo (imekamilika)
Python Implementation
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."""
# Pata ukubwa wa faili kwa ajili ya hesabu ya maendeleo
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Fanyia kipande kazi
await process_chunk(chunk)
processed += len(chunk)
# Tuma taarifa ya maendeleo
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)
# Ripoti maendeleo baada ya kila kipengee
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 Implementation
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);
// Tuma taarifa ya maendeleo
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) }] };
}
});
Client Handling (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Sajili mshughuliki
session.on_notification("notifications/progress", handle_progress)
# Piga simu zana (mabadiliko ya maendeleo yatawasilishwa kupitia mshughuliki)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Request Cancellation
Ruhusu wateja kughairi maombi ambayo hayahitajiki tena au yanachukua muda mrefu sana.
Python Implementation
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): # Tafuta kupitia kurasa nyingi
# Angalia kama kuachishwa kulihitajika
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Tafakari utafutaji wa ukurasa
page_results = await search_page(query, page)
results.extend(page_results)
# Muda mfupi unaruhusu ukaguzi wa kuachishwa
await asyncio.sleep(0.1)
except CancelledError:
# Rudisha matokeo ya sehemu
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"
Implementing Cancellation Context
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 # Muda wa kawaida umeisha, endelea
Client-Side Cancellation
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:
# Ombi la kughairi
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Resource Templates
Templeti za rasilimali huruhusu ujenzi wa URI zinazobadilika kwa vigezo, muhimu kwa API na hifadhidata.
Defining Templates
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."""
# Changanua URI ili kutoa parameta
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 Implementation
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;
// Tumia URI ya tatizo la 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. Server Lifecycle Events
Uanzishaji na kufunga kwa usahihi husaidia usimamizi safi wa rasilimali.
Python Lifecycle Management
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Hali iliyoshirikiwa
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Mwanzo
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # Seva inaendesha hapa
# Kuzimwa
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 Lifecycle
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() {
// Anzisha rasilimali
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Anzisha seva
await this.server.connect(transport);
}
async stop() {
// Safisha rasilimali
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) => {
// Tumia this.dbConnection kwa usalama
// ...
});
}
}
// Matumizi na kufunga kwa heshima
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Logging Control
MCP inaunga mkono viwango vya kumbukumbu upande wa seva ambavyo wateja wanaweza kudhibiti.
Implementing Logging Levels
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Fananisha viwango vya MCP na viwango vya kurekodi (logging) vya 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
Sending Log Messages to Client
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Tuma taarifa ya kumbukumbu kwa mteja
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Fanya kazi...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Error Handling Patterns
Usimamizi thabiti wa makosa hubaoresha utambuzi wa hitilafu na uzoefu wa mtumiaji.
MCP Error Codes
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)
Structured Error Responses
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Thibitisha ingizo
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:
# Angalia ruhusa
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Fanya operesheni
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:
# Rekodi makosa yasiyotarajiwa
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Error Handling in 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"
);
}
// Uthibitishaji zaidi...
}
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; // Tayari ni kosa la MCP
}
// Badilisha makosa mengine
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Kosa lisilojulikana
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Experimental Features (MCP 2025-11-25)
Vipengele hivi vimewekwa alama kama vya majaribio kwenye specs:
Tasks (Long-Running Operations)
# Majukumu huruhusu kufuatilia shughuli za muda mrefu zenye hali
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Ripoti kazi iliyoanza
await ctx.report_status("running", "Initializing training...")
# Mzunguko wa mafunzo
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"
Tool Annotations
# Maelezo ya ziada hutoa metadata kuhusu tabia ya chombo
@app.tool(
annotations={
"destructive": False, # Haiathiri data
"idempotent": True, # Salama kujaribu tena
"timeout_seconds": 30, # Muda wa juu unaotarajiwa
"requires_approval": False # Hakuna idhini ya mtumiaji inahitajika
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
What's Next
Additional Resources
Angalizo: Nyaraka hii imetafsiriwa kwa kutumia huduma ya tafsiri ya AI Co-op Translator. Ingawa tunajitahidi kwa usahihi, tafadhali fahamu kwamba tafsiri za kiotomatiki zinaweza kuwa na makosa au upungufu wa usahihi. Nyaraka ya awali katika lugha yake ya asili inapaswa kuzingatiwa kama chanzo chenye dhamana. Kwa taarifa muhimu, tafsiri ya mtaalamu wa binadamu inashauriwa. Hatutojitahadharisha kwa kutoelewana au tafsiri zisizo sahihi zinazotokana na matumizi ya tafsiri hii.