[Langchain网页抓取与天气查询实战]MCP篇

MCP (Model Context Protocol) 网页抓取与天气查询实战

📁 项目概述

🏗️ 架构设计

┌─────────────────────────────────────────────────────────────┐ │ 用户输入 │ │ "上海天气怎么样？" │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ MCP Web 客户端 │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ LLM (qwen-plus) │ │ │ │ tool_choice="auto" │ │ │ │ 自动判断是否需要调用工具 │ │ │ └─────────────────────────────────────────────────────┘ │ │ │ │ │ ┌───────────┴───────────┐ │ │ ▼ ▼ │ │ [无需工具，直接回答] [需要工具，调用 MCP] │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ MCP Web 服务端 │ │ ┌──────────────┐ ┌──────────────────┐ │ │ │ fetch_webpage│ │ get_weather │ │ │ │ 网页抓取工具 │ │ 天气查询工具 │ │ │ └──────────────┘ └──────────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ httpx.get() OpenWeather API │ │ 任意网页 2d49a9ae... │ └─────────────────────────────────────────────────────────────┘

📂 文件清单

🔧 核心实现

1. MCP 服务端 (McpServer.py)

1.1 极简 MCP 服务类

class MCPWebServer: """极简版 MCP 服务类，支持 SSE 传输协议""" ​ def __init__(self, name: str, host: str, port: int): self.name = name self.host = host self.port = port self._tools = {} # 存储注册的工具函数 ​ def tool(self): """实现 @mcp.tool() 装饰器""" def decorator(func): self._tools[func.__name__] = func return func return decorator ​ def run(self, transport: str): """启动 MCP 服务""" logger.info(f"启动 MCP Web 服务器，监听 http://{self.host}:{self.port}/sse") self._keep_alive()

设计要点：

• 无第三方依赖，纯原生实现

• 适配 Python 3.13+

• @mcp.tool()装饰器注册工具函数

• _tools字典存储所有已注册的工具

1.2 工具1: 网页抓取 (fetch_webpage)

@mcp.tool() def fetch_webpage(url: str, max_length: int = 5000) -> str: """ 抓取指定网页的文本内容。 参数: url: 网页的完整 URL 地址 max_length: 最大抓取字符数，默认 5000 返回: 网页的文本内容（已去除 HTML 标签） """ headers = {"User-Agent": "Mozilla/5.0..."} resp = httpx.get(url, headers=headers, timeout=15) # HTML 标签去除 html = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.DOTALL) html = re.sub(r'<style[^>]*>.*?</style>', '', html, flags=re.DOTALL) text = re.sub(r'<[^>]+>', '', html) # 截断超长内容 if len(text) > max_length: text = text[:max_length] + "...[内容已截断]" return json.dumps({"url": url, "content": text, "status": "success"})

功能特点：

• ✅ 自动去除<script>和<style>标签

• ✅ 移除所有 HTML 标签

• ✅ 清理多余空白字符

• ✅ 自动截断超长内容

• ❌ 无法处理 JavaScript 动态渲染的页面

1.3 工具2: 天气查询 (get_weather)

@mcp.tool() def get_weather(city: str) -> str: """ 查询指定城市的即时天气信息。 参数 city: 城市英文名，如 Beijing 返回: OpenWeather API 的 JSON 字符串 """ api_key = os.getenv("OPENWEATHER_API_KEY") if not api_key: api_key = "####################" # 备用默认值 params = { "q": city, "appid": api_key, "units": "metric", "lang": "zh_cn" } resp = httpx.get("https://api.openweathermap.org/data/2.5/weather", params=params) return json.dumps(resp.json())

API Key 优先级：

环境变量 OPENWEATHER_API_KEY > 内置默认值

2. MCP 客户端 (McpClient_WebFetch.py)

2.1 MCP 客户端类

# ---------------------- LLM 配置 ---------------------- # 使用阿里云 DashScope API client = OpenAI( api_key=os.getenv("QWEN_API_KEY", os.getenv("aliQwen-api", "")), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) MODEL_NAME = "qwen-plus"

class MCPWebClient: """MCP Web 服务客户端，由 LLM 自动判断调用哪个工具""" ​ def __init__(self, mcp_instance): self.mcp_instance = mcp_instance self.available_tools = mcp_instance._tools logger.info(f"已连接 MCP 服务，可用工具: {list(self.available_tools.keys())}") ​ def call_tool(self, tool_name: str, **kwargs): """调用指定工具""" if tool_name not in self.available_tools: logger.error(f"工具 '{tool_name}' 不存在") return None return self.available_tools[tool_name](**kwargs)

2.2 动态构建 Tools Schema

def build_tools_schema(): """根据 MCP 注册的工具，自动生成 OpenAI function calling 的 tools 参数""" tools = [] # 工具1: fetch_webpage tools.append({ "type": "function", "function": { "name": "fetch_webpage", "description": "抓取指定网页的文本内容...", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "网页 URL"}, "max_length": {"type": "integer", "default": 5000} }, "required": ["url"] } } }) # 工具2: get_weather tools.append({ "type": "function", "function": { "name": "get_weather", "description": "查询指定城市的即时天气信息...", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市英文名"} }, "required": ["city"] } } }) return tools

2.3 核心对话逻辑 (三阶段)

def chat(user_input: str, mcp_client: MCPWebClient): """LLM 自动判断是否调用工具，并返回最终回复""" # ========== 阶段1: LLM 判断 ========== response = client.chat.completions.create( model="qwen-plus", messages=[{"role": "user", "content": user_input}], tools=build_tools_schema(), tool_choice="auto" # LLM 自动判断是否调用工具 ) # ========== 阶段2: 执行工具 ========== if message.tool_calls: for tool_call in message.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) # 执行 MCP 工具 tool_result = mcp_client.call_tool(tool_name, **tool_args) # 把工具结果返回给 LLM messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": tool_result }) # ========== 阶段3: 生成最终回复 ========== final_response = client.chat.completions.create( model="qwen-plus", messages=messages ) return final_response.choices[0].message.content

三阶段流程图：

用户输入 │ ▼ ┌───────────────────────────────────────┐ │ 阶段1: LLM 判断 │ │ tool_choice="auto" │ │ 判断是否需要调用工具？ │ └───────────────┬───────────────────────┘ │ ┌───────┴───────┐ ▼ ▼ 是，需要工具 否，不需要工具 │ │ ▼ │ ┌───────────────────┐ │ │ 阶段2: 执行工具 │ │ │ mcp_client │ │ │ .call_tool() │ │ └─────────┬─────────┘ │ │ │ ▼ │ ┌───────────────────┐ │ │ 阶段3: 生成回复 │ │ │ LLM 根据工具结果 │ │ │ 生成最终回复 │ │ └─────────┬─────────┘ │ │ │ └──────┬──────┘ ▼ 最终回复

🚀 使用方法

环境准备

# 安装依赖 pip install loguru httpx openai python-dotenv ​ # 或使用 uv uv pip install loguru httpx openai python-dotenv

预期输出：

2026-05-05 09:00:00 | INFO | MCP Web 服务器 (提供网页抓取 + 天气查询) 2026-05-05 09:00:00 | INFO | 监听地址: http://127.0.0.1:8000/sse 2026-05-05 09:00:00 | INFO | 已注册工具: ['fetch_webpage', 'get_weather']

启动客户端

# 终端2 python McpClient_WebFetch.py

对话示例

================================================== MCP 智能客户端 (输入问题，LLM 自动选择工具) 可用工具: fetch_webpage / get_weather 输入 'quit' 退出 ================================================== ​ 你: 上海天气怎么样 [LLM 决定调用工具] get_weather({"city": "Shanghai"}) [工具返回结果] {"coord": {"lon": 121.45, "lat": 31.22}, "weather": [{"description": "多云"}], "main": {"temp": 24.55...}} ​ 助手: 上海目前天气为**多云**，气温约 **24.6°C**，体感温度约 **24.0°C**； - 湿度：37% - 气压：1017 hPa - 风速：3.94 m/s ​ 你: 帮我抓取百度首页 [LLM 决定调用工具] fetch_webpage({"url": "https://www.baidu.com"}) [工具返回结果] {"url": "https://www.baidu.com", "content": "", "status": "success"} ​ 助手: 已成功访问百度首页，但网页内容未返回（可能因反爬机制）。 ​ 你: 你好 助手: 你好！有什么我可以帮你的吗？ ​ 你: quit 再见!

📊 测试结果

⚠️ 已知问题

1. 网页抓取返回空内容

现象：抓取百度等大型网站时，返回content: ""

原因：

• 目标网站有反爬机制

• 网页内容由 JavaScript 动态渲染

• 简单的 HTML 解析无法获取动态内容

解决方案：

• 对于动态网页，考虑使用 Selenium/Playwright

• 对于反爬网站，考虑添加更多请求头或使用代理

2. Python 3.13 兼容性问题

现象：McpServerByFastMCP.py使用 FastMCP 时报错

ModuleNotFoundError: No module named 'pywintypes'

原因：pywin32尚未适配 Python 3.13

解决方案：使用本项目的极简版 MCP 服务类，不依赖 FastMCP

📚 扩展方向

1. 增加更多工具

@mcp.tool() def search_news(keyword: str) -> str: """搜索新闻""" # 实现搜索逻辑 pass ​ @mcp.tool() def translate(text: str, target_lang: str) -> str: """翻译文本""" # 实现翻译逻辑 pass

2. 支持 STDIO 传输协议

# 当前: SSE mcp.run(transport="sse") ​ # 扩展: STDIO mcp.run(transport="stdio")

3. 集成 LangChain

from langchain.agents import initialize_agent, Tool from langchain.llms import OpenAI ​ # 将 MCP 工具转换为 LangChain Tools tools = [ Tool( name="fetch_webpage", func=fetch_webpage, description="抓取网页内容" ), Tool( name="get_weather", func=get_weather, description="查询天气" ) ] ​ # 使用 LangChain Agent agent = initialize_agent(tools, llm, agent="zero-shot-react-description")

📝 总结

本项目实现了一个完整的 MCP (Model Context Protocol) 示例：

核心价值：

• 理解了 MCP 协议的基本工作原理

• 掌握了 LLM Function Calling 的使用

• 实现了 Tool 动态注册和调用机制

• 为 LangChain Agent 开发打下基础