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

تتعطل خوادم MCP بطريقة يسهل إغفالها على غير العادة. فحين تُرجع أداة نتيجة خاطئة لا ينهار النموذج، بل يدمج النتيجة الخاطئة ويجيب بثقة. لا تتبّع استدعاءات ولا صفحة خطأ. والمكان الوحيد الذي يظهر فيه هذا الإخفاق هو ما اخترت أنت تسجيله.

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

مقياس واحد هنا يفوق البقية قيمة: عدد النداءات حسب الأداة. فهو يخبرك أي الأدوات يستخدمها النموذج فعلاً. والأداة التي تُستدعى أكثر بكثير من المتوقع يكون وصفها عادةً متداخلاً مع وصف أداة أخرى. والأداة التي لا تُستدعى إطلاقاً إما أن وصفها رديء أو أنها ما كان ينبغي أن تُكتب — وفي الحالتين لا يمكنك معرفة ذلك من السجلات وحدها.

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

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: "استدعاءات الأدوات تستغرق وقتاً طويلاً"

التالي: مستوى الاختبار الذي يلتقط ما تعجز عنه اختبارات الوحدة بنيوياً. :::