كيفية بناء أول خادم MCP خاص بك باستخدام Novita AI

كيفية بناء أول خادم MCP خاص بك باستخدام Novita AI

بروتوكول سياق النموذج (MCP) أصبح بسرعة أحد أكثر المفاهيم تداولاً في عالم الذكاء الاصطناعي. على الرغم من حداثته النسبية، إلا أنه جذب بالفعل انتباه المطورين وشركات التكنولوجيا الكبرى على حد سواء.

تم إنشاء MCP بواسطة Anthropic، وهو بروتوكول قياسي مفتوح، مشابه لبروتوكولات أخرى مثل HTTP. ومع ذلك، بينما تم تصميم HTTP لتوصيل المستخدمين بالموارد عبر خادم ومتصفح ويب (أو عميل HTTP)، تم تصميم MCP لتوصيل نماذج اللغة الكبيرة (LLMs) بالأدوات والبيانات الخارجية.

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

هنا يأتي دور خادم MCP.

يستضيف خادم MCP العديد من الأدوات التي يمكن لروبوت المحادثة استدعاؤها باستخدام نموذج اللغة الكبير الداخلي الخاص به للوصول إلى هذه الإمكانيات. في حالة GitHub، يمكننا استخدام خادم MCP الرسمي لـ GitHub لأداء جميع المهام المذكورة أعلاه.

في هذه المقالة، سنقوم ببناء خادم MCP باستخدام Novita API. باستخدام خادم MCP هذا، ستتمكن أي تطبيق يدعم بروتوكول MCP من الوصول إلى جميع نماذج اللغة الكبيرة المتاحة على Novita، وإنشاء الصور، وإنشاء مقاطع الفيديو، وإجراء التركيب الصوتي.

فهم بنية MCP

بنية MCP

بنية MCP [المصدر]

تشبه بنية MCP إعداد خادم-عميل نموذجي. يتكون من أربعة مكونات مهمة:

  • عميل MCP
  • خادم MCP
  • وسائط نقل MCP
  • مضيف MCP

عميل MCP

يعمل عميل MCP كبوابة اتصال بين تطبيق الذكاء الاصطناعي وخادم MCP. بمجرد إنشاء هذا الاتصال، يمكن للتطبيق الوصول إلى جميع الأدوات والموارد التي يمتلكها الخادم.

خادم MCP

يستضيف خادم MCP جميع الأدوات والبيانات التي يمكن للعميل توفيرها لتطبيق الذكاء الاصطناعي. يستضيف ما يلي:

  • الأدوات
  • الموارد
  • الاستدعاءات

الأدوات

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

يمكن لخادم MCP استضافة عدة أدوات. تمكن هذه الأدوات نموذج اللغة الكبير من العمل كعامل، بسبب الإجراءات التي يمكنه تنفيذها.

تتم إدارة الأدوات بواسطة العامل وقد تتضمن إشرافًا بشريًا اختياريًا أثناء الاستخدام. وهي مشابهة لطلبات POST في بروتوكول HTTP لأنها تؤدي تأثيرات جانبية.

الموارد

الموارد هي قطع معلومات للقراءة فقط توفر البيانات لتطبيق الذكاء الاصطناعي. وهي مشابهة لطلبات HTTP GET، لأنها ليست مخصصة لإحداث أي تأثيرات جانبية. يُقصد من الموارد أن تتم إدارتها بواسطة التطبيق، الذي يقرر كيفية تفاعل المستخدمين أو العامل معها. تتضمن أمثلة الموارد محتويات الملفات، واستجابات API، والسجلات من قواعد البيانات.

الاستدعاءات

يمكن لخوادم MCP استضافة قوالب استدعاء. تمكن هذه الاستدعاءات المستخدم من جلب الاستدعاءات من خادم MCP، والتي يمكن استخدامها بعد ذلك في تطبيق الذكاء الاصطناعي. تتم إدارة الاستدعاءات بواسطة المستخدم، الذي يقرر أي الاستدعاءات يقدمها للعامل.

كل هذه الأشياء تشكل خادم MCP. في هذه المقالة، سنقوم فقط بتنفيذ الأدوات، لأنها الجزء الأكثر استخدامًا من خادم MCP.

وسائط نقل MCP

تشير وسائط نقل MCP إلى الطريقة التي يتواصل بها عميل MCP مع خادم MCP. يمكن أن يحدث هذا الاتصال محليًا، عندما يعمل كل من العميل والخادم على نفس الجهاز، أو عن بُعد، عندما يكونان على أجهزة منفصلة. يدعم MCP حاليًا آليتي نقل رئيسيتين:

  • STDIO: في هذا الوضع، يعمل خادم MCP والعميل على نفس الجهاز ويتواصلان عبر الإدخال والإخراج القياسيين.
  • SSE: في هذا الوضع، يعمل خادم MCP عبر HTTP. يُستخدم HTTP POST لإرسال الرسائل إلى الخادم، بينما تُستخدم أحداث الخادم المرسلة (Server-Sent Events) لإرسال الرسائل من الخادم إلى العميل.
  • HTTP قابل للبث: يستخدم هذا الوضع أيضًا HTTP ولكنه يعتمد على طلبات HTTP GET و POST. يتراجع إلى SSE فقط عندما يحتاج إلى بث رسائل متعددة من الخادم إلى العميل.

الإدخال/الإخراج القياسي

نقل باستخدام STDIO

نقل MCP باستخدام STDIO [المصدر]

التسمية التوضيحية: نقل MCP باستخدام STDIO [المصدر]

يمكن لعميل MCP والخادم التواصل عبر الإدخال والإخراج القياسيين (STDIO). عند استخدام STDIO، يعمل كل من العميل والخادم على نفس الجهاز. يكتب العميل جميع الطلبات إلى stdin، بينما يكتب الخادم الاستجابة إلى stdout.

HTTP قابل للبث

نقل MCP باستخدام HTTP قابل للبث

نقل MCP باستخدام HTTP قابل للبث [المصدر]

يسمح HTTP القابل للبث لعميل MCP والخادم بالتواصل عبر HTTP. يستخدم طريقة HTTP POST لإرسال الطلبات من العميل واستلام الردود من الخادم. عند الحاجة، يمكنه التبديل اختياريًا إلى أحداث الخادم المرسلة (SSE) لبث الرسائل من الخادم إلى العميل.

HTTP القابل للبث مناسب تمامًا للاتصال عن بُعد بين العميل والخادم ويعمل كبديل لنقل MCP SSE القديم.

مضيف MCP

حيث يختلف MCP عن بنية خادم-عميل النموذجية هو دور المضيف. يعرّف مواصفات MCP على أنها بنية خادم-عميل-مضيف. وذلك لأن العميل والمضيف يشكلان الواجهة الأمامية للبنية. يتكون مضيف MCP من مكونين مهمين:

  • عميل (عملاء) MCP
  • نموذج لغة كبير

مهمة عميل MCP هي جلب الأدوات والموارد والاستدعاءات التي يحتاجها نموذج اللغة الكبير من الخادم. بمجرد جمع الموارد، يتم وضعها داخل سياق النموذج. يمكن أن يحتوي مضيف MCP الواحد على عدة عملاء، كل منهم متصل بخادم MCP الخاص به.

يمكن أيضًا اعتبار مضيف MCP هو التطبيق الذي يعمل عليه المستخدم، والذي قد يؤدي وظائف أخرى. على سبيل المثال، Claude Desktop هو مضيف MCP يعمل كروبوت محادثة، وCursor هو مضيف MCP يعمل أيضًا كبيئة تطوير متكاملة (IDE)، وClaude Code هو مضيف MCP مصمم كعامل ترميز ذكي.

العمل مع خادم MCP

قبل أن نبدأ في بناء خادم MCP الخاص بنا، دعنا نرى كيف يمكننا العمل مع خادم MCP موجود. سنستخدم Python MCP SDK للتفاعل مع novita-mcp-server.

في وقت كتابة هذه المقالة، يوفر novita-mcp-server الأدوات التالية:

  • أداة لسرد جميع مجموعات Novita
  • أداة لسرد منتجات مثيلات GPU من Novita
  • أداة لسرد جميع مثيلات GPU قيد التشغيل
  • أداة لإنشاء مثيلات GPU جديدة

دعنا نؤكد ذلك عن طريق كتابة نص بسيط يتصل بخادم MCP عبر STDIO ويسرد جميع الأدوات المتاحة على الخادم.

للبدء، دعنا نقوم بتثبيت MCP SDK: pip install "mcp[cli]"

بعد تثبيت SDK، يمكننا إنشاء ملف باسم client.py. ثم سنقوم بعمل جميع الاستيرادات الضرورية:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import os

سنستخدم بعد ذلك StdioServerParameters لتعيين معاملاتنا لتشغيل خادم MCP الخاص بنا عبر stdio. يتم تنفيذ novita-mcp-server في Node.js، لذا لاستخدامه، نحتاج إلى تشغيله باستخدام الأمر npx. نحتاج أيضًا إلى تخزين مفتاح API الخاص بـ Novita في متغير البيئة NOVITA_API_KEY.

# إنشاء معاملات الخادم للاتصال عبر stdio
server_params = StdioServerParameters(
   command="npx",
   args=["-y", "@novitalabs/novita-mcp-server"],
   env={"NOVITA_API_KEY":  os.environ["NOVITA_API_KEY"]},
)

بعد ذلك، دعنا ننشئ دالة غير متزامنة. داخل الدالة، سنمرر معاملات الخادم إلى دالة stdio_client، والتي ستنشئ سياقًا يعيد تيارات القراءة والكتابة. تمكننا هذه التيارات من القراءة من stdio والكتابة إليه، على التوالي.

async def run():
   async with stdio_client(server_params) as (read, write):
       async with ClientSession(read, write) as session:
           # تهيئة الاتصال
           await session.initialize()

           # سرد الأدوات المتاحة
           tools = await session.list_tools()
           print("Available tools:", tools)

تُستخدم هذه التيارات بعد ذلك لإنشاء جلسة باستخدام فئة ClientSession. بمجرد إنشاء الجلسة، يمكننا تهيئتها ثم سرد جميع الأدوات على الخادم.

لتشغيل هذا البرنامج، نحتاج فقط إلى استخدام مكتبة asyncio لاستدعاء الدالة run. إليك الكود الكامل:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import os

# إنشاء معاملات الخادم للاتصال عبر stdio
server_params = StdioServerParameters(
   command="npx",
   args=["-y", "@novitalabs/novita-mcp-server"],
   env={"NOVITA_API_KEY":  os.environ["NOVITA_API_KEY"]},
)

async def run():
   async with stdio_client(server_params) as (read, write):
       async with ClientSession(read, write) as session:
           # تهيئة الاتصال
           await session.initialize()

           # سرد الأدوات المتاحة
           tools = await session.list_tools()
           print("Available tools:", tools)

if __name__ == "__main__":
   import asyncio

   asyncio.run(run())

من خلال تنفيذ البرنامج النصي، يمكننا رؤية جميع الأدوات على خادم Novita MCP.

يوضح هذا المثال كيف يمكنك استخدام عميل MCP. يمكننا إضافة نموذج لغة كبير إلى هذا المثال وتحويله بشكل فعال إلى مضيف، ولكن هذا ليس محور هذا البرنامج التعليمي. المحور هو بناء خادم MCP.

بناء خادم MCP باستخدام FastMCP

في Python MCP SDK، هناك طريقتان لبناء خوادم MCP. إحداهما باستخدام الخادم منخفض المستوى، والأخرى باستخدام FastMCP. FastMCP هي فئة داخل MCP Python SDK تستلهم من FastAPI لجعل إنشاء خوادم MCP أسهل.

دعنا نستخدم FastMCP لإنشاء خادم MCP الخاص بنا. قبل البدء، دعنا نفكر في وظائف خادم MCP الخاص بنا. خادم Novita MCP الحالي ينفذ فقط أدوات تتعامل مع إدارة GPU. دعنا نحاول بناء خادم حول Novita API يتجاوز إدارة GPU.

بدلاً من ذلك، دعنا نبني خادم MCP يوفر الوصول إلى النماذج على منصة Novita. سيكون خادم MCP الخاص بنا يحتوي فقط على أدوات، بدون موارد أو استدعاءات. فيما يلي قائمة بالأدوات التي سيحتويها الخادم:

  • list_models: ستسرد هذه الأداة جميع نماذج اللغة الكبيرة على منصة Novita.
  • get_model: ستسترد هذه الأداة نموذج لغة كبير معين وتستخدمه.
  • text2image: ستنشئ هذه الأداة صورًا من استدعاء معين.
  • task_result: ستستخدم هذه الأداة للحصول على حالة مهمة قيد التشغيل عبر معرفها.
  • text_to_speech: ستحول هذه الأداة النص المقدم إلى كلام.
  • generate_video: ستنشئ هذه الأداة مقطع فيديو من استدعاء معين.

الآن بعد أن عرفنا الأدوات التي سنبنيها، دعنا نبدأ في إنشاء الخادم. أولاً، قم بإنشاء ملف باسم server.py وأضف الكود التالي إليه:

import os
import sys
from mcp.server.fastmcp import FastMCP
import requests
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Mount

base_url = "https://api.novita.ai/v3"

headers = {
   "Content-Type": "application/json",
   "Authorization": f"Bearer {os.environ['NOVITA_API_KEY']}"
}

mcp = FastMCP("Novita_API")

بهذا، قمنا باستيراد الوحدات الضرورية، وتحديد عنوان URL الأساسي ورؤوس API، وإنشاء مثيل لفئة FastMCP. دعنا ننتقل إلى إنشاء أدواتنا.

قائمة النماذج

لإنشاء أداة في FastMCP، نحتاج إلى تزيين دالة بطريقة الأداة من مثيل FastMCP. تقوم دالة list_models بإجراء استدعاء لنقطة النهاية لقائمة النماذج للحصول على قائمة النماذج. يتم تنسيق الرد بشكل صحيح بحيث يمكن تمريره إلى نموذج اللغة الكبير الذي استدعاه.

@mcp.tool()
def list_models() -> str:
   """
   سرد جميع النماذج المتاحة من Novita API.
   """

   url = base_url + "/openai/models"

   response = requests.request("GET", url, headers=headers)
   data = response.json()["data"]
  
   text = ""
   for i, model in enumerate(data, start=1):
       text += f"Model id: {model['id']}\n"
       text += f"Model description: {model['description']}\n"
       text += f"Model type: {model['model_type']}\n\n"
  
   return text

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

الحصول على النماذج

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

@mcp.tool()
def get_model(model_id: str, message):
   """
   توفير معرف النموذج ورسالة للحصول على رد من Novita API.
   """

   url = base_url + "/openai/chat/completions"

   payload = {
       "model": model_id,
       "messages": [
           {
               "content": message,
               "role": "user",
           }
       ],
       "max_tokens": 200,
       "response_format": {
           "type": "text",
       },
   }

   response = requests.request("POST", url, json=payload, headers=headers)
  
   content = response.json()["choices"][0]["message"]["content"]

   return content

تحويل النص إلى صورة

تنشئ هذه الأداة صورًا من استدعاء. تستخدم نقطة نهاية تحويل النص إلى صورة من Novita داخليًا. نقطة النهاية هذه غير متزامنة، لذا لا تعيد الصورة فورًا؛ بل تعيد معرف مهمة.

@mcp.tool()
def text2Image(prompt):
   """
   إنشاء صورة من استدعاء نصي باستخدام Novita API.
   """

   url = base_url + "/async/txt2img"

   payload = {
       "request": {
           "model_name": "sd_xl_base_1.0.safetensors",
           "prompt": prompt,
           "width": 1024,
           "height": 1024,
           "image_num": 1,
           "steps": 20,
           "clip_skip": 1,
           "sampler_name": "Euler a",
           "guidance_scale": 7.5,
       },
       "extra": {
           "response_image_type": "jpeg"
       }
   }

   response = requests.request("POST", url, json=payload, headers=headers)

   return response.json()["task_id"]

نتيجة المهمة

تُستخدم هذه الأداة للحصول على حالة مهمة قيد التشغيل على Novita API. تأخذ فقط معرف المهمة. على سبيل المثال، تنشئ أداة الصورة صورة بشكل غير متزامن وتعيد معرف مهمة إلى المضيف. يمكن للمستخدم بعد ذلك أن يطلب من المضيف استرداد نتيجة تلك المهمة باستخدام أداة نتيجة المهمة.

@mcp.tool()
def task_result(task_id: str):
   """
   الحصول على الحالة الحالية لمهمة قيد التشغيل باستخدام معرف المهمة الخاص بها.
   """

   url = base_url + f'/async/task-result?task_id={task_id}'

   response = requests.request("GET", url, headers=headers)
   return response.json()

إنشاء فيديو

تنشئ هذه الأداة مقاطع فيديو باستخدام نقطة نهاية Kling AI V1.6 لتحويل النص إلى فيديو من Novita API. تمامًا مثل أداة الصورة، تنشئ مقاطع الفيديو بشكل غير متزامن وتعيد معرف مهمة.

@mcp.tool()
def generateVideo(prompt: str):
   """
   إنشاء صورة باستخدام استدعاء.
   """

   url = base_url + "/async/kling-v1.6-t2v"

   payload = {
       "mode": "Standard",
       "prompt": prompt,
       "negative_prompt": "low quality",
       "guidance_scale": 0.6
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()

تحويل النص إلى كلام

تنشئ هذه الأداة كلامًا باستخدام نقطة نهاية تحويل النص إلى كلام من Novita API. وهي أيضًا نقطة نهاية غير متزامنة، لذا تعيد معرف مهمة أيضًا.

@mcp.tool()
def textToSpeech(text, voice_id) -> str:
   """
   إنشاء كلام باستخدام نص ومعرف صوت.
   تعيد معرف مهمة الكلام المنشأ.
  
   معرفات الصوت المتاحة هي:
   - Emily
   - James
   - Olivia
   - Michael
   - Sarah
   - John
   """

   url = base_url + "/async/txt2speech"

   payload = {
       "request": {
           "voice_id": voice_id,
           "language": "en-US",
           "texts": [text]
       }
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()["task_id"]

مع تعريف جميع أدواتنا، يمكننا بعد ذلك إعداد آلية النقل الخاصة بنا. سنستخدم stdio.

if __name__ == "__main__":
   # التشغيل باستخدام نقل stdio
   mcp.run(transport="stdio")

يمكننا الآن تجميع كل الكود معًا:

import os
import sys

from mcp.server.fastmcp import FastMCP

import requests

import uvicorn

from starlette.applications import Starlette
from starlette.routing import Mount

base_url = "https://api.novita.ai/v3"

headers = {
   "Content-Type": "application/json",
   "Authorization": f"Bearer {os.environ['NOVITA_API_KEY']}"
}

mcp = FastMCP("Novita_API")

@mcp.tool()
def list_models() -> str:
   """
   سرد جميع النماذج المتاحة من Novita API.
   """

   url = base_url + "/openai/models"

   response = requests.request("GET", url, headers=headers)
   data = response.json()["data"]
  
   text = ""
   for i, model in enumerate(data, start=1):
       text += f"Model id: {model['id']}\n"
       text += f"Model description: {model['description']}\n"
       text += f"Model type: {model['model_type']}\n\n"
  
   return text

@mcp.tool()
def get_model(model_id: str, message) -> str:
   """
   توفير معرف النموذج ورسالة للحصول على رد من Novita API.
   """

   url = base_url + "/openai/chat/completions"

   payload = {
       "model": model_id,
       "messages": [
           {
               "content": message,
               "role": "user",
           }
       ],
       "max_tokens": 200,
       "response_format": {
           "type": "text",
       },
   }

   response = requests.request("POST", url, json=payload, headers=headers)
  
   content = response.json()["choices"][0]["message"]["content"]

   return content

@mcp.tool()
def text2Image(prompt: str) -> str:
   """
   إنشاء صورة من استدعاء نصي باستخدام Novita API.
   """

   url = base_url + "/async/txt2img"

   payload = {
       "request": {
           "model_name": "sd_xl_base_1.0.safetensors",
           "prompt": prompt,
           "width": 1024,
           "height": 1024,
           "image_num": 1,
           "steps": 20,
           "clip_skip": 1,
           "sampler_name": "Euler a",
           "guidance_scale": 7.5,
       },
       "extra": {
           "response_image_type": "jpeg"
       }
   }

   response = requests.request("POST", url, json=payload, headers=headers)

   return response.json()["task_id"]

@mcp.tool()
def task_result(task_id: str) -> str:
   """
   الحصول على الحالة الحالية لمهمة قيد التشغيل باستخدام معرف المهمة الخاص بها.
   """

   url = base_url + f'/async/task-result?task_id={task_id}'

   response = requests.request("GET", url, headers=headers)
   return response.json()

@mcp.tool()
def generateVideo(prompt: str) -> str:
   """
   إنشاء صورة باستخدام استدعاء.
   """

   url = base_url + "/async/kling-v1.6-t2v"

   payload = {
       "mode": "Standard",
       "prompt": prompt,
       "negative_prompt": "low quality",
       "guidance_scale": 0.6
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()["task_id"]

@mcp.tool()
def textToSpeech(text, voice_id) -> str:
   """
   إنشاء كلام باستخدام نص ومعرف صوت.
   تعيد معرف مهمة الكلام المنشأ.
  
   معرفات الصوت المتاحة هي:
   - Emily
   - James
   - Olivia
   - Michael
   - Sarah
   - John
   """

   url = base_url + "/async/txt2speech"

   payload = {
       "request": {
           "voice_id": voice_id,
           "language": "en-US",
           "texts": [text]
       }
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()["task_id"]

if __name__ == "__main__":
   # التشغيل باستخدام نقل stdio
   mcp.run(transport="stdio")

يمكننا اختبار خادم MCP الخاص بنا على أي مضيف MCP، مثل Claude Desktop أو Cursor أو VS Code، باستخدام الإعداد التالي:

{
     "command": "python",
     "args": [
       "path/to/server.py" 
     ],
     "env": {
       "NOVITA_API_KEY": "sk_...."
     }
   }

يمكننا أيضًا استخدام البرنامج النصي للعميل الذي طورناه سابقًا لاختبار الخادم الخاص بنا.

استخدام خادم MCP منخفض المستوى

يستخدم الخادم الذي بنيناه في القسم السابق فئة FastMCP، التي توفر واجهة عالية المستوى لبناء خوادم MCP. يمكنك أيضًا بناء خوادم MCP باستخدام الخادم منخفض المستوى، الذي يمنحك تحكمًا دقيقًا في بروتوكول MCP. دعنا نرى كيف يمكننا تعديل الخادم من القسم السابق لاستخدام الخادم منخفض المستوى.

إدارة الأدوات

أحد الأشياء التي يقوم بها FastMCP هو إدارة الأدوات التي تعرفها. عندما تزين دالة بمُزين الأداة لفئة FastMCP، يضيف FastMCP تلك الأداة إلى قائمة. عندما يطلب العامل تلك الأداة، يقوم FastMCP بجلبها، واستدعائها، ثم إرسال النتيجة مرة أخرى إلى العامل.

مع الخادم منخفض المستوى، يمكننا إدارة هذه العملية بأنفسنا باستخدام مُزين call_tool.

@app.call_tool()
async def manage_tool(name: str, arguments: dict ) -> list[types.TextContent]:
   if name == "list_models":
       return await list_models_tool()
  
   if name == "get_model":
       return await get_model_tool(arguments)
  
   else:
       raise ValueError(f"Unknown tool: {name}")

يظهر الكود أعلاه دالة تم تزيينها بطريقة call_tool. عندما يستدعي العامل أداة، يمرر اسم الأداة وأي وسائط تتوقعها الأداة. باستخدام اسم الدالة، يمكننا بعد ذلك تحديد الأداة التي يرغب العامل في استدعائها وتنفيذها.

نحصل أيضًا على إدارة سرد الأدوات باستخدام الخادم منخفض المستوى، على عكس FastMCP الذي يفعل ذلك تلقائيًا عندما يتم تزيين دالة بطريقة الأداة.

@app.list_tools()
async def list_tools() -> list[types.Tool]:
   return [
       types.Tool(
           name="list_models",
           description="سرد جميع النماذج المتاحة من Novita API.",
           inputSchema={"type": "object", "properties": {}},
       ),
       types.Tool(
           name="get_model",
           description="توفير معرف النموذج ورسالة للحصول على رد من Novita API.",
           inputSchema={
               "type": "object",
               "required": ["model_id", "message"],
               "properties": {
                   "model_id": {
                       "type": "string",
                       "description": "معرف النموذج المراد استخدامه.",
                   },
                   "message": {
                       "type": "string",
                       "description": "رسالة الإدخال المراد إرسالها.",
                   },
               },
           },
       ),
   ]

إليك الكود الكامل للخادم منخفض المستوى. يحتوي على أداتين ويمكنه التواصل مع العميل عبر الإدخال والإخراج القياسيين.

import os
import asyncio
import requests
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

base_url = "https://api.novita.ai/v3"

headers = {
   "Content-Type": "application/json",
   "Authorization": f"Bearer {os.environ['NOVITA_API_KEY']}"
}

app = Server("Novita_API")

async def list_models_tool():
   """
   يسرد جميع النماذج المتاحة من Novita API.
   """
   url = base_url + "/openai/models"
   response = requests.get(url, headers=headers)
   data = response.json()["data"]

   text = ""
   for i, model in enumerate(data, start=1):
       text += f"Model id: {model['id']}\n"
       text += f"Model description: {model['description']}\n"
       text += f"Model type: {model['model_type']}\n\n"

   return [types.TextContent(type="text", text=text)]

async def get_model_tool(arguments: dict):
   """
   بالنظر إلى معرف النموذج ورسالة المستخدم، جلب رد من Novita API.
   """
   model_id = arguments.get("model_id")
   message = arguments.get("message")

   if not model_id or not message:
       raise ValueError("كل من 'model_id' و 'message' مطلوبان.")

   url = base_url + "/openai/chat/completions"

   payload = {
       "model": model_id,
       "messages": [
           {
               "content": message,
               "role": "user",
           }
       ],
       "max_tokens": 200,
       "response_format": {
           "type": "text",
       },
   }

   response = requests.post(url, json=payload, headers=headers)
   content = response.json()["choices"][0]["message"]["content"]
   return [types.TextContent(type="text", text=content)]

@app.call_tool()
async def manage_tool(name: str, arguments: dict ) -> list[types.TextContent]:
   if name == "list_models":
       return await list_models_tool()
  
   if name == "get_model":
       return await get_model_tool(arguments)
  
   else:
       raise ValueError(f"أداة غير معروفة: {name}")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
   return [
       types.Tool(
           name="list_models",
           description="سرد جميع النماذج المتاحة من Novita API.",
           inputSchema={"type": "object", "properties": {}},
       ),
       types.Tool(
           name="get_model",
           description="توفير معرف النموذج ورسالة للحصول على رد من Novita API.",
           inputSchema={
               "type": "object",
               "required": ["model_id", "message"],
               "properties": {
                   "model_id": {
                       "type": "string",
                       "description": "معرف النموذج المراد استخدامه.",
                   },
                   "message": {
                       "type": "string",
                       "description": "رسالة الإدخال المراد إرسالها.",
                   },
               },
           },
       ),
   ]

async def main():
   async with stdio_server() as streams:
       await app.run(streams[0], streams[1], app.create_initialization_options())

if __name__ == "__main__":
   asyncio.run(main())

لا يمنحك الخادم منخفض المستوى تحكمًا دقيقًا في إدارة الأدوات فحسب؛ بل يوفر أيضًا تحكمًا منخفض المستوى في الموارد والاستدعاءات والأجزاء الأخرى من بروتوكول MCP، مثل إدارة دورة الحياة.

الخلاصة

في هذا البرنامج التعليمي، بنينا خادم MCP يستخدم Novita API لتعزيز قدرات أي مضيف MCP. لقد تعلمت أساسيات ما يلزم لإنشاء خادم MCP، بما في ذلك كيفية تعريف الأدوات وكشفها.

باستخدام هذا الأساس، يمكنك البدء في استكشاف موضوعات أكثر تقدمًا مثل المصادقة، ونشر خوادم MCP عن بُعد، والعمل مباشرة مع تنفيذ Python منخفض المستوى.

كما استكشفنا Novita API نفسها ورأينا قدراتها، بدءًا من مجموعة متنوعة من نماذج اللغة إلى أدوات التوليد للفيديو والصوت والصور. قم بزيارة Novita LLM Playground لتجربة واجهات API الأخرى التي لم نتمكن من تغطيتها أيضًا، مثل نقاط نهاية تحرير الصور والوجه.

Novita AI عبارة عن منصة سحابية للذكاء الاصطناعي توفر للمطورين طريقة سهلة لنشر نماذج الذكاء الاصطناعي باستخدام واجهة برمجة التطبيقات البسيطة الخاصة بنا، مع توفير سحابة GPU موثوقة وبأسعار معقولة للبناء والتوسع.