पाइथन में MCP सर्वर बनाना: वेबसर्च और स्क्रेप गाइड

# वर्चुअल एन्वायर्नमेंट बनाएं और सक्रिय करें
python -m venv mcp-env
source mcp-env/bin/activate  # Windows पर: mcp-env\Scripts\activate

# MCP SDK और वेब स्क्रैपिंग लाइब्रेरीज इंस्टॉल करें
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"'{query}' के लिए {max_results} परिणाम मिले"

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 "N/A"
                })

        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"   URL: {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": "निकालने के लिए URL"},
                    "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"'{query}' के लिए {len(results)} परिणाम मिले:\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\nURL: {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 का उपयोग करके वेब खोजें। शीर्षक, स्निपेट और URL वापस करते हैं।",
            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": "निकालने के लिए URL"
                    },
                    "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("वेबसर्च-स्क्रैपर 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("Python MCP ट्यूटोरियल", 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 = "नेटवर्क त्रुटि"
    PARSE_ERROR = "पार्स त्रुटि"
    RATE_LIMIT = "रेट लिमिट पार हो गया"
    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()