نحن على أعتاب نوع جديد من البرمجيات. في خطابه الرئيسي الأخير في مدرسة YC للشركات الناشئة، وصف أندريه كارباثي البرمجيات 3.0، عالم حيث تصبح اللغة الطبيعية واجهة البرمجة الأساسية، وتعمل نماذج اللغة الكبيرة (LLMs) كمحركات حسابية وشركاء مستقلين.
على عكس البرمجيات 1.0، حيث كان المطورون يكتبون تعليمات صريحة، أو البرمجيات 2.0، حيث تتعلم الشبكات العصبية الأنماط من البيانات، تعمل البرمجيات 3.0 على اللغة الطبيعية. تصبح مطالباتنا برامج، وتترجم نماذج اللغة الكبيرة نوايانا إلى سلوك قابل للتنفيذ.
في هذا البرنامج التعليمي، سنقوم ببناء خادم MCP يوفر وصولاً بعيداً إلى صندوق رمل Novita AI حيث يمكن لوكيل الذكاء الاصطناعي الخاص بنا تنفيذ الكود بأمان. سيتم إنشاء الوكيل نفسه باستخدام مكتبة mcp-use، التي تدير أيضاً الاتصال بخادم MCP تلقائياً.
ما هو بروتوكول سياق النموذج (MCP)؟
بروتوكول سياق النموذج (MCP) هو معيار مفتوح طورته Anthropic لتتمكن نماذج الذكاء الاصطناعي من التواصل مع الخدمات والأدوات ومصادر البيانات الخارجية. فكر فيه كـ “USB-C للذكاء الاصطناعي”؛ فهو يوفر واجهة قياسية بين الوكلاء والعالم الخارجي.
المكونات الأساسية الثلاثة:
- الأدوات: دوال / واجهات برمجة تطبيقات قابلة للاستدعاء يقدمها الخادم (مثل تصفح الويب، تنفيذ الكود).
- الموارد: بيانات خارجية يقدمها الخادم، عادة ما تكون كسياق للوكيل الذكي (مثل الملفات، البيانات الوصفية، مجموعات البيانات).
- الأوامر: هذه التعليمات التفصيلية تهدف إلى توجيه سلوك الوكيل عند التفاعل مع الأدوات أو الموارد المذكورة أعلاه.
للحصول على نظرة أعمق على MCP وهندسته ومكوناته، راجع مدونتنا حول بناء خادم MCP الأول الخاص بك
لماذا يعتبر MCP مهماً لتكامل أدوات الذكاء الاصطناعي
ينشئ MCP اتصالاً ثنائي الاتجاه بين وكلاء الذكاء الاصطناعي والأنظمة الخارجية. هذا يسهل “توصيل” نماذج اللغة الكبيرة بأدوات مخصصة مع تقليل عمل التكامل وتقليل احتمالية الأخطاء.
مقدمة لمكتبة mcp-use
مكتبة mcp-use هي حزمة بايثون تبسط بناء وكلاء الذكاء الاصطناعي الذين يتفاعلون مع خوادم MCP. فهي تتولى إنشاء الوكيل وتدير الاتصال بخوادم MCP للوصول إلى الأدوات ومصادر البيانات الخارجية، مما يسمح لك بالتركيز فقط على منطق تطبيقك.
نظرة عامة على Novita AI Sandbox وواجهة برمجة تطبيقات النماذج
ما هو الصندوق الرمل ولماذا هو مهم
الصندوق الرمل هو بيئة تشغيل آمنة ومنعزلة يمكن فيها تنفيذ كود غير موثوق دون التأثير على نظام المضيف. هو في الأساس حاسوب افتراضي خفيف الوزن يمكن لوكيل الذكاء الاصطناعي الخاص بك من تنفيذ الكود والأوامر وإنشاء الملفات وما إلى ذلك.
تقدم Novita AI هذا الصندوق الرمل على السحابة ليتمكن وكيلك من الوصول إليه بسرعة عند الطلب، مع فواتير مرنة لكل ثانية بناءً على الموارد المستخدمة.
الميزات الرئيسية لصندوق رمل Novita:
- عزل آمن: يحصل كل صندوق رمل على نظام ملفات وبيئة منعزلة خاصة به، مما يحمي البيانات ويمنع التفاعلات غير المقصودة.
- بدء سريع: تطلق نسخ الصندوق الرمل في أقل من 200 مللي ثانية في المتوسط، مما يجعله مثالياً للسيناريوهات منخفضة التأخير.
- دعم متعدد اللغات: يمكنك تشغيل الكود بلغات برمجة متعددة، بما في ذلك بايثون وجافا سكريبت و تايب سكريبت وغيرها.
- إيقاف مؤقت واستئناف سريع: يمكنك إيقاف الصندوق الرمل مؤقتاً في أي وقت واستئنافه عند الحاجة، مع استعادة كاملة لحالة نظام الملفات والعمليات.
- تنفيذ في الخلفية: يدعم تنفيذ المهام في الخلفية وهو مناسب للسيناريوهات التي تتطلب انتظار نتيجة.
واجهة برمجة تطبيقات نماذج Novita:
تقدم Novita مكتبة ضخمة من نماذج الذكاء الاصطناعي مفتوحة المصدر من مختبرات أبحاث رائدة مثل OpenAI و Google و DeepSeek و Qwen وغيرها. تشمل هذه النماذج نماذج للغة والرؤية والصوت والفيديو والتضمين. نماذج اللغة الخاصة بنا متوافقة أيضاً بالكامل مع حزمة تطوير برامج OpenAI، لذا فإن التبديل من OpenAI إلى Novita يتطلب فقط تحديث عنوان URL الأساسي ومفتاح API في العميل الخاص بك، ثم تحديد نموذج Novita.
| from openai import OpenAI client = OpenAI( base_url=“https://api.novita.ai/v3/openai”, api_key=“ ) |
إعداد بيئة التطوير الخاصة بك
للبدء، سنقوم باستنساخ مستودع GitHub، وإعداد بيئة بايثون نظيفة، وتثبيت جميع التبعيات المطلوبة، والحصول على مفاتيح Novita AI.
استنساخ مستودع GitHub وتثبيت التبعيات باستخدام uv.
1. قم بتثبيت uv (مدير حزم بايثون خفيف الوزن)
| pip install uv |
2. استنسخ المستودع (مستودع GitHub) وانتقل إليه.
| git clone https://github.com/Studio1HQ/mcp\_remote\_execution.git cd mcp_remote_execution |
3. إنشاء وتفعيل البيئة الافتراضية لـ uv
| # Creates a virtual environment uv venv # Activate the virtual environment source .venv/bin/activate # For Mac/Linux # or .venv\Scripts\activate # For Windows |
4. تثبيت تبعيات المشروع
| # Install dependencies uv sync |
إنشاء حساب Novita AI والحصول على مفتاح API
1. سجل في novita.ai.
2. في لوحة التحكم، مرر مؤشر الفأرة فوق أيقونة ملف تعريف المستخدم وانقر على مفاتيح API في النافذة المنبثقة.
3. في صفحة إدارة المفاتيح، انقر على إضافة مفتاح جديد. في النافذة المنبثقة، أدخل اسماً لمفتاحك، وانقر على تأكيد، ثم انسخ المفتاح الذي تم إنشاؤه.
4. الآن داخل دليل المشروع، قم بإنشاء ملف .env، والصق ما يلي أدناه.
| NOVITA_API_KEY=“ NOVITA_BASE_URL=“https://api.novita.ai/v3/openai” NOVITA_E2B_DOMAIN=“sandbox.novita.ai” NOVITA_E2B_TEMPLATE=“code-interpreter-v1” |
إضافة رصيد إلى حساب Novita AI.
لاستخدام صندوق رمل Novita، تحتاج إلى إضافة رصيد إلى حسابك. في علامة تبويب لوحة التحكم، انقر على ‘الفواتير’. ثم، في صفحة الفواتير، أضف طريقة دفع وقم بشحن رصيد لا يقل عن 10 دولارات.
بناء خادم MCP وتكامل الوكيل الذكي
الآن بعد أن تم إعداد بيئتنا، لنبدأ البناء. ستتكون هذه العملية من جزأين: أولاً، سنمر بإنشاء خادم MCP، ثم سنقوم ببناء تطبيق الوكيل الذكي (الذي يعمل كعميل MCP). ولكن قبل أن نبدأ، لننشئ مدير الصندوق الرمل الذي سوف يستخدمه خادم MCP.
مدير الصندوق الرمل
سيكون هذا المسؤول عن بدء وإيقاف نسخ الصندوق الرمل. في إعدادنا، سنقيد الخادم بنسخة واحدة من الصندوق الرمل في كل مرة. سيتولى مدير الصندوق الرمل أيضاً تنفيذ كود بايثون وتشغيل أوامر الشل في الصندوق الرمل.
أولاً، في ملف sandbox_manager.py، لدينا فئة SandboxManager، التي تأخذ المعلمات التالية:
- sandbox_template: القالب المستخدم لإنشاء نسخة من الصندوق الرمل. سنستخدم “code-interpreter-v1”، الذي يأتي مثبتاً مسبقاً مع حزم بايثون المستخدمة بشكل شائع (مثل pandas و numpy).
- sandbox_domain: نقطة نهاية النطاق المستخدمة للاتصال بنسخة صندوق رمل Novita AI.
- sandbox_timeout: المدة (بالثواني) التي تحدد مدة بقاء الصندوق الرمل نشطاً قبل إيقافه تلقائياً.
| from novita_sandbox.code_interpreter import Sandbox class SandboxManager: def __init__( self, sandbox_template: str, sandbox_domain: str, sandbox_timeout: int, ): self.sandbox_template = sandbox_template self.sandbox_domain = sandbox_domain self.sandbox_timeout = sandbox_timeout |
الآن سنضيف طرقاً لإنشاء وإيقاف الصندوق الرمل.
- للإنشاء، نأخذ sandbox_api_key، ونستخدمه لبدء نسخة جديدة من الصندوق الرمل للمستخدم، ونرجع رسالة نجاح مع معرف الصندوق الرمل أو رسالة استثناء إذا حدث خطأ ما.
- للإيقاف، نأخذ كل من sandbox_api_key و sandbox_id، نتصل بالصندوق الرمل، ونوقفه إذا كان موجوداً. كما كان من قبل، نرجع رسالة نجاح أو رسالة استثناء اعتماداً على النتيجة.
| … class SandboxManager: … # below existing code def create_sandbox_session(self, sandbox_api_key: str) -> str: “”“ This will create a new sandbox instance. Args: sandbox_api_key (str): The API key for the sandbox. Returns: str: Success message with the sandbox ID of the new sandbox or error message. ”“” try: # create the new sandbox sandbox = Sandbox.create( template=self.sandbox_template, api_key=sandbox_api_key, domain=self.sandbox_domain, timeout=self.sandbox_timeout, ) return f"Successfully created sandbox. Sandbox ID: {sandbox.sandbox_id}“ except Exception as e: return f"Failed to create new sandbox:{str(e)}” def stop_sandbox_session(self, sandbox_api_key: str, sandbox_id: str) -> str: “”“ This will kill a sandbox instance if it exists. Args: sandbox_api_key (str): The API key for the sandbox. sandbox_id (str): The ID of the sandbox. Returns: str: Success message with the sandbox ID of the killed sandbox or error message. ”“” try: # connect to sandbox sandbox = Sandbox.connect( api_key=sandbox_api_key, sandbox_id=sandbox_id, ) sandbox.kill() return f"Successfully killed Sandbox ID: {sandbox_id}“ except Exception as e: return f"Failed to kill Sandbox ID: {sandbox_id}\ {str(e)}” |
أخيراً، نضيف طرقاً لتشغيل كود بايثون وأوامر الشل في صندوق رمل المستخدم بعد الاتصال به عبر مفتاح API والمعرف. يتم إرجاع جميع مخرجات الصندوق الرمل (بما في ذلك الاستثناءات والأخطاء) في قاموس.
| … class SandboxManager: … # below existing code def run_python_code( self, python_code: str, sandbox_api_key: str, sandbox_id: str ) -> dict: “”“ Runs the python code on the sandbox, and if there any image outputs they are skipped. Args: python_code (str): The python code to run. sandbox_api_key (str): The API key for the sandbox. sandbox_id (str): The ID of the sandbox. Returns: dict: Containing stdout, logs, error, etc. ”“” try: # connect to sandbox sandbox = Sandbox.connect( api_key=sandbox_api_key, sandbox_id=sandbox_id, ) execution = sandbox.run_code(python_code, language=“python”) return { # we will skip image outputs. “outputs”: [result for result in execution.results if not result.png], “logs”: execution.logs, “error”: execution.error, } except Exception as e: return {“error”: str(e)} def run_on_command_line( self, command: str, sandbox_api_key: str, sandbox_id: str ) -> dict: “”“ Runs the command on the sandbox. Args: command (str): The command to run. sandbox_api_key (str): The API key for the sandbox. sandbox_id (str): The ID of the sandbox. Returns: dict: Containing the output of the command and the execution error if any. ”“” try: # connect to sandbox sandbox = Sandbox.connect( api_key=sandbox_api_key, sandbox_id=sandbox_id, ) result = sandbox.commands.run(command) return { “output”: { “stdout”: result.stdout, “stderr”: result.stderr, “exit_code”: result.exit_code, “error”: result.error, }, “execution error”: None, } except Exception as e: return {“output”: None, “execution error”: str(e)} |
إنشاء خادم MCP
الآن بعد أن تم إعداد مدير الصندوق الرمل، حان الوقت للعمل على ملف mcp_server.py. أولاً، نقوم بإنشاء نسخة من FastMCP، الإطار الذي سيشغل الخادم، إلى جانب مدير الصندوق الرمل. نقوم أيضاً بإنشاء نسخة من وحدة التحكم الغنية لطباعة جميلة على الطرفية.
ملاحظة: مهلة انتهاء صلاحية الصندوق الرمل التي تم تمريرها إلى مدير الصندوق الرمل تستخدم لكل صندوق رمل يبدأ على هذا الخادم. هي المدة القصوى (بالثواني) التي يمكن أن يبقى فيها كل صندوق رمل نشطاً (يعمل)، ما لم يتم إيقافه في وقت سابق.
| import asyncio import os from dotenv import load_dotenv from mcp.server.fastmcp import Context, FastMCP from rich.console import Console from rich.panel import Panel from rich.table import Table from starlette.requests import Request from sandbox_manager import SandboxManager # load .env variables load_dotenv() console = Console() # Initialize FastMCP server mcp = FastMCP(“MCP_Server”) # Initialize sandbox manager for the singleton sandbox instance. sandbox_manager = SandboxManager( sandbox_template=os.getenv(“NOVITA_E2B_TEMPLATE”), sandbox_domain=os.getenv(“NOVITA_E2B_DOMAIN”), sandbox_timeout=900, # 900 seconds (15 minutes), sandbox instance will be killed automatically after. ) |
الآن، نظراً لأننا نحتاج إلى مفتاح API للاتصال بصندوق رمل المستخدم، سنحصل عليه من طلب المستخدم إلى الخادم. كما سنرى لاحقاً في عميل MCP، سيتم إرسال هذا المفتاح في رأس التفويض لكل طلب. هذا يعني أننا نحتاج إلى طريقة لاستخراج مفتاح API من هذا الرأس على جانب الخادم.
يوفر FastMCP كائناً Context يحتفظ في وقت التشغيل بمعلومات عن المستخدم الذي يقدم الطلب. لذا سنقوم بإنشاء طريقة مساعدة get_user_api_key تأخذ كائن السياق، تستخرج مفتاح API من الرأس، وترجعه أو تثير استثناءً إذا كان مفقوداً.
| … # below existing code def get_user_api_key(ctx: Context) -> str: “”“ Returns the API key from the request header if it exists otherwise raise an exception. ”“” request: Request = ctx.request_context.request # Access request data auth_header = request.headers.get(“Authorization”) if auth_header: auth_header = auth_header.split(" ")[1] if not auth_header: raise Exception(“Missing API Key in Authorization Bearer header”) return auth_header |
حان الوقت لعرض الأوامر والأدوات والموارد على خادمنا. للقيام بذلك، نضيف ببساطة المزخرف @mcp.{prompt, tool, resource}() إلى دالة بايثون. نبدأ بأمرنا، الذي يعيد تعليمات حول كيفية استخدام وكيلنا للصندوق الرمل.
| … # below existing code (Note: Console display helper methods were skipped for brevity) @mcp.prompt() def instructions_for_sandbox_use() -> str: “”“ RETURNS MUST READ INSTRUCTIONS FOR SANDBOX USE. ”“” return “”“ When you want to use the sandbox function, you must first create a new sandbox session by calling the create_sandbox_session() function. Then you can use the run_python_code() or run_on_command_line() function to run on the sandbox. When you are done, you must kill the sandbox session by calling the stop_sandbox_session() function. Note: - The sandbox already comes pre-installed with the usual data analysis packages but if there’s a package you are not sure exists or your code had an import error due to a missing package, you can check if it’s installed and if not install it. ”“” |
بالنسبة للأدوات، نعرض طرقاً تستخدم نسخة sandbox_manager لإنشاء وإيقاف وتنفيذ الكود أو الأوامر على صندوق رمل المستخدم.
ملاحظة: من بين الوسائط التي تأخذها طرق الأدوات أدناه، سنقوم بتضمين ctx: Context. يخبر هذا FastMCP بإدراج سياق الطلب تلقائياً في تلك المعلمة، وهي عملية تعرف باسم الحقن التابع. ثم نحصل على مفتاح api إذا كان موجوداً عن طريق تمرير هذا السياق إلى طريقة المساعدة الخاصة بنا (get_user_api_key).
| … # below existing code @mcp.tool() def create_sandbox_session(ctx: Context) -> str: “”“ This will create a sandbox instance and return success message with the sandbox id or error message. ”“” try: return sandbox_manager.create_sandbox_session(get_user_api_key(ctx)) except Exception as e: return e @mcp.tool() def stop_sandbox_session(sandbox_id: str, ctx: Context) -> str: “”“ This will kill a sandbox instance if it exists. ”“” try: return sandbox_manager.stop_sandbox_session(get_user_api_key(ctx), sandbox_id) except Exception as e: return e @mcp.tool() def run_python_code(python_code: str, sandbox_id: str, ctx: Context) -> dict: “”“ Runs the python code on the sandbox, and if there any image outputs they are skipped. Args: python_code (str): The python code to run. sandbox_id (str): The ID of the sandbox. Note: The ctx (Context) is a dependency injection object that is automatically passed. Returns: dict: Containing stdout, logs, error, etc. ”“” console.print( Panel( python_code, title=“Agent Executing Python Code”, border_style=“blue”, ) ) try: result = sandbox_manager.run_python_code( python_code, get_user_api_key(ctx), sandbox_id ) # display the result. Note: only do this in test not in prod. display_sandbox_code_output(result) return result except Exception as e: return {“error”: str(e)} @mcp.tool() def run_on_command_line(command_line: str, sandbox_id: str, ctx: Context) -> dict: “”“ Runs the command on the sandbox. Args: command_line (str): The command to run. sandbox_id (str): The ID of the sandbox. Note: The ctx (Context) is a dependency injection object that is automatically passed. Returns: dict: Containing the output of the command and the execution error if any. ”“” console.print( Panel( command_line, title=“Agent Executing Command Line”, border_style=“blue”, ) ) try: result = sandbox_manager.run_on_command_line( command_line, get_user_api_key(ctx), sandbox_id ) # display the result. Note: only do this in test not in prod. display_sandbox_command_output(result) return result except Exception as e: return {“execution error”: str(e)} |
الآن سنعرض مورداً واحداً لأحد سيناريوهات العرض التوضيحي الخاصة بنا، والتي سنراها لاحقاً. يعيد هذا المورد محاكاة لمحفظة أسهم المستخدم. على عكس الأوامر والأدوات، تتطلب الموارد تحديد عنوان URL حيث ستكون متاحة.
| … # below existing code @mcp.resource(“data://user_stock_portfolio”) def get_user_portfolio() -> dict: “”“ Returns the user’s portfolio holdings across major index ETFs and individual stocks. Returns: dict: Portfolio with ticker symbols, quantities, and average purchase prices ”“” portfolio = { “holdings”: [ # Major Index ETFs { “ticker”: “SPY”, “name”: “SPDR S&P 500 ETF”, “quantity”: 4, “avg_purchase_price”: 670.13, “asset_type”: “ETF”, }, … # skipped for brevity ] } return portfolio |
أخيراً، نضيف الكود الذي يبدأ خادمنا. لن نستخدم نقل MCP stdio لأن وحدة التحكم الغنية تطبع إلى الطرفية، وهي عملية حظرية ستتداخل. بدلاً من ذلك، سنستخدم نقل MCP streamable-http (هذا ما ستستخدمه أيضاً في الإنتاج، لأنه الأفضل للاتصال عن بعد بين العميل والخادم عبر HTTP).
| … # below existing code if __name__ == “__main__”: # run the server # Note: We use streamable-http as the transport protocol instead of stdio because we are printing to the console which would block stdio. # Also in production you should use SSE or streamable-http rather than stdio. asyncio.run(mcp.run(transport=“streamable-http”)) |
بفف، لنبدأ خادم MCP الخاص بنا حتى نتمكن من الحصول على عنوان URL الذي يعمل عليه، والذي سنحتاجه لتوصيل وكيل الذكاء الاصطناعي الخاص بنا. قم بتشغيل الأمر أدناه في الطرفية.
| uv run mcp_server.py |
يجب أن ترى الخادم يعمل الآن. قم بتدوين عنوان URL الذي يعمل عليه
تكامل الوكيل الذكي مع خادم MCP
الآن سنبدأ العمل على الوكيل الذكي في ملف mcp_client.py. تجعل مكتبة mcp-use هذه العملية سهلة. أولاً، نقوم بتعيين مستوى التصحيح إلى INFO لمعرفة ما يفعله الوكيل. في الطريقة الرئيسية، نقوم بإنشاء قاموس تكوين لعميل mcp. يحدد خوادم mcp المتاحة، باستخدام اسم (استخدمت “stock&sandbox”) وعنوان URL الذي يعمل عليه خادم mcp (تذكر إضافة /mcp)، كما ندرج مفتاح API للمستخدم كقيمة لـ “auth” سيتم إدراج هذا تلقائياً بواسطة mcp-use في رأس حامل التفويض لكل طلب.
بما أن mcp-use يعتمد على langchain-openai، نمرر عنوان URL الأساسي لـ novita ومفتاح api واسم نموذج LLM، وهذا يعمل لأن Novita متوافق مع OpenAI.
وبالإضافة إلى استجابة الوكيل، نود تضمين معرف الصندوق الرمل الذي استخدمه على الخادم، إن وجد (سنشرح السبب لاحقاً). للقيام بذلك، سنقوم بتعريف فئة Pydantic لتمثيل تنسيق الاستجابة الخاص بنا.
| import asyncio import os from datetime import datetime from typing import Optional import mcp_use from dotenv import load_dotenv from langchain_openai import ChatOpenAI from mcp_use import MCPAgent, MCPClient from pydantic import BaseModel, Field from rich.console import Console from rich.panel import Panel from rich.prompt import Prompt # Load environment variables load_dotenv() console = Console() # Note: 1 for INFO level, 2 for full verbose DEBUG level and 0 for NO debug output. mcp_use.set_debug(1) class ResponseFormat(BaseModel): response: str id_of_used_sandbox: Optional[str] = Field( …, description=“The id of the sandbox used if any” ) async def main(model: str, base_url: str, api_key: str): # Create configuration dictionary config = { “mcpServers”: { “stock&sandbox”: { # If the url the mcp server is running at is different replace below, # also remember to add /mcp. “url”: “http://127.0.0.1:8000/mcp”, “auth”: api_key, } } } # Create MCPClient from configuration dictionary client = MCPClient(config) # Create LLM llm = ChatOpenAI(model=model, base_url=base_url, api_key=api_key) … |
ثم نقوم بإنشاء وكيل MCP عن طريق تمرير LLM وعميل MCP والحد الأقصى للخطوات (الذي يحدد الإجراءات التي يمكن للوكيل اتخاذها قبل الاستجابة)، وتفعيل الذاكرة حتى تتمكن mcp-use من التعامل مع سجل المحادثات الخاص بنا. نوفر أيضاً أمر نظام يتضمن التاريخ والوقت الحاليين، إلى جانب تعليمات مخصصة لاستخدام الصندوق الرمل (أضفت هذا لأن بعض النماذج تنسى قراءة أوامر التعليمات المعروضة على الخادم).
أخيراً، نقوم بإعداد حلقة المحادثة القياسية: الحصول على إدخال المستخدم، تمريره إلى agent.run()، وطباعة الاستجابة. إذا تم استخدام صندوق رمل، نقوم بعد ذلك بإنشاء جلسة لاستدعاء طريقة الإيقاف يدوياً على الخادم كإجراء سلامة، في حال نسخ النموذج إغلاقه.
| … async def main(model: str, base_url: str, api_key: str): … # below existing code # Create agent with the client agent = MCPAgent( llm=llm, client=client, max_steps=25, memory_enabled=True, # mcp-use will auto handle the conversation history. system_prompt=f"“” You are a helpful assistant and the current date is {datetime.now().strftime(‘%Y-%m-%d’)} MUST REMEMBER: - Before any tool call first call instructions_for_sandbox_use() so you can read them. - Ensure you call stop_sandbox_session() after using the sandbox before responding to the user. “”“, ) console.print( Panel( ”[bold green]MCP Session Started[/bold green]\ Type ‘quit()’ to exit.“, title=“MCP Session”, border_style=“green”, ) ) while True: user_input = Prompt.ask(”\ [bold yellow]>>> User Message[/bold yellow]“) if user_input.lower().strip() == “quit()”: break # Pass the query to the agent and await the response. response_obj = await agent.run(user_input, output_schema=ResponseFormat) console.print( f”\ [bold green]>>> Assistant Response: {response_obj.response} [/]" ) if response_obj.id_of_used_sandbox: # Will trigger closure of sandbox on MCP server if it sill active. session = await client.create_session(“stock&sandbox”) await session.call_tool( name=“stop_sandbox_session”, arguments={“sandbox_id”: response_obj.id_of_used_sandbox}, ) await session.disconnect() |
أخيراً، نضيف الكود الذي يبدأ العميل:
| … # below existing code if __name__ == “__main__”: asyncio.run( main( model=“qwen/qwen3-coder-480b-a35b-instruct”, base_url=os.getenv(“NOVITA_BASE_URL”), api_key=os.getenv(“NOVITA_API_KEY”), ) ) |
اختبار تشغيل وكيل الذكاء الاصطناعي MCP الخاص بنا:
يمكنك بدء تطبيق الوكيل الذكي بتشغيل الأمر أدناه في الطرفية
| uv run mcp_client.py |
أدناه روابط فيديو لعروض توضيحية للأوامر التالية:
- موجه المستخدم 1: “لدي 2000 دولار. احصل على أداء المؤشرات الأمريكية الرئيسية من yfinance على مدى الأشهر الستة الماضية وقم بتشغيل نماذج التعلم الآلي للتنبؤ بكيفية توزيع هذا الاستثمار لتعظيم العوائد المحتملة على مدى الشهرين القادمين.”(أضف الرابط هنا)
- موجه المستخدم 2: “قم بتشغيل محاكاة متعددة لانهيار اقتصادي انكماشي في الولايات المتحدة، اختر الأكثر احتمالاً، واشرح التأثير الذي سيحدثه على محفظة الأسهم الخاصة بي.”(أضف الرابط هنا)
نصائح لخوادم MCP في الإنتاج
بينما ركز هذا البرنامج التعليمي على بناء خادم MCP عاملاً مع Novita Sandbox، فإن النشر في الإنتاج يتطلب اعتبارات إضافية:
استخدم النقل المناسب: بينما يعمل “stdio” للتطوير المحلي، يجب أن تستخدم خوادم MCP في الإنتاج “streamable-http” لتمكين الاتصالات عن بعد، تماماً كما فعلنا أعلاه.
قم بتنفيذ المصادقة: كما فعلنا أعلاه، تأكد من تأمين نقاط نهاية خادم mcp الخاص بك بالمصادقة. تأكد من أن كل عميل لديه إذن فقط للوصول إلى الأدوات والموارد التي يحتاجها. يمكنك الاطلاع على المزيد من طرق المصادقة في FastMCP authentication
تفعيل التسجيل: استخدم مسجلاً لمراقبة نشاط الخادم، وتصحيح الأخطاء، وتتبع أنماط الاستخدام. هذا أمر حيوي للصيانة واستكشاف الأخطاء وإصلاحها.
تحديد معدل الطلبات والحصص: احمِ خادمك من الاستخدام السيء عن طريق تنفيذ حدود لمعدل الطلبات وحصص. هذا مهم بشكل خاص عند كشف أدوات كثيفة الموارد.
التوثيق وإدارة الإصدارات: حافظ على وثائق واضحة لواجهة برمجة تطبيقات خادم MCP الخاص بك وإدارة الإصدارات لتسهيل التكامل الأسهل للمطورين و LLMs.
الخاتمة
بفف، يمكنك أخيراً بناء خادم MCP حيث يمكن لوكلاء الذكاء الاصطناعي تنفيذ الكود عن بعد من خلال تعليمات اللغة الطبيعية، وهو تطبيق عملي لبرنامج 3.0 قيد التشغيل.
في هذا البرنامج التعليمي، تعلمت كيفية بناء خادم MCP مع قدرة تنفيذ الكود، وإدارة دورات حياة الصندوق الرمل، وإنشاء وكيل ذكي باستخدام mcp-use يتصل بخادمك. بعد ذلك، جرب توسيعه بإضافة الوصول إلى قاعدة البيانات أو البحث على الويب أو ربط خوادم متعددة بوكيل واحد. توجه إلى Novita، لدينا الأدوات التي تحتاجها لبناء وكلاء الذكاء الاصطناعي الخاصة بك.
Novita AI هي منصة سحابة للذكاء الاصطناعي تقدم للمطورين طريقة سهلة لنشر نماذج الذكاء الاصطناعي باستخدام واجهة برمجة التطبيقات البسيطة الخاصة بنا، مع توفير سحابة GPU بأسعار معقولة وموثوقة للبناء والتوسع.
