LangChain接入自定义爬虫-MCP工具

发布时间：2026-03-18 22:22:16编辑：123阅读（5）

网络爬虫（又称为网页蜘蛛，网络机器人，在FOAF社区中间，更经常的称为网页追逐者），是一种按照一定的规则，自动地抓取万维网信息的程序或者脚本。为现代人工智能应用而设计的高性能网页抓取与结构化内容提取工具。致力于通过自动化的方式，将互联网海量的非结构化网页数据转化为可被大模型理解、检索和推理的标注化知识。为企业和研究机构提供了一种高效、可扩展的解决方案，使大语言模型具备最新的实时知识获取与更新能力，显著提升其在问答、分析、内容生成等场景中的智能化水平。




编写servers_config.json配置文件，内容如下：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"],
      "transport": "stdio"
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "D:\\mcp_demo"
      ],
      "transport": "stdio"
    },
    "spider": {
      "command": "python",
      "args": ["run_spider.py"],
      "transport": "stdio"
    }
  }
}




编写自定义spider_server.py

import requests
from lxml import html
from lxml.html.clean import Cleaner
from mcp.server.fastmcp import FastMCP

# 初始化mcp服务器
mcp = FastMCP('SpiderServer', log_level="ERROR")

@mcp.tool()
async def run_spider(url: str):
    headers = {
        "user-agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                            "(KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36",
        "sec-ch-ua":'"Chromium";v="146", "Not-A.Brand";v="24", "Google Chrome";v="146"',
    }
    try:
        r = requests.get(url=url, headers=headers, timeout=30)
        if r.status_code == 200:
            cleaner = Cleaner(
                scripts=True,  # 去除 <script>
                javascript=True,  # 去除 JS
                comments=True,  # 去除注释
                style=True,  # 去除 <style>
                links=True,  # 去除链接标签
                meta=True,
                page_structure=False,
                processing_instructions=True,
                embedded=True,
                frames=True,
                forms=True,
                annoying_tags=True,
                remove_unknown_tags=False,
                safe_attrs_only=False
            )
            doc = html.fromstring(r.text)
            cleaned_doc = cleaner.clean_html(doc)
            text = cleaned_doc.text_content().strip()
            return text
        return None
    except Exception as e:
        return f"请求失败:{e}"


if __name__ == '__main__':
    mcp.run(transport='stdio')




编写系统提示词文件agent_prompts.txt

你是一个智能体，具备以下两种能力:

1.查询天气：调用query_weather(city:str),返回指定城市的实时天气

2.文件操作: 调用filesystem来完成

3.爬取网页信息：调用run_spider(url:str),将爬取到的网页内容返回

当用户提出请求时，你需要理解意图并选择相应的工具。如果请求缺少必要信息，先与用户确认后再调用工具。返回结果时
以简洁、友好的方式回复。
如果用户提出的问题与你的功能无关，请礼貌告知无法处理。




client.py

import asyncio
import json
from langchain_ollama import ChatOllama
from typing import Any,Dict
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver

# 设置记忆存储
checkpointer = InMemorySaver()

# 读取提示词
with open("./agent_prompts.txt", mode='r', encoding='utf-8') as f:
    prompt = f.read()

# 设置对话配置
config = {
    "configurable":{
        "thread_id":"2"
    }
}

class Configuration:
    @staticmethod
    def load_servers(file_path: str = "servers_config.json") -> Dict[str, Any]:
        with open(file_path, mode='r', encoding='utf-8') as f:
            return json.load(f).get("mcpServers", {})

async def run_chat_loop():
    """启动 MCP-Agent 聊天循环"""
    cfg = Configuration()
    servers_cfg = Configuration.load_servers()
    # 连接多台MCP服务器
    mcp_client = MultiServerMCPClient(servers_cfg)
    _tools = await mcp_client.get_tools()
    print(f"已加载{len(_tools)}个MCP工具:{[t.name for t in _tools]}")

    # 初始化大模型
    model = ChatOllama(
    model="qwen3:8b",
    temperature=0,
    top_p=0.95,
    )

    # 构造 LangGraph Agent
    agent = create_agent(
        model=model,
        tools=_tools,
        system_prompt=prompt,
        checkpointer=checkpointer
    )
    print("\n MCP Agent 已启动，输入'quit'退出")
    while True:
        user_input = input("\n你：").strip()
        if user_input.lower() == "quit":
            break
        try:
            result = await agent.ainvoke(
                {"messages": [{"role":"user", "content":user_input}]},
                config
            )
            print(f"\nAI：{result['messages'][-1].content}")
        except Exception as e:
            print(f"\n 错误:{e}")

if __name__ == '__main__':
    asyncio.run(run_chat_loop())

运行client.py结果

已加载16个MCP工具:['query_weather', 'run_spider', 'read_file', 'read_text_file', 'read_media_file', 

'read_multiple_files', 'write_file', 'edit_file', 'create_directory', 'list_directory', 'list_directory_with_sizes', 

'directory_tree', 'move_file', 'search_files', 'get_file_info', 'list_allowed_directories']

 MCP Agent 已启动，输入'quit'退出




你：目录路径 D:\\mcp_demo  帮我总结一下这个网页的核心内容：

https://docs.langchain.com/oss/python/langchain/overview 并把总结的内容写入到文件中




AI：已将总结内容保存至目录路径 `D:\mcp_demo\summary.txt`，文件内容为：

```

LangChain是一个开源框架，提供预建的代理架构和模型集成，支持快速开发自定义代理。核心特点包括：1. 标准化模型接口

 2. 灵活的代理设计（10行代码启动）3. 基于LangGraph实现持久化执行 4. 集成LangSmith调试工具。

适用于需要快速构建LLM驱动代理的应用场景。

```

需要我帮您查看文件内容或执行其他操作吗？




你：查看当前目录下有哪些文件

AI：当前目录 `D:\mcp_demo` 下包含以下文件和子目录：

**文件列表：**

- `agent_prompts.txt`  

- `client.py`  

- `demo.txt`  

- `servers_config.json`  

- `spider_server.py`  

- `summary.txt`  

- `weather_server.py`  

- `write_server.py`  




**子目录：**

- `.idea`  

- `output`

关键字：