使用 Python 构建 MCP 服务器：网络搜索与爬取指南

使用 Python 示例构建 AI 助手的 MCP 服务器

目录

模型上下文协议（MCP）正在革新AI助手与外部数据源和工具的交互方式。在本指南中，我们将探讨如何构建 MCP 服务器（Python），重点介绍网络搜索和爬取功能的示例。

MCP 机器人

什么是模型上下文协议？

模型上下文协议（MCP）是由Anthropic引入的一种开放协议，用于标准化AI助手与外部系统的连接方式。通过MCP，无需为每个数据源构建自定义集成，而是提供统一的接口，允许：

  • AI助手（如Claude、ChatGPT或自定义LLM应用）发现并使用工具
  • 开发人员通过标准化协议暴露数据源、工具和提示
  • 无缝集成，无需为每个用例重新发明轮子

该协议基于客户端-服务器架构，其中：

  • MCP客户端（AI助手）发现并使用功能
  • MCP服务器暴露资源、工具和提示
  • 通信通过JSON-RPC在stdio或HTTP/SSE上进行

如果您有兴趣在其他语言中实现MCP服务器，请查看我们的指南 实现MCP服务器（Go），该指南详细介绍了协议规范和消息结构。

为什么选择Python构建MCP服务器？

Python是MCP服务器开发的绝佳选择，因为：

  1. 丰富的生态系统：requests、beautifulsoup4、selenium和playwright等库使网络爬取变得简单直接
  2. MCP SDK：官方Python SDK（mcp）提供强大的服务器实现支持
  3. 快速开发：Python的简洁性允许快速原型设计和迭代
  4. AI/ML集成：与langchain、openai等AI库和数据处理工具轻松集成
  5. 社区支持：庞大的社区拥有丰富的文档和示例

设置开发环境

首先，创建一个虚拟环境并安装所需的依赖项。使用虚拟环境对于Python项目隔离至关重要 - 如果您需要复习，请查看我们的 venv速查表 以获取详细命令和最佳实践。

# 创建并激活虚拟环境
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的浏览器驱动

现代替代方案：如果您希望更快地解决依赖项并进行安装，请考虑使用 uv - 现代Python包和环境管理器，它在大型项目中比pip快得多。

构建基本的MCP服务器

让我们从一个最小的MCP服务器结构开始。如果您是Python新手或需要语法和常见模式的快速参考，请查看我们的 Python速查表，它提供了Python基础的全面概述。

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())

实现网络搜索功能

现在，让我们使用DuckDuckGo（无需API密钥）实现一个真正的网络搜索工具：

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)

添加网络爬取功能

让我们添加一个工具，用于从网页上爬取和提取内容。当使用LLM处理HTML内容时，您可能还想将其转换为Markdown格式以获得更好的处理效果。为此，请查看我们的全面指南 使用Python将HTML转换为Markdown，该指南比较了6个不同的库，并提供了基准测试和实用建议。

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"]
            }
        )
    ]

完整的MCP服务器实现

这是一个具有搜索和爬取功能的完整、生产就绪的MCP服务器：

#!/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\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("启动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())

配置您的MCP服务器

要使用您的MCP服务器与Claude Desktop或其他MCP客户端配合使用，请创建一个配置文件：

对于Claude Desktop（claude_desktop_config.json）：

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

位置：

  • macOS：~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows：%APPDATA%\Claude\claude_desktop_config.json
  • Linux：~/.config/Claude/claude_desktop_config.json

测试您的MCP服务器

创建一个测试脚本以验证功能：

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())

高级功能和最佳实践

1. 速率限制

实现速率限制以避免压垮目标服务器：

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)

2. 缓存

添加缓存以提高性能：

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)

3. 错误处理

实现强大的错误处理：

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)}"

4. 输入验证

在处理之前验证用户输入：

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

部署考虑

使用SSE传输进行网络部署

对于基于网络的部署，请使用SSE（Server-Sent Events）传输。如果您正在考虑无服务器部署，您可能会对比较 AWS Lambda在JavaScript、Python和Golang中的性能 感兴趣，以便就运行时做出明智的决策：

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()

AWS Lambda部署

MCP服务器也可以作为AWS Lambda函数部署，尤其是在使用SSE传输时。有关Lambda部署的全面指南：

Docker部署

为容器化部署创建 Dockerfile：

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"]

性能优化

异步操作

使用asyncio进行并发操作：

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()

安全最佳实践

  1. 输入清理：始终验证并清理用户输入
  2. URL 白名单：考虑为爬虫实现 URL 白名单
  3. 超时控制：设置适当的超时时间以防止资源耗尽
  4. 内容限制：限制爬取内容的大小
  5. 身份验证：在生产部署中实现身份验证
  6. HTTPS：在生产环境中使用 HTTPS 进行 SSE 传输

与不同 LLM 提供商的协作

虽然 MCP 是由 Anthropic 为 Claude 开发的，但该协议的设计目的是与任何 LLM 兼容。如果您正在构建与多个 AI 提供商交互并需要结构化输出的 MCP 服务器，您需要查看我们的主流 LLM 提供商结构化输出比较，包括 OpenAI、Gemini、Anthropic、Mistral 和 AWS Bedrock。

有用的链接和资源

相关资源

MCP 和协议实现

Python 开发

网页爬取和内容处理

无服务器部署资源

LLM 集成

结论

使用 Python 构建 MCP 服务器为扩展 AI 助手提供了强大的可能性，使其能够使用自定义工具和数据源。此处展示的网络搜索和爬取功能只是一个开始——您可以在此基础上扩展，以集成数据库、API、文件系统和几乎所有外部系统。

Model Context Protocol 仍在不断发展，但其对 AI 工具集成的标准化方法使其成为开发下一代 AI 驱动应用程序的令人兴奋的技术。无论您是为组织创建内部工具，还是为社区构建公共 MCP 服务器，Python 都为快速开发和部署提供了出色的平台。