我的个人博客

程序化调用MCP-Microsoft Playwright MCP 实现招投标数据采集

0 阅读
0 评论

前提需要安装nodejs环境

版本如下：

安装UV 工具


pip install uv

运行以下命令把MCP转化为标准的restful风格的API


uvx mcpo --host 0.0.0.0 --port 8000 -- npx @playwright/mcp@latest --ignore-https-errors --isolated  --save-trace

无头模式运行


uvx mcpo --host 0.0.0.0 --port 8000 -- npx @playwright/mcp@latest --ignore-https-errors --isolated  --save-trace  --no-sandbox --image-responses omit --headless

或者使用python


pip install mcpo
mcpo --host 0.0.0.0 --port 8000 -- npx @playwright/mcp@latest --ignore-https-errors --isolated  --save-trace  --headless

启动如下：

访问http://localhost:8000/docs 可以看到API list

编写MCP client代码模拟客户端

playwright_mcp_client.py


import requests
import json

class PlaywrightMCPClient:
    def __init__(self, base_url):
        """
        初始化 Playwright MCP 客户端
        :param base_url: 服务的基地址，例如 "http://localhost:8080"
        """
        self.base_url = base_url

    def _make_request(self, endpoint, method="POST", data=None):
        """
        发送请求到指定的端点
        :param endpoint: 请求的端点路径
        :param method: 请求方法，默认为 POST
        :param data: 请求体数据
        :return: 响应内容
        """
        url = f"{self.base_url}{endpoint}"
        headers = {"Content-Type": "application/json"}
        response = requests.request(method, url, headers=headers, data=json.dumps(data))
        response.raise_for_status()  # 如果响应状态码不是 200，抛出异常
        return response.json()

    def browser_close(self):
        """
        关闭浏览器页面
        """
        return self._make_request("/browser_close")

    def browser_wait(self, time):
        """
        等待指定的时间（秒）
        :param time: 等待的时间（秒）
        """
        data = {"time": time}
        return self._make_request("/browser_wait", data=data)

    def browser_file_upload(self, paths):
        """
        上传文件
        :param paths: 文件路径列表
        """
        data = {"paths": paths}
        return self._make_request("/browser_file_upload", data=data)

    def browser_install(self):
        """
        安装浏览器
        """
        return self._make_request("/browser_install")

    def browser_press_key(self, key):
        """
        按下键盘上的键
        :param key: 要按下的键，例如 "ArrowLeft" 或 "a"
        """
        data = {"key": key}
        return self._make_request("/browser_press_key", data=data)

    def browser_navigate(self, url):
        """
        导航到指定的 URL
        :param url: 要导航到的 URL
        """
        data = {"url": url}
        result=[]
        result.append(self._make_request("/browser_navigate", data=data))
        res = self.browser_snapshot()
        print(res)
        result.append(res)
        return result

    def browser_navigate_back(self):
        """
        返回到上一页
        """
        return self._make_request("/browser_navigate_back")

    def browser_navigate_forward(self):
        """
        前进到下一页
        """
        return self._make_request("/browser_navigate_forward")

    def browser_pdf_save(self):
        """
        将页面保存为 PDF
        """
        return self._make_request("/browser_pdf_save")

    def browser_snapshot(self):
        """
        捕获当前页面的可访问性快照
        """
        result= self._make_request("/browser_snapshot")
        print(result)
        return result

    def browser_click(self, element, ref):
        """
        在页面上执行点击操作
        :param element: 人类可读的元素描述
        :param ref: 页面快照中的目标元素引用
        """
        data = {"element": element, "ref": ref}
        return self._make_request("/browser_click", data=data)

    def browser_drag(self, start_element, start_ref, end_element, end_ref):
        """
        执行拖动操作
        :param start_element: 源元素的人类可读描述
        :param start_ref: 页面快照中的源元素引用
        :param end_element: 目标元素的人类可读描述
        :param end_ref: 页面快照中的目标元素引用
        """
        data = {
            "startElement": start_element,
            "startRef": start_ref,
            "endElement": end_element,
            "endRef": end_ref,
        }
        return self._make_request("/browser_drag", data=data)

    def browser_hover(self, element, ref):
        """
        在页面元素上执行悬停操作
        :param element: 人类可读的元素描述
        :param ref: 页面快照中的目标元素引用
        """
        data = {"element": element, "ref": ref}
        return self._make_request("/browser_hover", data=data)

    def browser_type(self, element, ref, text, submit=False, slowly=False):
        """
        在可编辑元素中输入文本
        :param element: 人类可读的元素描述
        :param ref: 页面快照中的目标元素引用
        :param text: 要输入的文本
        :param submit: 是否提交输入的文本（按下 Enter 键），默认为 False
        :param slowly: 是否逐字符输入，用于触发页面中的按键处理程序，默认为 False
        """
        data = {
            "element": element,
            "ref": ref,
            "text": text,
            "submit": submit,
            "slowly": slowly,
        }
        return self._make_request("/browser_type", data=data)

    def browser_select_option(self, element, ref, values):
        """
        在下拉菜单中选择选项
        :param element: 人类可读的元素描述
        :param ref: 页面快照中的目标元素引用
        :param values: 要选择的值数组
        """
        data = {"element": element, "ref": ref, "values": values}
        return self._make_request("/browser_select_option", data=data)

    def browser_take_screenshot(self, raw=False):
        """
        捕获当前页面的截图
        :param raw: 是否返回未压缩的图像（PNG 格式），默认为 False，返回 JPEG 图像
        """
        data = {"raw": raw}
        return self._make_request("/browser_take_screenshot", data=data)

    def browser_tab_list(self):
        """
        列出浏览器标签页
        """
        return self._make_request("/browser_tab_list")

    def browser_tab_new(self, url=None):
        """
        打开一个新标签页
        :param url: 新标签页要导航到的 URL，如果不提供，则新标签页为空白
        """
        data = {"url": url} if url else None
        return self._make_request("/browser_tab_new", data=data)

    def browser_tab_select(self, index):
        """
        通过索引选择标签页
        :param index: 要选择的标签页的索引
        """
        data = {"index": index}
        return self._make_request("/browser_tab_select", data=data)

    def browser_tab_close(self, index=None):
        """
        关闭标签页
        :param index: 要关闭的标签页的索引，如果不提供，则关闭当前标签页
        """
        data = {"index": index} if index is not None else None
        return self._make_request("/browser_tab_close", data=data)

编写测试代码自动调用MCP

mcpo-test.py


import requests
import json
from playwright_mcp_client import PlaywrightMCPClient
from openai import OpenAI

class Host:
    def __init__(self):
        self.mcp_server = PlaywrightMCPClient("http://localhost:8000")
        self.llm = OpenAI(api_key="<替换自己deepseek API KEY>", base_url="https://api.deepseek.com")
        self.context = []
        self.discover_capabilities()
    
    def discover_capabilities(self):
        """发现MCP服务器的能力"""
        self.tools = {
            "browser_navigate": {
                "description": "导航到指定URL",
                "parameters": {"url": "string"}
            },
            "browser_snapshot": {
                "description": "获取当前页面的内容",
                "parameters": {}
            },
            "browser_click": {
                "description": "点击页面元素",
                "parameters": {"element": "string", "ref": "string"}
            },
            "browser_type": {
                "description": "在页面元素中输入文本",
                "parameters": {"element": "string", "ref": "string", "text": "string", "submit": "boolean", "slowly": "boolean"}
            },
            "browser_take_screenshot": {
                "description": "截取页面截图",
                "parameters": {"raw": "boolean"}
            },
            "browser_wait": {
                "description": "等待指定的时间（秒）",
                "parameters": {"time": "number"}
            },
            "browser_tab_list": {
                "description": "列出浏览器标签页",
                "parameters": {}
            },
            "browser_tab_new": {
                "description": "打开新标签页",
                "parameters": {"url": "string"}
            },
            "browser_tab_select": {
                "description": "选择标签页",
                "parameters": {"index": "number"}
            }
        }
        print("已发现MCP服务器能力:", list(self.tools.keys()))
    
    def process_user_instruction(self, instruction):
        """处理用户指令"""
        print(f"收到用户指令: {instruction}")
        
        # 构建更详细的系统提示
        system_prompt = (
            "你是一个网页快照数据提取专家。\n"
            f"你可以使用以下工具: {json.dumps(self.tools, ensure_ascii=False)}\n\n"
            "工具调用格式示例:\n
json\n{\"tool\": \"工具名\", \"parameters\": {\"参数名\": \"参数值\"}}\n
\n\n"
        )
        
        self.context.append({"role": "system", "content": system_prompt})
        self.context.append({"role": "user", "content": instruction})
        
        # 发送给LLM进行规划
        return self.send_to_llm()
    
    def send_to_llm(self):
        """将上下文发送给LLM"""
        print("发送上下文给LLM进行规划...")
        response = self.llm.chat.completions.create(
            model="deepseek-chat",
            messages=self.context,
            stream=False
        )
        print(f"上下文信息：{self.context}")
        
        llm_response = response.choices[0].message.content
        self.context.append({"role": "assistant", "content": llm_response})
        
        # 解析LLM的响应，查找工具调用
        return self.parse_llm_response(llm_response)
    
    def parse_llm_response(self, response):
        """解析LLM响应，提取工具调用请求"""
        print("LLM响应:", response + "..." )
        
        # 尝试查找JSON格式的工具调用
        import re
        tool_pattern = r'
json\s(\{.?\})\s*
'
        tool_matches = re.findall(tool_pattern, response, re.DOTALL)
        
        if tool_matches:
            try:
                tool_call = json.loads(tool_matches[0])
                tool_name = tool_call.get("tool")
                parameters = tool_call.get("parameters", {})
                if tool_name in self.tools:
                    return self.execute_tool(tool_name, parameters)
            except json.JSONDecodeError:
                pass
        
        # 如果没有找到JSON格式的工具调用，使用简单的文本匹配
        if "browser_navigate" in response.lower():
            url = self.extract_url(response)
            if url:
                return self.execute_tool("browser_navigate", {"url": url})
        
        if "browser_snapshot" in response.lower():
            return self.execute_tool("browser_snapshot", {})
            
        if "browser_click" in response.lower():
            element_match = re.search(r'element["\']?\s*:\s*["\']([^"\']+)["\']', response)
            ref_match = re.search(r'ref["\']?\s*:\s*["\']([^"\']+)["\']', response)
            if element_match and ref_match:
                return self.execute_tool("browser_click", {
                    "element": element_match.group(1),
                    "ref": ref_match.group(1)
                })
                
        if "browser_type" in response.lower():
            element_match = re.search(r'element["\']?\s*:\s*["\']([^"\']+)["\']', response)
            ref_match = re.search(r'ref["\']?\s*:\s*["\']([^"\']+)["\']', response)
            text_match = re.search(r'text["\']?\s*:\s*["\']([^"\']+)["\']', response)
            if element_match and ref_match and text_match:
                return self.execute_tool("browser_type", {
                    "element": element_match.group(1),
                    "ref": ref_match.group(1),
                    "text": text_match.group(1),
                    "submit": False,
                    "slowly": False
                })
        
        if "get_content" in response.lower():
            return self.execute_tool("get_content", {})
            
        if "click" in response.lower():
            selector_match = re.search(r'selector["\']?\s*:\s*["\']([^"\']+)["\']', response)
            if selector_match:
                selector = selector_match.group(1)
                return self.execute_tool("click", {"selector": selector})
                
        if "fill" in response.lower():
            selector_match = re.search(r'selector["\']?\s*:\s*["\']([^"\']+)["\']', response)
            value_match = re.search(r'value["\']?\s*:\s*["\']([^"\']+)["\']', response)
            if selector_match and value_match:
                return self.execute_tool("fill", {
                    "selector": selector_match.group(1),
                    "value": value_match.group(1)
                })
        
        # 如果没有找到工具调用，返回LLM的响应
        return response
    
    def execute_tool(self, tool_name, parameters):
        """执行工具调用"""
        print(f"执行工具: {tool_name}, 参数: {parameters}")
        
        # 调用对应的MCP服务器工具
        try:
            if tool_name == "browser_navigate":
                result = self.mcp_server.browser_navigate(**parameters)
            elif tool_name == "browser_snapshot":
                result = self.mcp_server.browser_snapshot()
                # 添加HTML解析指导
                self.context.append({
                    "role": "user", 
                    "content": f"工具 {tool_name} 执行结果包含页面快照。请分析页面结构，找出采购公告列表。" +
                              "通常公告会在表格、列表或特定div中。提取每个公告的标题、日期和链接。" +
                              "返回JSON格式的结果，确保数据与网页实际内容一致。"
                })
                return self.send_to_llm()
            elif tool_name == "browser_click":
                result = self.mcp_server.browser_click(**parameters)
            elif tool_name == "browser_type":
                result = self.mcp_server.browser_type(**parameters)
            elif tool_name == "browser_take_screenshot":
                result = self.mcp_server.browser_take_screenshot(**parameters)
            elif tool_name == "browser_wait":
                result = self.mcp_server.browser_wait(**parameters)
            elif tool_name == "browser_tab_list":
                result = self.mcp_server.browser_tab_list()
            elif tool_name == "browser_tab_new":
                result = self.mcp_server.browser_tab_new(**parameters)
            elif tool_name == "browser_tab_select":
                result = self.mcp_server.browser_tab_select(**parameters)
            else:
                return f"未知工具: {tool_name}"
        except Exception as e:
            error_msg = f"执行工具 {tool_name} 时出错: {str(e)}"
            print(error_msg)
            self.context.append({"role": "user", "content": error_msg})
            return self.send_to_llm()
        
            
        self.context.append({"role": "user", "content": f"工具 {tool_name} 执行结果: {result}"})
        
        # 再次发送给LLM
        return self.send_to_llm()

def extract_data(url, instruction):
    """主函数：提取数据"""
    host = Host()
    result = host.process_user_instruction(f"从网页 {url} 提取数据。{instruction}")
    return result

if __name__ == "__main__":
    #url = "http://www.cqship.com/html/qygg/"
    url="https://zc.cqa.cn/eip/websit/index.do"
    instruction = "从网页信息中提取2025-04-10的公告标题，公告链接，发布时间，并以JSON格式输出。"
    
    result = extract_data(url, instruction)
    print("\n最终结果:")
    print(result)

测试结果

测试页面


LLM响应: 从网页快照中提取到2025-04-10的公告信息如下：
json

[

{

"公告标题": "第三届民航科教创新成果展重庆机场展位设计及搭建一体化服务项目拟成交结果公示",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=HYEG27r3T/27AOQqQKEKLJMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "重庆江北国际机场有限公司飞行区航务管理部车辆维修及保养（民航D4115）项目询价通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=PnEKQ2erTkOfDMHxERJIZpMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "2025年至2026年计量器具检测供应商采购项目（第二次）比选通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=LL7iHIaFQxCr7lEg/M6pwZMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "江北机场安检基地保障性租赁住房装修及配套家具家电项目-厨房小家电采购项目竞争性比选通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=phcn8tStQaeWmZ6yf7vw8ZMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "航站楼内外钢架除防锈服务项目拟成交结果公示",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=ZaZSDGEXTYeT20xyrXWzcJMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "农药及化肥供应商竞争性比选采购通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=IX5KsTMQQtGGbsxrB8n8YJMG2yM=",

"发布时间": "2025-04-10"

}

]

...

最终结果:
从网页快照中提取到2025-04-10的公告信息如下：
json

[

{

"公告标题": "第三届民航科教创新成果展重庆机场展位设计及搭建一体化服务项目拟成交结果公示",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=HYEG27r3T/27AOQqQKEKLJMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "重庆江北国际机场有限公司飞行区航务管理部车辆维修及保养（民航D4115）项目询价通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=PnEKQ2erTkOfDMHxERJIZpMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "2025年至2026年计量器具检测供应商采购项目（第二次）比选通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=LL7iHIaFQxCr7lEg/M6pwZMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "江北机场安检基地保障性租赁住房装修及配套家具家电项目-厨房小家电采购项目竞争性比选通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=phcn8tStQaeWmZ6yf7vw8ZMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "航站楼内外钢架除防锈服务项目拟成交结果公示",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=ZaZSDGEXTYeT20xyrXWzcJMG2yM=",

"发布时间": "2025-04-10"

},

{

"公告标题": "农药及化肥供应商竞争性比选采购通知",

"公告链接": "https://zc.cqa.cn/eip/websit/sup/initNoticeDetail.do?listModulePath=/websit/sup&id=IX5KsTMQQtGGbsxrB8n8YJMG2yM=",

"发布时间": "2025-04-10"

}

]

```

结论

这种方法适合自动化部署运行，数据提取依赖LLM，可以灵活处理页面。对于大部分页面结构良好的页面提取成功率高。

评论 (0)

暂无评论，快来抢沙发吧！