بناء خوادم MCP في بايثون: دليل البحث على الويب والتنقيب

# إنشاء وتفعيل بيئة افتراضية
python -m venv mcp-env
source mcp-env/bin/activate  # على Windows: mcp-env\Scripts\activate

# تثبيت مكتبة MCP ومحررات التنقيب عبر الإنترنت
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # تثبيت محركات المتصفح لـ Playwright

import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

# إنشاء خادم
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """تحديد الأدوات المتاحة"""
    return [
        Tool(
            name="search_web",
            description="البحث عبر الإنترنت عن المعلومات",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "استعلام البحث"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "عدد النتائج القصوى",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """معالجة تنفيذ الأدوات"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # تنفيذ منطق البحث هنا
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"نتائج البحث عن '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"أداة غير معروفة: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """مُحلّف لتنفيذ البحث الفعلي"""
    return f"تم العثور على {max_results} نتائج لـ: {query}"

async def main():
    """تشغيل الخادم"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

import asyncio
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus

async def search_duckduckgo(query: str, max_results: int = 5) -> list[dict]:
    """البحث في DuckDuckGo وتحليل النتائج"""
    
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title_elem = result.select_one('.result__title')
            snippet_elem = result.select_one('.result__snippet')
            url_elem = result.select_one('.result__url')
            
            if title_elem and snippet_elem:
                results.append({
                    "title": title_elem.get_text(strip=True),
                    "snippet": snippet_elem.get_text(strip=True),
                    "url": url_elem.get_text(strip=True) if url_elem else "غير متوفر"
                })
        
        return results
        
    except Exception as e:
        raise Exception(f"فشل البحث: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """تنسيق نتائج البحث للعرض"""
    if not results:
        return "لم يتم العثور على نتائج."
    
    formatted = []
    for i, result in enumerate(results, 1):
        formatted.append(
            f"{i}. {result['title']}\n"
            f"   {result['snippet']}\n"
            f"   الرابط: {result['url']}\n"
        )
    
    return "\n".join(formatted)

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """تنقيب المحتوى من صفحة ويب باستخدام Playwright"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            
            # انتظر لتحميل المحتوى
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # استخراج عنصر محدد
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "لم يتم العثور على المحدد"
            else:
                # استخراج المحتوى الرئيسي
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # تقييد طول المحتوى
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# إضافة أداة التنقيب إلى خادم MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="البحث عبر الإنترنت باستخدام DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "استعلام البحث"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="تنقيب المحتوى من صفحة ويب",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "الرابط لتنقيبه"},
                    "selector": {
                        "type": "string",
                        "description": "محدد CSS خاص للمحتوى"
                    }
                },
                "required": ["url"]
            }
        )
    ]

#!/usr/bin/env python3
"""
خادم MCP لبحث وتنقيب الويب
يوفر أدوات للبحث عبر الإنترنت واستخراج المحتوى من الصفحات
"""

import asyncio
import logging
from typing import Any
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
from playwright.async_api import async_playwright

from mcp.server import Server
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
import mcp.server.stdio

# تكوين السجل
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# إنشاء خادم
app = Server("websearch-scraper")

# تنفيذ البحث
async def search_web(query: str, max_results: int = 5) -> str:
    """البحث في DuckDuckGo وإرجاع النتائج المُنسقة"""
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {"User-Agent": "Mozilla/5.0"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title = result.select_one('.result__title')
            snippet = result.select_one('.result__snippet')
            link = result.select_one('.result__url')
            
            if title and snippet:
                results.append({
                    "title": title.get_text(strip=True),
                    "snippet": snippet.get_text(strip=True),
                    "url": link.get_text(strip=True) if link else ""
                })
        
        # تنسيق النتائج
        if not results:
            return "لم يتم العثور على نتائج."
        
        formatted = [f"تم العثور على {len(results)} نتائج لـ '{query}':\n"]
        for i, r in enumerate(results, 1):
            formatted.append(f"\n{i}. **{r['title']}**")
            formatted.append(f"   {r['snippet']}")
            formatted.append(f"   {r['url']}")
        
        return "\n".join(formatted)
        
    except Exception as e:
        logger.error(f"فشل البحث: {e}")
        return f"خطأ في البحث: {str(e)}"

# تنفيذ التنقيب
async def scrape_page(url: str, selector: str = None) -> str:
    """تنقيب محتوى الصفحة باستخدام Playwright"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            await page.wait_for_load_state('networkidle')
            
            title = await page.title()
            
            if selector:
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "لم يتم العثور على المحدد"
            else:
                content = await page.inner_text('body')
            
            # تقييد طول المحتوى
            content = content[:8000] + "..." if len(content) > 8000 else content
            
            result = f"**{title}**\n\nالرابط: {url}\n\n{content}"
            return result
            
        except Exception as e:
            logger.error(f"فشل التنقيب: {e}")
            return f"خطأ في التنقيب: {str(e)}"
        finally:
            await browser.close()

# تعريف أدوات MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """قائمة الأدوات المتاحة في MCP"""
    return [
        Tool(
            name="search_web",
            description="البحث عبر الإنترنت باستخدام DuckDuckGo. يعيد العناوين، المقتطفات، والروابط.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "الاستعلام البحثي"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "عدد النتائج القصوى (الافتراضي: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="استخراج محتوى من صفحة ويب. يمكن استهداف عناصر محددة باستخدام محددات CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "الرابط لتنقيبه"
                    },
                    "selector": {
                        "type": "string",
                        "description": "محدد CSS خاص لاستخراج محتوى محدد"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """معالجة تنفيذ الأدوات"""
    try:
        if name == "search_web":
            query = arguments["query"]
            max_results = arguments.get("max_results", 5)
            result = await search_web(query, max_results)
            return [TextContent(type="text", text=result)]
        
        elif name == "scrape_webpage":
            url = arguments["url"]
            selector = arguments.get("selector")
            result = await scrape_page(url, selector)
            return [TextContent(type="text", text=result)]
        
        else:
            raise ValueError(f"أداة غير معروفة: {name}")
    
    except Exception as e:
        logger.error(f"فشل تنفيذ الأداة: {e}")
        return [TextContent(
            type="text",
            text=f"خطأ في تنفيذ {name}: {str(e)}"
        )]

async def main():
    """تشغيل خادم MCP"""
    logger.info("تشغيل خادم WebSearch-Scraper MCP")
    
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/path/to/your/mcp_server.py"
      ],
      "env": {}
    }
  }
}

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """اختبار خادم MCP محليًا"""
    
    # اختبار البحث
    print("اختبار البحث عبر الإنترنت...")
    results = await search_web("دورة تدريبية MCP لـ Python", 3)
    print(results)
    
    # اختبار التنقيب
    print("\n\nاختبار مُحلّل الصفحة...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
    
    async def acquire(self, key: str):
        now = time.time()
        self.requests[key] = [
            t for t in self.requests[key] 
            if now - t < self.time_window
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            raise Exception("تم تجاوز حد الطلب")
        
        self.requests[key].append(now)

limiter = RateLimiter(max_requests=10, time_window=60)

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
async def cached_search(query: str, max_results: int):
    return await search_web(query, max_results)

from enum import Enum

class ErrorType(Enum):
    NETWORK_ERROR = "network_error"
    PARSE_ERROR = "parse_error"
    RATE_LIMIT = "rate_limit_exceeded"
    INVALID_INPUT = "invalid_input"

def handle_error(error: Exception, error_type: ErrorType) -> str:
    logger.error(f"{error_type.value}: {str(error)}")
    return f"خطأ ({error_type.value}): {str(error)}"

from urllib.parse import urlparse

def validate_url(url: str) -> bool:
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except:
        return False

def validate_query(query: str) -> bool:
    return len(query.strip()) > 0 and len(query) < 500

import mcp.server.sse

async def main_sse():
    """تشغيل الخادم مع نقل SSE"""
    from starlette.applications import Starlette
    from starlette.routing import Mount
    
    sse = mcp.server.sse.SseServerTransport("/messages")
    
    starlette_app = Starlette(
        routes=[
            Mount("/mcp", app=sse.get_server())
        ]
    )
    
    import uvicorn
    await uvicorn.Server(
        config=uvicorn.Config(starlette_app, host="0.0.0.0", port=8000)
    ).serve()

FROM python:3.11-slim

WORKDIR /app

# تثبيت الاعتماديات النظامية
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# تثبيت الاعتماديات لـ Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# نسخ التطبيق
COPY mcp_server.py .

CMD ["python", "mcp_server.py"]

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """البحث في عدة استعلامات بشكل متزامن"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

import aiohttp

session = None

async def get_session():
    global session
    if session is None:
        session = aiohttp.ClientSession()
    return session

async def fetch_url(url: str) -> str:
    session = await get_session()
    async with session.get(url) as response:
        return await response.text()