كيفية ربط OpenAI Agent SDK بخوادم MCP

يوفر SDK لوكلاء OpenAI طريقة قوية لبناء تطبيقات الذكاء الاصطناعي مع قدرات الوكلاء. واحدة من ميزاته الرئيسية هي دعم بروتوكول سياق النموذج (MCP)، الذي يعطي معيارًا لكيفية تقديم التطبيقات للسياق والأدوات لنماذج اللغة الكبيرة (LLMs). ستوجهك هذه المقالة عبر ربط SDK لوكلاء OpenAI بخوادم MCP، مع خطوات تفصيلية وعينات من التعليمات البرمجية.

فهم بروتوكول سياق النموذج (MCP)

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

أنواع خوادم MCP

تحدد مواصفات MCP نوعين من الخوادم بناءً على آلية النقل الخاصة بها:

1. خوادم Stdio: تعمل هذه كعملية فرعية من تطبيقك، تعمل بشكل أساسي "محليًا".
2. خوادم SSE: تعمل هذه عن بعد، وتتصل بها عبر عنوان URL باستخدام HTTP عبر أحداث الخادم المرسل (SSE).

يوفر SDK لوكلاء OpenAI فئات تتوافق مع كلا النوعين من الخوادم:

• MCPServerStdio لخوادم stdio المحلية
• MCPServerSse لخوادم SSE البعيدة

المتطلبات المسبقة

قبل أن تبدأ، تأكد من أن لديك:

Python 3.10 أو أعلى مثبت

تم تثبيت SDK لوكلاء OpenAI:

pip install openai-agents

لاستخدام خوادم stdio المحلية، قد تحتاج إلى أدوات إضافية مثل npx (لخوادم MCP المستندة إلى JavaScript)

الربط مع خادم MCP Stdio

دعنا نبدأ بالربط مع خادم MCP stdio محلي. سنستخدم خادم MCP لنظام الملفات كمثال.

الخطوة 1: تثبيت الاعتماديات المطلوبة

إذا كنت تخطط لاستخدام خادم MCP لنظام الملفات، ستحتاج إلى Node.js و NPX:

# تثبيت Node.js (إذا لم يكن مثبتًا مسبقًا)
# لأوبنتو/ديبيان
sudo apt update
sudo apt install nodejs npm

# لنظام macOS
brew install node

# تحقق من التثبيت
node --version
npx --version

الخطوة 2: إعداد بنية مشروعك

أنشئ بنية مشروع أساسية:

my_agent_project/
├── sample_data/
│   ├── file1.txt
│   └── file2.md
├── main.py
└── README.md

الخطوة 3: الاتصال بخادم MCP Stdio

إليك مثال كامل للاتصال بخادم MCP لنظام الملفات باستخدام MCPServerStdio:

import asyncio
import os
import shutil
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerStdio

async def run_agent_with_mcp():
    # احصل على مسار الدليل لبيانات العينة
    current_dir = os.path.dirname(os.path.abspath(__file__))
    sample_data_dir = os.path.join(current_dir, "sample_data")

    # إنشاء وتكوين خادم MCP
    async with MCPServerStdio(
        name="خادم نظام الملفات",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
        },
    ) as mcp_server:
        # إنشاء وكيل يستخدم خادم MCP
        agent = Agent(
            name="مساعد الملفات",
            instructions="استخدم أدوات نظام الملفات لقراءة وتحليل الملفات في دليل sample_data.",
            mcp_servers=[mcp_server]
        )

        # تنفيذ الوكيل مع استعلام من المستخدم
        message = "ضم جميع الملفات في الدليل وقم بتلخيص محتوياتها."
        print(f"تنفيذ الاستعلام: {message}\\\\n")

        # توليد معرف التتبع لتصحيح الأخطاء
        trace_id = gen_trace_id()
        with trace(workflow_name="مثال نظام الملفات MCP"، trace_id=trace_id):
            print(f"عرض التتبع: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

if __name__ == "__main__":
    # تحقق مما إذا كان npx مثبتًا
    if not shutil.which("npx"):
        raise RuntimeError("npx غير مثبت. يرجى تثبيته باستخدام `npm install -g npx`.")

    asyncio.run(run_agent_with_mcp())

في هذا المثال:

1. نقوم بإنشاء مثيل MCPServerStdio الذي يقوم بتشغيل خادم MCP لنظام الملفات كعملية فرعية.
2. نمرر هذا الخادم إلى منشئ Agent عبر معلمة mcp_servers.
3. عندما يتم تشغيل الوكيل، فإنه يستدعي تلقائيًا list_tools() على خادم MCP لجعل LLM على دراية بالأدوات المتاحة.
4. إذا قرر LLM استخدام أي من أدوات MCP، فإن SDK يستدعي call_tool() على الخادم.

الاتصال بخادم MCP SSE

لننظر الآن في كيفية الاتصال بخادم MCP بعيد باستخدام SSE:

الخطوة 1: فهم خوادم SSE MCP

تعمل خوادم SSE (أحداث الخادم المرسلة) MCP عن بعد وتعرض وظيفتها عبر واجهات HTTP. على عكس خوادم stdio، فإنها لا تعمل كعمليات فرعية لتطبيقك.

الخطوة 2: الاتصال بخادم MCP SSE

إليك رمز عينة للاتصال بخادم MCP SSE:

import asyncio
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerSse
from agents.model_settings import ModelSettings

async def run_agent_with_remote_mcp():
    # إنشاء وتكوين الاتصال بخادم SSE MCP
    async with MCPServerSse(
        name="خدمة الطقس",
        params={
            "url": "<https://example.com/mcp/sse>",
            # معلمات المصادقة الاختيارية
            "headers": {
                "Authorization": "Bearer your_api_key_here"
            }
        },
    ) as mcp_server:
        # إنشاء وكيل باستخدام خادم MCP البعيد
        agent = Agent(
            name="مساعد الطقس",
            instructions="استخدم أدوات الطقس للإجابة على أسئلة حول الظروف الجوية الحالية.",
            mcp_servers=[mcp_server],
            # إجبار الوكيل على استخدام الأدوات عند توفرها
            model_settings=ModelSettings(tool_choice="required")
        )

        # تنفيذ الوكيل مع استعلام من المستخدم
        message = "كيف هي حالة الطقس في طوكيو اليوم؟"
        print(f"تنفيذ الاستعلام: {message}\\\\n")

        trace_id = gen_trace_id()
        with trace(workflow_name="مثال لاستخدام MCP للطقس"، trace_id=trace_id):
            print(f"عرض التتبع: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

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

إنشاء خادم MCP SSE محلي بسيط

لفهم كيفية عمل MCP بشكل كامل، يساعد تنفيذ خادم MCP بسيط. إليك مثال لخادم MCP SSE بسيط:

import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional

app = FastAPI()

# تحديد الأدوات التي ستوفرها خادم MCP الخاص بنا
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "add",
            "description": "أضف عددين معًا",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "العدد الأول"},
                    "b": {"type": "number", "description": "العدد الثاني"}
                },
                "required": ["a", "b"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "احصل على الطقس لموقع معين",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "اسم المدينة"}
                },
                "required": ["location"]
            }
        }
    }
]

# تنفيذ وظيفة الأداة الفعلية
async def call_tool(tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
    if tool_name == "add":
        return {"result": parameters["a"] + parameters["b"]}

    elif tool_name == "get_weather":
        # في التنفيذ الحقيقي، ستتصل بواجهة برمجة تطبيقات الطقس الفعلية
        weather_data = {
            "Tokyo": {"condition": "مشمس", "temperature": 25},
            "New York": {"condition": "غائم", "temperature": 18},
            "London": {"condition": "ممطر", "temperature": 15}
        }
        location = parameters["location"]
        if location in weather_data:
            return {"weather": weather_data[location]}
        return {"error": f"لا تتوفر بيانات الطقس لـ {location}"}

    return {"error": f"أداة غير معروفة: {tool_name}"}

async def sse_event_generator(request: Request):
    # قراءة جسم الطلب
    body_bytes = await request.body()
    body = json.loads(body_bytes)

    # التعامل مع عمليات MCP المختلفة
    if body["action"] == "list_tools":
        yield f"data: {json.dumps({'tools': TOOLS})}\\\\n\\\\n"

    elif body["action"] == "call_tool":
        tool_name = body["tool_name"]
        parameters = body["parameters"]
        result = await call_tool(tool_name, parameters)
        yield f"data: {json.dumps({'result': result})}\\\\n\\\\n"

@app.post("/sse")
async def sse_endpoint(request: Request):
    return StreamingResponse(
        sse_event_generator(request),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

ميزات متقدمة

تخزين قوائم الأدوات

لتحسين الأداء، يمكنك تخزين قائمة الأدوات من خوادم MCP:

# إنشاء خادم MCP مع تخزين الأدوات
async with MCPServerSse(
    name="خدمة الطقس",
    params={"url": "<https://example.com/mcp/sse>"},
    cache_tools_list=True  # تمكين التخزين
) as mcp_server:
    # استخدام الخادم كما كان من قبل...

عند تعيين cache_tools_list=True، سيستدعي SDK list_tools() على خادم MCP مرة واحدة فقط ويعيد استخدام النتيجة لتشغيلات الوكيل التالية. هذا يقلل من زمن الانتقال، خاصة بالنسبة للخوادم البعيدة.

لتInvalid الكاش إذا لزم الأمر:

mcp_server.invalidate_tools_cache()

تتبع عمليات MCP

يتضمن SDK لوكلاء OpenAI قدرات تتبع مدمجة تلتقط تلقائيًا عمليات MCP:

1. الاتصالات بخادم MCP لتسجيل الأدوات
2. معلومات متعلقة بـ MCP عن استدعاءات الوظائف

يمكنك عرض هذه التتبع في https://platform.openai.com/traces/ عندما تستخدم مدير السياق trace كما هو موضح في الأمثلة أعلاه.

استخدام عدة خوادم MCP

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

async def run_with_multiple_servers():
    async with MCPServerStdio(
        name="نظام الملفات",
        params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
    ) as fs_server, MCPServerSse(
        name="واجهة برمجة تطبيقات الطقس",
        params={"url": "<https://example.com/weather/mcp/sse>"}
    ) as weather_server:
        # إنشاء وكيل مع كلا خادمي MCP
        agent = Agent(
            name="مساعد متعدد الأدوات",
            instructions="استخدم جميع الأدوات المتاحة لمساعدة المستخدم.",
            mcp_servers=[fs_server, weather_server]
        )

        # تشغيل الوكيل
        result = await Runner.run(
            starting_agent=agent,
            input="تحقق أولًا من الطقس في طوكيو، ثم اقرأ محتويات ملف report.txt."
        )
        print(result.final_output)

معالجة الأخطاء وتصحيحها

عند العمل مع خوادم MCP، قد تواجه مشكلات مختلفة. إليك بعض المشاكل الشائعة وكيفية التعامل معها:

مشكلات الاتصال

إذا كانت خادم MCP الخاص بك لا يستجيب:

try:
    async with MCPServerSse(
        name="خدمة الطقس",
        params={"url": "<https://example.com/mcp/sse>"}
    ) as mcp_server:
        # استخدام الخادم...
except Exception as e:
    print(f"فشل الاتصال بخادم MCP: {e}")
    # تنفيذ استراتيجية التراجع

أخطاء تنفيذ الأداة

عندما تفشل عملية تنفيذ أداة، تعامل معها بلطف:

try:
    result = await Runner.run(starting_agent=agent, input=user_query)
    print(result.final_output)
except Exception as e:
    print(f"خطأ أثناء تنفيذ الوكيل: {e}")
    # قد تريد التحقق من سجلات التتبع لمعلومات خطأ مفصلة

الخاتمة

يساعد دعم SDK لوكلاء OpenAI لبروتوكول MCP على توسيع نطاق وكلائك بمجموعة واسعة من الأدوات والقدرات. سواء كنت تستخدم خوادم stdio المحلية أو نهايات SSE البعيدة، فإن التكامل سهل وفعّال.

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

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

تمنياتنا بالتوفيق في البرمجة مع وكلاء OpenAI وMCP!