المراقبة والملاحظة

تحتاج خوادم MCP الإنتاجية إلى مراقبة شاملة لضمان الموثوقية وتصحيح المشاكل.

الركائز الثلاث

الركيزة	الغرض	الأدوات
السجلات	تفاصيل الأحداث	التسجيل المنظم
المقاييس	القياسات	Prometheus، StatsD
التتبعات	تدفق الطلبات	OpenTelemetry

التسجيل المنظم

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        if hasattr(record, "extra"):
            log_data.update(record.extra)
        return json.dumps(log_data)

# تكوين التسجيل
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger("mcp")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# الاستخدام
logger.info("أداة مستدعاة", extra={"tool": "search", "user": "user123"})

المقاييس مع Prometheus

from prometheus_client import Counter, Histogram, generate_latest
from starlette.responses import Response

# تعريف المقاييس
TOOL_CALLS = Counter(
    "mcp_tool_calls_total",
    "إجمالي استدعاءات الأدوات",
    ["tool_name", "status"]
)

TOOL_LATENCY = Histogram(
    "mcp_tool_latency_seconds",
    "زمن انتظار استدعاء الأداة",
    ["tool_name"]
)

# قياس استدعاءات الأدوات
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    with TOOL_LATENCY.labels(tool_name=name).time():
        try:
            result = await execute_tool(name, arguments)
            TOOL_CALLS.labels(tool_name=name, status="success").inc()
            return result
        except Exception as e:
            TOOL_CALLS.labels(tool_name=name, status="error").inc()
            raise

# نقطة نهاية المقاييس
@app.route("/metrics")
async def metrics(request):
    return Response(generate_latest(), media_type="text/plain")

فحوصات الصحة

@app.route("/health")
async def health(request):
    checks = {
        "database": await check_database(),
        "redis": await check_redis(),
        "external_api": await check_external_api(),
    }

    all_healthy = all(checks.values())
    status_code = 200 if all_healthy else 503

    return JSONResponse(
        {"status": "healthy" if all_healthy else "unhealthy", "checks": checks},
        status_code=status_code
    )

async def check_database():
    try:
        await db.execute("SELECT 1")
        return True
    except:
        return False

قواعد التنبيه

تكوين التنبيهات في Prometheus/Grafana:

groups:
  - name: mcp-alerts
    rules:
      - alert: HighErrorRate
        expr: rate(mcp_tool_calls_total{status="error"}[5m]) > 0.1
        for: 5m
        annotations:
          summary: "معدل خطأ عالٍ في خادم MCP"

      - alert: SlowToolCalls
        expr: histogram_quantile(0.95, mcp_tool_latency_seconds) > 5
        for: 5m
        annotations:
          summary: "استدعاءات الأدوات تستغرق وقتاً طويلاً"

في القسم التالي، سنستكشف استراتيجيات الاختبار. :::