wehub-resource-sync
37 KiB
37 KiB
Bästa metoder och optimering
🎯 Vad den här labben täcker
Den här examenslabben sammanfattar bästa metoder, optimeringstekniker och produktionsriktlinjer för att bygga robusta, skalbara och säkra MCP-servrar med databasintegration. Du kommer att lära dig från verklig erfarenhet och branschstandarder för att säkerställa att din implementation är redo för produktion.
Översikt
Att bygga en framgångsrik MCP-server handlar om mer än att få koden att fungera. Den här labben täcker de viktiga metoder som skiljer konceptbevis från produktionsklara system som kan skalas, prestera pålitligt och upprätthålla säkerhetsstandarder.
Dessa bästa metoder är härledda från verkliga implementationer, samhällsfeedback och erfarenheter från företagsimplementeringar.
Lärandemål
I slutet av den här labben kommer du kunna att:
- Tillämpa prestandaoptimeringstekniker för MCP-servrar och databaser
- Implementera omfattande säkerhetshärdningsåtgärder
- Designa skalbara arkitekturmönster för produktionsmiljöer
- Etablera övervaknings-, underhålls- och driftprocedurer
- Optimera kostnader samtidigt som prestanda och tillförlitlighet bibehålls
- Bidra till MCP-communityn och ekosystemet
🚀 Prestandaoptimering
Databasprestanda
Anslutningspooloptimering
# Optimerad konfigurationspool för anslutningar
POOL_CONFIG = {
# Storlekskonfiguration
"min_size": max(2, cpu_count()), # Minst 2, anpassa efter CPU
"max_size": min(20, cpu_count() * 4), # Begränsa till rimligt maxvärde
# Tidskonfiguration
"max_inactive_connection_lifetime": 300, # 5 minuter
"command_timeout": 30, # 30 sekunder
"max_queries": 50000, # Rotera anslutningar
# PostgreSQL-inställningar
"server_settings": {
"application_name": "mcp-server-prod",
"jit": "off", # Inaktivera för konsekvens
"work_mem": "8MB", # Optimera för frågor
"shared_preload_libraries": "pg_stat_statements",
"log_statement": "mod", # Logga endast ändringar
"log_min_duration_statement": "1s", # Logga långsamma frågor
}
}
Frågeoptimeringsmönster
class QueryOptimizer:
"""Database query optimization utilities."""
def __init__(self):
self.query_cache = {}
self.slow_query_threshold = 1.0 # sekunder
async def execute_optimized_query(
self,
query: str,
params: tuple = None,
cache_key: str = None,
cache_ttl: int = 300
):
"""Execute query with optimization and caching."""
# Kontrollera cache först
if cache_key and cache_key in self.query_cache:
cache_entry = self.query_cache[cache_key]
if time.time() - cache_entry['timestamp'] < cache_ttl:
return cache_entry['result']
# Kör med övervakning
start_time = time.time()
try:
async with db_provider.get_connection() as conn:
# Optimera frågeexekvering
await conn.execute("SET enable_seqscan = off") # Föredra index
await conn.execute("SET work_mem = '16MB'") # Mer minne för denna fråga
result = await conn.fetch(query, *params if params else ())
duration = time.time() - start_time
# Logga långsamma frågor
if duration > self.slow_query_threshold:
logger.warning(f"Slow query detected: {duration:.2f}s", extra={
"query": query[:200],
"duration": duration,
"params_count": len(params) if params else 0
})
# Cachera framgångsrika resultat
if cache_key and len(result) < 1000: # Cachera inte stora resultat
self.query_cache[cache_key] = {
'result': result,
'timestamp': time.time()
}
return result
except Exception as e:
logger.error(f"Query optimization failed: {e}")
raise
# Indexrekommendationer
RECOMMENDED_INDEXES = [
# Kärnverksamhetens index
"CREATE INDEX CONCURRENTLY idx_orders_store_date ON retail.orders (store_id, order_date DESC);",
"CREATE INDEX CONCURRENTLY idx_order_items_product ON retail.order_items (product_id);",
"CREATE INDEX CONCURRENTLY idx_customers_store_email ON retail.customers (store_id, email);",
# Analysindex
"CREATE INDEX CONCURRENTLY idx_orders_date_amount ON retail.orders (order_date, total_amount);",
"CREATE INDEX CONCURRENTLY idx_products_category_price ON retail.products (category_id, unit_price);",
# Vektorsökningsoptimering
"CREATE INDEX CONCURRENTLY idx_embeddings_vector ON retail.product_description_embeddings USING ivfflat (description_embedding vector_cosine_ops) WITH (lists = 100);",
]
Applikationsprestanda
Asynkrona programmerings bästa metoder
import asyncio
from asyncio import Semaphore
from typing import List, Any
class AsyncOptimizer:
"""Async operation optimization patterns."""
def __init__(self, max_concurrent: int = 10):
self.semaphore = Semaphore(max_concurrent)
self.circuit_breaker = CircuitBreaker()
async def batch_process(
self,
items: List[Any],
process_func: callable,
batch_size: int = 100
):
"""Process items in optimized batches."""
async def process_batch(batch):
async with self.semaphore:
return await asyncio.gather(
*[process_func(item) for item in batch],
return_exceptions=True
)
# Bearbeta i batchar för att undvika att överbelasta systemet
results = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
batch_results = await process_batch(batch)
results.extend(batch_results)
# Liten fördröjning mellan batcharna för att förhindra resursutarmning
if i + batch_size < len(items):
await asyncio.sleep(0.1)
return results
@circuit_breaker_decorator
async def resilient_operation(self, operation: callable, *args, **kwargs):
"""Execute operation with circuit breaker protection."""
return await operation(*args, **kwargs)
# Implementering av säkring
class CircuitBreaker:
"""Circuit breaker for external service calls."""
def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # STÄNGD, ÖPPEN, HALVÖPPEN
async def call(self, func, *args, **kwargs):
"""Execute function with circuit breaker protection."""
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "HALF_OPEN"
else:
raise Exception("Circuit breaker is OPEN")
try:
result = await func(*args, **kwargs)
# Återställ vid framgång
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
raise
Cachningsstrategier
import redis
import pickle
from typing import Union, Optional
class SmartCache:
"""Multi-level caching system."""
def __init__(self, redis_url: Optional[str] = None):
self.memory_cache = {}
self.redis_client = redis.Redis.from_url(redis_url) if redis_url else None
self.max_memory_items = 1000
async def get(self, key: str) -> Optional[Any]:
"""Get from cache with fallback levels."""
# Nivå 1: Minnescache
if key in self.memory_cache:
return self.memory_cache[key]['value']
# Nivå 2: Redis-cache
if self.redis_client:
try:
cached_data = self.redis_client.get(key)
if cached_data:
value = pickle.loads(cached_data)
# Främja till minnescache
self._set_memory_cache(key, value)
return value
except Exception as e:
logger.warning(f"Redis cache error: {e}")
return None
async def set(
self,
key: str,
value: Any,
ttl: int = 300,
cache_level: str = "both"
):
"""Set cache value at specified levels."""
if cache_level in ["memory", "both"]:
self._set_memory_cache(key, value, ttl)
if cache_level in ["redis", "both"] and self.redis_client:
try:
self.redis_client.setex(
key,
ttl,
pickle.dumps(value)
)
except Exception as e:
logger.warning(f"Redis set error: {e}")
def _set_memory_cache(self, key: str, value: Any, ttl: int = 300):
"""Set value in memory cache with LRU eviction."""
# Implementera LRU-utvisning
if len(self.memory_cache) >= self.max_memory_items:
oldest_key = min(
self.memory_cache.keys(),
key=lambda k: self.memory_cache[k]['timestamp']
)
del self.memory_cache[oldest_key]
self.memory_cache[key] = {
'value': value,
'timestamp': time.time(),
'ttl': ttl
}
# Generering av cache-nyckel
def generate_cache_key(query: str, user_context: str, params: dict = None) -> str:
"""Generate consistent cache keys."""
key_components = [
query.strip().lower(),
user_context,
json.dumps(params, sort_keys=True) if params else ""
]
key_string = "|".join(key_components)
return hashlib.sha256(key_string.encode()).hexdigest()
🔒 Säkerhetshärdning
Autentisering och Auktorisering
from azure.identity import DefaultAzureCredential, ClientSecretCredential
from azure.keyvault.secrets import SecretClient
import jwt
from typing import Dict, List
class SecurityManager:
"""Comprehensive security management."""
def __init__(self):
self.key_vault_client = self._setup_key_vault()
self.token_blacklist = set()
def _setup_key_vault(self) -> SecretClient:
"""Initialize Azure Key Vault client."""
credential = DefaultAzureCredential()
vault_url = os.getenv("AZURE_KEY_VAULT_URL")
if vault_url:
return SecretClient(vault_url=vault_url, credential=credential)
return None
async def validate_request(self, request_headers: Dict[str, str]) -> Dict[str, Any]:
"""Comprehensive request validation."""
# Extrahera och validera autentisering
auth_token = request_headers.get("authorization", "").replace("Bearer ", "")
if not auth_token:
raise AuthenticationError("Missing authentication token")
# Validera token
user_context = await self._validate_token(auth_token)
# Kontrollera hastighetsbegränsning
await self._check_rate_limit(user_context["user_id"])
# Validera RLS-kontekst
rls_user_id = request_headers.get("x-rls-user-id")
if not self._validate_rls_access(user_context, rls_user_id):
raise AuthorizationError("Invalid RLS context for user")
return {
"user_id": user_context["user_id"],
"roles": user_context["roles"],
"rls_user_id": rls_user_id,
"permissions": user_context["permissions"]
}
async def _validate_token(self, token: str) -> Dict[str, Any]:
"""Validate JWT token."""
if token in self.token_blacklist:
raise AuthenticationError("Token has been revoked")
try:
# Hämta offentlig nyckel från Key Vault eller cache
public_key = await self._get_public_key()
# Avkoda och validera token
payload = jwt.decode(
token,
public_key,
algorithms=["RS256"],
audience="mcp-server",
issuer="zava-auth"
)
return {
"user_id": payload["sub"],
"roles": payload.get("roles", []),
"permissions": payload.get("permissions", []),
"expires_at": payload["exp"]
}
except jwt.InvalidTokenError as e:
raise AuthenticationError(f"Invalid token: {e}")
def _validate_rls_access(self, user_context: Dict, rls_user_id: str) -> bool:
"""Validate RLS context access."""
# Superadministratörer kan få åtkomst till vilken kontext som helst
if "super_admin" in user_context["roles"]:
return True
# Butikschefer kan endast få åtkomst till sin egen butik
if "store_manager" in user_context["roles"]:
allowed_stores = user_context.get("allowed_stores", [])
return rls_user_id in allowed_stores
# Regionchefer kan få åtkomst till flera butiker
if "regional_manager" in user_context["roles"]:
allowed_regions = user_context.get("allowed_regions", [])
return self._check_store_in_regions(rls_user_id, allowed_regions)
return False
# Indatavalidering och sanering
class InputValidator:
"""SQL injection prevention and input validation."""
@staticmethod
def validate_sql_query(query: str) -> bool:
"""Validate SQL query for safety."""
# Förbjudna mönster
forbidden_patterns = [
r";\s*(DROP|DELETE|UPDATE|INSERT|ALTER|CREATE)\s+",
r"--.*",
r"/\*.*\*/",
r"xp_cmdshell",
r"sp_executesql",
r"EXEC\s*\(",
]
query_upper = query.upper()
for pattern in forbidden_patterns:
if re.search(pattern, query_upper, re.IGNORECASE):
logger.warning(f"Blocked potentially dangerous query: {pattern}")
return False
# Tillåt endast SELECT-satser
if not query_upper.strip().startswith("SELECT"):
return False
return True
@staticmethod
def sanitize_table_name(table_name: str) -> str:
"""Sanitize table name input."""
# Tillåt endast alfanumeriska tecken, understreck och punkt
if not re.match(r"^[a-zA-Z0-9_.]+$", table_name):
raise ValueError("Invalid table name format")
# Validera mot tillåtna tabeller
if table_name not in VALID_TABLES:
raise ValueError(f"Table {table_name} not allowed")
return table_name
Dataskydd
from cryptography.fernet import Fernet
import hashlib
class DataProtection:
"""Data encryption and protection utilities."""
def __init__(self):
self.encryption_key = self._get_encryption_key()
self.cipher_suite = Fernet(self.encryption_key)
def _get_encryption_key(self) -> bytes:
"""Get encryption key from secure storage."""
# I produktion, hämta från Azure Key Vault
key_vault_secret = os.getenv("ENCRYPTION_KEY_SECRET_NAME")
if key_vault_secret and self.key_vault_client:
secret = self.key_vault_client.get_secret(key_vault_secret)
return secret.value.encode()
# Reservlösning för utveckling (inte för produktion!)
dev_key = os.getenv("DEV_ENCRYPTION_KEY")
if dev_key:
return dev_key.encode()
raise ValueError("No encryption key available")
def encrypt_sensitive_data(self, data: str) -> str:
"""Encrypt sensitive data."""
return self.cipher_suite.encrypt(data.encode()).decode()
def decrypt_sensitive_data(self, encrypted_data: str) -> str:
"""Decrypt sensitive data."""
return self.cipher_suite.decrypt(encrypted_data.encode()).decode()
@staticmethod
def hash_password(password: str, salt: str = None) -> tuple:
"""Hash password with salt."""
if not salt:
salt = os.urandom(32).hex()
password_hash = hashlib.pbkdf2_hmac(
'sha256',
password.encode(),
salt.encode(),
100000 # iterationer
).hex()
return password_hash, salt
@staticmethod
def mask_sensitive_logs(log_data: dict) -> dict:
"""Mask sensitive information in logs."""
sensitive_fields = [
'password', 'token', 'secret', 'key', 'authorization',
'x-api-key', 'client_secret', 'connection_string'
]
masked_data = log_data.copy()
for field in sensitive_fields:
if field in masked_data:
value = str(masked_data[field])
if len(value) > 4:
masked_data[field] = value[:2] + "*" * (len(value) - 4) + value[-2:]
else:
masked_data[field] = "***"
return masked_data
📊 Produktionsdistribution Riktlinjer
Infrastruktur som kod
# azure-pipelines.yml
trigger:
branches:
include:
- main
- release/*
variables:
- group: mcp-server-secrets
- name: imageRepository
value: 'zava-mcp-server'
- name: containerRegistry
value: 'zavamcpregistry.azurecr.io'
stages:
- stage: Build
displayName: Build and Test
jobs:
- job: Build
displayName: Build
pool:
vmImage: ubuntu-latest
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '3.11'
displayName: 'Use Python 3.11'
- script: |
python -m pip install --upgrade pip
pip install -r requirements.lock.txt
pip install pytest pytest-cov
displayName: 'Install dependencies'
- script: |
pytest tests/ --cov=mcp_server --cov-report=xml
displayName: 'Run tests with coverage'
- task: PublishCodeCoverageResults@1
inputs:
codeCoverageTool: Cobertura
summaryFileLocation: 'coverage.xml'
- task: Docker@2
displayName: Build Docker image
inputs:
command: build
repository: $(imageRepository)
dockerfile: Dockerfile
tags: |
$(Build.BuildId)
latest
- stage: Deploy
displayName: Deploy to Production
dependsOn: Build
condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
jobs:
- deployment: DeployProduction
displayName: Deploy to Production
environment: 'production'
pool:
vmImage: ubuntu-latest
strategy:
runOnce:
deploy:
steps:
- task: AzureContainerApps@1
inputs:
azureSubscription: $(azureServiceConnection)
containerAppName: 'zava-mcp-server'
resourceGroup: '$(resourceGroupName)'
imageToDeploy: '$(containerRegistry)/$(imageRepository):$(Build.BuildId)'
Containeroptimering
# Multi-stage Dockerfile for production
FROM python:3.11-slim as builder
# Install build dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Create virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Copy requirements and install Python dependencies
COPY requirements.lock.txt .
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir -r requirements.lock.txt
# Production stage
FROM python:3.11-slim as production
# Create non-root user
RUN groupadd -r mcpserver && useradd -r -g mcpserver mcpserver
# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Set working directory
WORKDIR /app
# Copy application code
COPY mcp_server/ ./mcp_server/
COPY --chown=mcpserver:mcpserver . .
# Set security configurations
RUN chmod -R 755 /app && \
chown -R mcpserver:mcpserver /app
# Switch to non-root user
USER mcpserver
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Expose port
EXPOSE 8000
# Start application
CMD ["python", "-m", "mcp_server.sales_analysis"]
Miljökonfiguration
# Produktionskonfigurationshantering
class ProductionConfig:
"""Production-specific configuration."""
def __init__(self):
self.validate_production_requirements()
self.setup_logging()
self.configure_security()
def validate_production_requirements(self):
"""Validate all required production settings."""
required_settings = [
"AZURE_CLIENT_ID",
"AZURE_CLIENT_SECRET",
"AZURE_TENANT_ID",
"PROJECT_ENDPOINT",
"AZURE_OPENAI_ENDPOINT",
"POSTGRES_HOST",
"POSTGRES_PASSWORD",
"APPLICATIONINSIGHTS_CONNECTION_STRING"
]
missing_settings = [
setting for setting in required_settings
if not os.getenv(setting)
]
if missing_settings:
raise EnvironmentError(
f"Missing required production settings: {missing_settings}"
)
def setup_logging(self):
"""Configure production logging."""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.handlers.RotatingFileHandler(
'/var/log/mcp-server.log',
maxBytes=50*1024*1024, # 50MB
backupCount=5
)
]
)
# Ställ in tredjepartsloggare på VARNING
logging.getLogger('azure').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
def configure_security(self):
"""Configure production security settings."""
# Inaktivera felsökningsläge
os.environ['DEBUG'] = 'False'
# Ställ in säkra headers
os.environ['SECURE_SSL_REDIRECT'] = 'True'
os.environ['SECURE_HSTS_SECONDS'] = '31536000'
os.environ['SECURE_CONTENT_TYPE_NOSNIFF'] = 'True'
os.environ['SECURE_BROWSER_XSS_FILTER'] = 'True'
💰 Kostnadsoptimering
Resurshantering
class CostOptimizer:
"""Cost optimization strategies."""
def __init__(self):
self.metrics_collector = MetricsCollector()
self.auto_scaler = AutoScaler()
async def optimize_database_connections(self):
"""Dynamically adjust connection pool based on load."""
current_load = await self.metrics_collector.get_current_load()
if current_load < 0.3: # Låg belastning
target_pool_size = max(2, int(current_load * 10))
elif current_load < 0.7: # Medellast
target_pool_size = max(5, int(current_load * 15))
else: # Hög belastning
target_pool_size = min(20, int(current_load * 25))
await db_provider.adjust_pool_size(target_pool_size)
logger.info(f"Adjusted pool size to {target_pool_size} for load {current_load}")
async def implement_smart_caching(self):
"""Implement intelligent caching to reduce compute costs."""
# Cacha kostsamma operationer
expensive_queries = await self.identify_expensive_queries()
for query in expensive_queries:
cache_key = self.generate_cache_key(query)
ttl = self.calculate_optimal_ttl(query)
await smart_cache.set(cache_key, None, ttl=ttl)
def calculate_azure_costs(self) -> Dict[str, float]:
"""Calculate estimated Azure resource costs."""
return {
"container_apps": self.estimate_container_costs(),
"postgresql": self.estimate_database_costs(),
"openai": self.estimate_ai_costs(),
"application_insights": self.estimate_monitoring_costs(),
"storage": self.estimate_storage_costs()
}
# Automatisk skalningskonfiguration
class AutoScaler:
"""Automatic scaling based on metrics."""
async def scale_decision(self) -> str:
"""Determine scaling action based on metrics."""
metrics = await self.collect_scaling_metrics()
# CPU-baserad skalning
if metrics['cpu_usage'] > 80:
return "scale_up"
elif metrics['cpu_usage'] < 20 and metrics['instance_count'] > 1:
return "scale_down"
# Minnesbaserad skalning
if metrics['memory_usage'] > 85:
return "scale_up"
# Skalning av förfrågningskö
if metrics['queue_length'] > 100:
return "scale_up"
elif metrics['queue_length'] < 10 and metrics['instance_count'] > 1:
return "scale_down"
return "no_action"
🔧 Underhåll och drift
Hälsomonitorering
class OperationalHealth:
"""Comprehensive operational health monitoring."""
def __init__(self):
self.alert_manager = AlertManager()
self.health_checks = {}
async def comprehensive_health_check(self) -> Dict[str, Any]:
"""Perform comprehensive system health check."""
health_report = {
"timestamp": datetime.utcnow().isoformat(),
"overall_status": "healthy",
"components": {}
}
# Databasens hälsa
db_health = await self.check_database_health()
health_report["components"]["database"] = db_health
# Hälsa för externa tjänster
ai_health = await self.check_ai_service_health()
health_report["components"]["ai_service"] = ai_health
# Systemresurser
system_health = await self.check_system_resources()
health_report["components"]["system"] = system_health
# Applikationsmått
app_health = await self.check_application_health()
health_report["components"]["application"] = app_health
# Bestäm övergripande status
failed_components = [
name for name, status in health_report["components"].items()
if status.get("status") != "healthy"
]
if failed_components:
health_report["overall_status"] = "unhealthy"
health_report["failed_components"] = failed_components
# Utlös varningar
await self.alert_manager.send_alert(
severity="high",
message=f"Health check failed for: {failed_components}",
details=health_report
)
return health_report
async def check_database_health(self) -> Dict[str, Any]:
"""Check database connectivity and performance."""
try:
start_time = time.time()
async with db_provider.get_connection() as conn:
# Grundläggande anslutning
await conn.fetchval("SELECT 1")
# Kontrollera långsamma frågor
slow_queries = await conn.fetch("""
SELECT query, mean_exec_time, calls
FROM pg_stat_statements
WHERE mean_exec_time > 1000
ORDER BY mean_exec_time DESC
LIMIT 5
""")
# Kontrollera antal anslutningar
connection_count = await conn.fetchval("""
SELECT count(*) FROM pg_stat_activity
WHERE state = 'active'
""")
response_time = time.time() - start_time
return {
"status": "healthy",
"response_time_ms": response_time * 1000,
"active_connections": connection_count,
"slow_queries_count": len(slow_queries),
"pool_size": db_provider.connection_pool.get_size()
}
except Exception as e:
return {
"status": "unhealthy",
"error": str(e),
"last_check": datetime.utcnow().isoformat()
}
# Automatisk säkerhetskopiering och återställning
class BackupManager:
"""Database backup and recovery management."""
async def create_backup(self, backup_type: str = "full") -> str:
"""Create database backup."""
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
backup_name = f"zava_backup_{backup_type}_{timestamp}"
if backup_type == "full":
await self.create_full_backup(backup_name)
elif backup_type == "incremental":
await self.create_incremental_backup(backup_name)
# Ladda upp till Azure Blob Storage
await self.upload_backup_to_azure(backup_name)
return backup_name
async def schedule_automated_backups(self):
"""Schedule regular automated backups."""
# Daglig fullständig säkerhetskopia kl 2 UTC
schedule.every().day.at("02:00").do(
lambda: asyncio.create_task(self.create_backup("full"))
)
# Timvisa inkrementella säkerhetskopior
schedule.every().hour.do(
lambda: asyncio.create_task(self.create_backup("incremental"))
)
🌍 Communitybidrag
Öppen källkods bästa metoder
# Contributing to MCP Database Integration
## Development Guidelines
### Code Quality Standards
- Follow PEP 8 for Python code style
- Maintain test coverage above 90%
- Use type hints throughout the codebase
- Write comprehensive docstrings
### Testing Requirements
- Unit tests for all new functionality
- Integration tests for database operations
- Performance benchmarks for critical paths
- Security tests for authentication/authorization
### Documentation Standards
- Update README.md for any new features
- Add inline code documentation
- Create examples for new tools or patterns
- Maintain API documentation
## Security Considerations
### Reporting Security Issues
- Report security vulnerabilities privately
- Use encrypted communication channels
- Provide detailed reproduction steps
- Include potential impact assessment
### Security Review Process
- All PRs undergo security review
- Static analysis tools required to pass
- Dependency vulnerability scanning
- Manual security testing for critical changes
Communityengagemang
class CommunityContributor:
"""Tools for community engagement and contribution."""
@staticmethod
def generate_contribution_guide():
"""Generate personalized contribution guide."""
return {
"getting_started": {
"setup": "Follow setup guide in Lab 03",
"first_contribution": "Start with documentation improvements",
"testing": "Run full test suite before submitting PR"
},
"contribution_areas": {
"documentation": "Improve learning labs and examples",
"testing": "Add test cases and improve coverage",
"features": "Implement new MCP tools and capabilities",
"performance": "Optimize queries and caching",
"security": "Enhance security measures and validation"
},
"community_resources": {
"discord": "https://discord.com/invite/ByRwuEEgH4",
"discussions": "GitHub Discussions for Q&A",
"issues": "GitHub Issues for bug reports",
"examples": "Share your implementation examples"
}
}
@staticmethod
def validate_contribution(pr_data: Dict) -> Dict[str, bool]:
"""Validate contribution meets standards."""
return {
"has_tests": "test" in pr_data.get("files_changed", []),
"has_documentation": "README" in str(pr_data.get("files_changed", [])),
"follows_conventions": True, # Skulle implementera faktiska kontroller
"security_reviewed": pr_data.get("security_review", False),
"performance_tested": pr_data.get("benchmark_results", False)
}
🎯 Viktiga slutsatser
Efter att ha slutfört denna omfattande lärandeväg bör du ha behärskat:
✅ Prestandaoptimering: Databasanpassning, asynkrona mönster och cachningsstrategier
✅ Säkerhetshärdning: Autentisering, auktorisering och dataskydd
✅ Produktionsdistribution: Infrastruktur som kod och containeroptimering
✅ Kostnadshantering: Resursoptimering och intelligent skalning
✅ Operativ excellens: Övervakning, underhåll och automatisering
✅ Communityengagemang: Bidra till MCP-ekosystemet
🏆 Certifiering och nästa steg
Praktisk bedömning
Slutför detta slutprojekt för att demonstrera din behärskning:
Bygg en produktionsklar MCP-server som inkluderar:
- Multi-tenant detaljhandelsanalys med RLS
- Semantisk sökning med Azure OpenAI
- Omfattande säkerhetsimplementering
- Produktionsdistribution på Azure
- Övervakning och larmuppsättning
- Dokumentation och testning
Avancerade lärandevägar
Fortsätt din MCP-resa med:
- MCP-arkitekturmönster: Avancerade serverarkitekturer
- Integration av flera modeller: Kombinera olika AI-modeller
- Företagsomfattning: Storskaliga MCP-distributioner
- Utveckling av specialanpassade verktyg: Bygga specialiserade MCP-verktyg
- MCP-ekosystem: Bidra till den bredare communityn
Community erkännande
Dela din prestation:
- GitHub-portfölj: Visa upp din implementation
- Communitybidrag: Skicka in förbättringar eller exempel
- Talarmöjligheter: Presentera på meetups eller konferenser
- Mentorskap: Hjälp andra utvecklare att lära sig MCP
📚 Ytterligare resurser
Avancerade ämnen
- PostgreSQL prestandajustering - Databasoptimering
- Azure Container Apps bästa metoder - Produktionsdistribution
- Python Async bästa metoder - Asynkron programmering
Säkerhetsresurser
- OWASP Topp 10 - Säkerhetsbrister
- Azure Security Bästa metoder - Molnsäkerhet
- Python säkerhetsriktlinjer - Säker kodning
Community
- MCP Community Discord - Live-diskussioner
- GitHub Discussions - Frågor och delning
- Stack Overflow - Tekniska frågor
🎉 Grattis! Du har slutfört den omfattande lärandevägen för MCP-databasintegration. Du har nu kunskapen och färdigheterna för att bygga produktionsklara MCP-servrar som länkar AI-assistenter med verkliga datasystem.
Redo att bidra? Gå med i vår community och hjälp andra att lära sig MCP genom att dela dina erfarenheter, bidra med kodförbättringar eller skapa ytterligare lärresurser.
Nästa: Tooling
Ansvarsfriskrivning: Detta dokument har översatts med hjälp av AI-översättningstjänsten Co-op Translator. Även om vi strävar efter noggrannhet, var vänlig notera att automatiska översättningar kan innehålla fel eller brister. Det ursprungliga dokumentet på dess modersmål bör betraktas som den auktoritativa källan. För kritisk information rekommenderas professionell mänsklig översättning. Vi ansvarar inte för några missförstånd eller feltolkningar som uppstår till följd av användningen av denna översättning.