بناء خوادم MCP
معالجة الأخطاء في MCP
لخادم MCP جمهور غير معتاد لرسائل أخطائه: نموذج لغوي. فهو لا يقرأ سجلاتك، ولن يفتح تتبّع الاستدعاءات، وسيختلق إجابة معقولة بلا تردد إن لم تخبره رسالة الإخفاق بشيء. معالجة الأخطاء هنا تدور حول كتابة إخفاقات يستطيع النموذج التصرف بناءً عليها، أكثر مما تدور حول الانهيار الآمن.
أنواع أخطاء MCP
يستخدم MCP رموز JSON-RPC 2.0 القياسية لإخفاقات البروتوكول العامة:
| الرمز | الاسم | الوصف |
|---|---|---|
| -32700 | خطأ تحليل | JSON غير صالح مستلم |
| -32600 | طلب غير صالح | طلب مشوه |
| -32601 | الطريقة غير موجودة | طريقة غير معروفة مستدعاة |
| -32602 | معلمات غير صالحة | معلمات خاطئة، ويشمل ذلك معرّف مورد لا يُفضي إلى شيء |
| -32603 | خطأ داخلي | خطأ من جانب الخادم |
وفوق ذلك تعرّف المواصفة رموزها الخاصة وتقسّم المدى المخصص للتنفيذات:
| الرمز | الاسم | متى |
|---|---|---|
| -32020 | HeaderMismatch | البيانات الوصفية في الغلاف تخالف جسم الطلب |
| -32021 | MissingRequiredClientCapability | الطلب يحتاج قدرة لم يعلنها العميل، ويسردها data.requiredCapabilities |
| -32022 | UnsupportedProtocolVersion | إصدار البروتوكول في الطلب ليس مما تتحدثه |
وهناك قاعدتان بشأن المديات تهمّان حين تبتكر رموزك الخاصة:
- المدى
-32000إلى-32019مغلق. فقد وُزّعت رموزه قبل السياسة الحالية، وينبغي للتنفيذات الجديدة ألا تستخدمها إطلاقاً. وأبرز المتقاعدين-32002(المورد غير موجود)، وقد حلّ محله-32602— اقبل-32002من الخوادم الأقدم، ولا تُصدره أبداً. - المدى
-32020إلى-32099ملك للمواصفة. لا تخصص فيه شيئاً. والرموز الخاصة بتطبيقك تذهب خارج مدى JSON-RPC المحجوز (-32768إلى-32000) بالكامل.
رفع الأخطاء في الأدوات
عندما تواجه أداة خطأ، ارفعه برسالة وصفية:
from mcp.types import McpError
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "get_user":
user_id = arguments.get("user_id")
if not user_id:
raise McpError(
code=-32602,
message="user_id مطلوب"
)
user = await db.get_user(user_id)
if not user:
raise McpError(
code=-32603,
message=f"المستخدم {user_id} غير موجود"
)
return [TextContent(type="text", text=user.to_json())]
فئات الأخطاء
كل استثناء يمكن أن يرفعه كودك يجب أن يقع على مسار واحد بالضبط من هذه المسارات. حدّد هذا الربط مرة واحدة في مكان واحد، وسيكتب المعالج التالي نفسه بنفسه:
توجيه الاستثناء إلى رمز الخطأ الصحيح
العمود الأيمن هو المهم هنا. فالنموذج الذي يعيد المحاولة على سجل غير موجود فعلاً يحرق أدواراً بلا فائدة، والنموذج الذي يستسلم أمام انقطاع مؤقت يفقد عملاً كان يمكن استرجاعه. ورسالتك أنت هي ما يحدد أي المسارين يقع.
تعامل مع الأخطاء على المستوى المناسب:
@server.call_tool()
async def call_tool(name: str, arguments: dict):
try:
# أخطاء التحقق
validate_arguments(name, arguments)
# أخطاء منطق الأعمال
result = await execute_tool(name, arguments)
return result
except ValidationError as e:
# خطأ العميل - معلمات سيئة
raise McpError(code=-32602, message=str(e))
except NotFoundError as e:
# المورد غير موجود
raise McpError(code=-32603, message=str(e))
except ExternalAPIError as e:
# فشل الخدمة الخارجية
raise McpError(
code=-32603,
message=f"الخدمة الخارجية غير متاحة: {e}"
)
except Exception as e:
# خطأ غير متوقع - سجله
logger.exception("خطأ غير متوقع في استدعاء الأداة")
raise McpError(
code=-32603,
message="حدث خطأ غير متوقع"
)
رسائل خطأ صديقة للمستخدم
النموذج يقرأ رسالة الخطأ ولا يقرأ سواها. اكتبها لقارئ لا يملك أي وصول إلى بنيتك التحتية وعليه أن يقرر الآن: هل يعيد المحاولة أم لا.
الإخفاق نفسه برسالتين
python1raise McpError(2 code=-32603,3 message="ERR_DB_CONN_FAIL"4)56# النموذج يرى رمزاً مبهماً.7# النتائج المعتادة: يعيد النداء نفسه في حلقة،8# أو يعتذر للمستخدم ويخترع إجابة من ذاكرته.
1raise McpError(2 code=-32603,3 message=(4 "تعذّر الوصول إلى قاعدة المعرفة (انتهت مهلة "5 "الاتصال بعد 5 ثوانٍ). هذا عارض غالباً، وإعادة "6 "المحاولة مرة واحدة تصرف معقول. وإن أخفقت ثانية، "7 "أبلغ المستخدم بأن قاعدة المعرفة غير متاحة بدل "8 "الإجابة من الذاكرة."9 )10)1112# تذكر ما الذي تعطّل، وهل هو عارض،13# وماذا تفعل إن أخفقت إعادة المحاولة أيضاً.
العبارة الأخيرة — بدل الإجابة من الذاكرة — تؤدي عملاً حقيقياً. فبغير توجيه بديل يلجأ إليه، يميل النموذج بعد استنفاد محاولاته إلى سدّ الفجوة بنفسه، وإجابة خاطئة بثقة أسوأ من خطأ ظاهر.
التسجيل لأغراض التصحيح
سجل الأخطاء دائماً مع السياق:
import logging
logger = logging.getLogger("mcp-server")
@server.call_tool()
async def call_tool(name: str, arguments: dict):
logger.info(f"أداة مستدعاة: {name}", extra={"arguments": arguments})
try:
result = await execute_tool(name, arguments)
logger.info(f"نجحت الأداة: {name}")
return result
except Exception as e:
logger.error(
f"فشلت الأداة: {name}",
extra={"arguments": arguments, "error": str(e)},
exc_info=True
)
raise
في القسم التالي، سنبني تمرين مختبر كامل للخادم. :::
سجّل الدخول للتقييم