أنظمة MCP الإنتاجية
المراقبة والملاحظة
تتعطل خوادم MCP بطريقة يسهل إغفالها على غير العادة. فحين تُرجع أداة نتيجة خاطئة لا ينهار النموذج، بل يدمج النتيجة الخاطئة ويجيب بثقة. لا تتبّع استدعاءات ولا صفحة خطأ. والمكان الوحيد الذي يظهر فيه هذا الإخفاق هو ما اخترت أنت تسجيله.
الركائز الثلاث
ماذا تسجّل، وعن أي سؤال تجيب كل طبقة
مقياس واحد هنا يفوق البقية قيمة: عدد النداءات حسب الأداة. فهو يخبرك أي الأدوات يستخدمها النموذج فعلاً. والأداة التي تُستدعى أكثر بكثير من المتوقع يكون وصفها عادةً متداخلاً مع وصف أداة أخرى. والأداة التي لا تُستدعى إطلاقاً إما أن وصفها رديء أو أنها ما كان ينبغي أن تُكتب — وفي الحالتين لا يمكنك معرفة ذلك من السجلات وحدها.
التسجيل المنظم
import logging
import json
from datetime import datetime, timezone
class JSONFormatter(logging.Formatter):
def format(self, record):
log_data = {
"timestamp": datetime.now(timezone.utc).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
import time
from prometheus_client import Counter, Histogram, generate_latest
from starlette.responses import Response
from mcp.server import MCPServer
from mcp.shared.exceptions import MCPError
TOOL_CALLS = Counter(
"mcp_tool_calls_total", "إجمالي استدعاءات الأدوات", ["tool_name", "status"]
)
TOOL_LATENCY = Histogram(
"mcp_tool_latency_seconds", "زمن استجابة استدعاء الأداة", ["tool_name"]
)
async def metrics_middleware(ctx, call_next):
"""تسجيل كل tools/call سواء نجح أو أطلق استثناءً."""
if ctx.method != "tools/call":
return await call_next(ctx)
tool = (ctx.params or {}).get("name", "unknown")
start = time.perf_counter()
try:
result = await call_next(ctx)
TOOL_CALLS.labels(tool_name=tool, status="success").inc()
return result
except MCPError:
TOOL_CALLS.labels(tool_name=tool, status="error").inc()
raise
finally:
TOOL_LATENCY.labels(tool_name=tool).observe(time.perf_counter() - start)
mcp = MCPServer(name="observable-server", middleware=[metrics_middleware])
القياس عبر الوسيط بدل وضعه داخل كل أداة هو ما يجعل الأرقام جديرة بالثقة. فالأداة التي تنسى تغليفها أداةٌ تُبلِّغ صامتةً عن صفر استدعاء، والمقياس الغائب يبدو تماماً كالمقياس السليم.
أمران يجب ضبطهما، وكتلة finally أعلاه تقوم بهما عن قصد:
- سجّل حالات الفشل. الوسيط الذي يَعُدّ في مسار النجاح فقط يُنتج معدل أخطاء يساوي صفراً مهما كان أداء الخادم سيئاً.
- سجّل زمن الفشل أيضاً. فحالات انتهاء المهلة هي أبطأ استدعاءاتك، واستبعادها يجعل زمن الاستجابة في الذيل يبدو في أفضل حالاته في اللحظة التي يكون فيها في أسوأ حالاته.
اكشف السجل من تطبيق ASGI نفسه الذي يخدم MCP:
app = mcp.streamable_http_app()
async def metrics(request):
return Response(generate_latest(), media_type="text/plain")
app.add_route("/metrics", metrics)
فحوصات الصحة
@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: "استدعاءات الأدوات تستغرق وقتاً طويلاً"
التالي: مستوى الاختبار الذي يلتقط ما تعجز عنه اختبارات الوحدة بنيوياً. :::
سجّل الدخول للتقييم