浏览器自动化作为服务
（注：最终版本应为）
浏览器自动化服务:这篇技术博客文章概述了如何在Amazon ECS Fargate平台上作为容器部署Amazon Nova Act的MCP协议服务器，提供浏览器自动化服务的能力。
由Nova Canvas生成的图像，使用提示：一个受Astro Bot启发的可爱的未来机器人，它有一个闪亮的铬头，大而表情丰富的LED眼睛，以及一个友好的微笑。机器人的胸前清晰地写着“MCP”字样，没有使用任何点。机器人明显地握着一个逼真的USB闪存驱动器——USB设备清晰可见。USB设备带有标准的三叉戟图标。机器人旁边是一个电脑屏幕，显示着一个带有彩色面板的Streamlit网络界面。场景中有充满活力的世界中发光的电路图案。风格简洁而清晰，高分辨率。
介绍：Amazon Nova Act 是一个研究预览版 AI 模型和 SDK，旨在帮助开发人员构建能够在网页浏览器中执行操作的可靠网络代理。通过将 Nova Act 与模型上下文协议（MCP）集成，其浏览器自动化功能可以实现标准化，并通过多种通信渠道向不同的客户端（例如 AI 助手、网页 UI、IDE 扩展）开放。
- 标准I/O（stdio），用于本地脚本和CLI工具，如Amazon Q Developer CLI及Cline VS Code扩展（开源）
- 通过HTTP连接的服务器发送事件SSE，用于实时双向通信
- 流式HTTP传输，此功能最近在最新的MCP规范中添加，用于可扩展的Web交互
此部署将Nova Act作为容器化应用部署在Amazon ECS上运行，并通过ALB作为前端。并通过HTTP连接使用SSE来实现实时通信功能。
注: Amazon Nova服务仅能在美国申请API密钥，因此解决方案部署在AWS的us-east-1（美国东部）区域，以确保符合要求。
nova.amazon.com/act
架构介绍这个解决方案包括以下组成部分：
- MCP 服务端：一个基于 FastAPI 并使用 FastAPI-MCP 实现模型上下文协议，同时提供 Nova Act 功能的 FastAPI 应用程序
- Streamlit 客户端界面：一个用于与 MCP 服务端交互的 web UI 和 MCP 客户端
- AWS 基础设施：ECS Fargate 任务、ALB 以及支持资源
Nova 服务器 MCP 运行在 ECS 上
部署指引 前提条件在将Nova Act MCP服务器（或简称Nova Act MCP）部署到ECS（弹性云服务器）之前，确保你已经做好了以下准备工作：
- 使用了适当的凭证配置了 AWS CLI
- 已经安装了 Node.js 和 npm
- 安装了 AWS CDK (
npm install -g aws-cdk) - 安装并启动了 Docker
- 拥有一个有效的 Nova Act API 密钥
克隆GitHub仓库: git clone https://github.com/awsdataarchitect/nova-act-ecs.git
进入目录: cd nova-act-ecs export NOVA_ACT_API_KEY=你的API密钥步骤 3：部署 CDK 资源堆栈
运行 npm install 命令进行依赖包安装，然后使用 cdk bootstrap 进行初始化，最后执行 cdk deploy 部署应用。部署完成后，CDK 会输出 ALB 的这个 DNS 名称。你可以访问：
- MCP 服务器:
http://<alb-dns-name>/mcp - Streamlit 用户界面:
http://<alb-dns-name>:8501
注：MCP在此指特定的服务器实现技术。
服务器通过fastapi_mcp实现了MCP协议，包含以下组件：
MCP 端点（由 fastapi_mcp 自动处理的）：
/mcp-- 事件流的 SSE 端点/mcp/schema-- 方法发现端点/mcp/jsonrpc-- JSON-RPC 方法调用端点
主要的 API 端点 ：
/browse- 用于所有浏览器自动化任务的综合端点接口。
额外的接口 ：
/health- 用于 ALB 的健康检查/logs- 获取最近服务器日志的端点
实现采用了一种简化的办法，使用一个简单的“浏览”功能来整合浏览器控制和执行指令。
@app.post("/browse", operation_id="browse")
async def browse(request: BrowseRequest) -> BrowseResponse:
"""
使用 Nova Act 来进行浏览任务。
该方法负责浏览器的初始化、导航以及指令的执行。
"""
# 此处省略具体实现细节...该浏览方法接受一个灵活的请求格式，如下所示，可以处理各种浏览需求：
class BrowseRequest(BaseModel):
starting_url: str
instructions: List[str] = Field(..., description='要依次执行的指令列表')
max_steps_per_instruction: int = 30
timeout_per_instruction: Optional[int] = None
schema: Optional[Dict[str, Any]] = None
headless: bool = True响应里包含了浏览会话的所有详细信息：
# `BrowseResponse` 类用于处理浏览请求的响应，包含状态、结果和错误信息。
class BrowseResponse(BaseModel):
# `status` 表示操作的状态
status: str
# `results` 是一个包含任意类型数据的字典列表
results: List[Dict[str, Any]]
# `errors` 是一个包含错误信息的字典列表
errors: List[Dict[str, Any]] = []- 单一全局 Nova Act 实例：服务器维护一个单一的全局 Nova Act 实例
- 无头模式：浏览器始终以无头模式运行，以兼容 ECS
- API 密钥管理：从环境变量或 AWS Secrets Manager 获取 API 密钥
- 结构化数据提取：支持基于模式的数据提取
- 错误处理：全面的错误处理和日志记录
- 线程池执行：使用线程池运行同步的 Nova Act 代码以避免 asyncio 冲突
- 资源监控：监控系统资源（CPU、内存）以进行调试
- 日志缓冲：维护一个循环缓冲区，保存最近的日志供客户端显示
- 控制台输出捕获：捕获所有 stdout/stderr 输出，包括 Nova Act 的输出步骤
服务器使用线程池来运行同步执行的 Nova Act 代码，以避免阻塞 FastAPI 事件循环：
# 在执行浏览序列时使用线程池
logger.info("在后台线程池中运行浏览序列")
browse_result = await asyncio.get_event_loop().run_in_executor(
thread_pool, run_browse_sequence
)服务器也实现了一个日志捕获机制，为客户端提供实时日志，包括拦截标准输出和错误输出来捕捉 Nova Act 的思维过程。Nova Act（简称 Nova Act）
# 日志缓冲实现
class LogBuffer:
def __init__(self, max_size=1000):
self.logs = collections.deque(maxlen=max_size)
self.lock = threading.Lock()
def add(self, log_entry):
with self.lock:
self.logs.append(log_entry)
def get_logs(self, limit=100):
with self.lock:
return list(self.logs)[-limit:]
# 自定义的stdout/stderr拦截器，用于捕获输出
class OutputInterceptor(StringIO):
def __init__(self, log_buffer, stream_name, original_stream):
super().__init__()
self.log_buffer = log_buffer
self.stream_name = stream_name
self.original_stream = original_stream
def write(self, text):
# 写入原始流
self.original_stream.write(text)
# 如果非空，则将其添加到日志缓冲
if text.strip():
self.log_buffer.add(text.rstrip())
def flush(self):
self.original_stream.flush()
# 获取日志的接口
@app.get("/logs")
async def get_logs(limit: int = 100):
return {"logs": log_buffer.get_logs(limit)}查看完整的服务器实现请参见GitHub仓库页面
MCP 客户端实现版本客户端实现提供了一个与Nova Act MCP服务器交互的Python接口。我实现了一个同步版本（即通过请求实现），因为它在Streamlit环境中更为稳定。
客户功能- 连接的管理
- 连接到服务器的健康端点以验证服务器是否可用
- 管理针对所有请求的HTTP会话
- 优雅地处理连接错误
2. API
浏览(starting_url, 指令, 每个指令的最大步骤数, 每个指令的超时时间, 结构, headless)- 执行浏览任务
3. 错误处理：
- 正确的错误传递
- 详细的错误信息
- 连接重试机制
4. 日志提取
get_logs(limit)- 获取最近的服务器日志记录
导入 requests
导入 logging
从 typing 导入 Optional, Dict, Any, List, Union
类 MCPClient:
def __init__(self, base_url: str):
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.connected = False
self._current_url = None
def connect(self) -> bool:
"""初始化到 MCP 服务器的连接"""
尝试:
response = self.session.get(f"{self.base_url}/health")
如果 response.status_code == 200:
self.connected = True
logger.info("已连接到 MCP 服务器")
返回 True
返回 False
除了 Exception as e:
logger.error(f"连接错误: {str(e)}")
返回 False
def browse(self, starting_url: str, instructions: Union[str, List[str]],
max_steps_per_instruction: int = 30,
timeout_per_instruction: Optional[int] = None,
schema: Optional[Dict[str, Any]] = None,
headless: bool = True) -> Dict[str, Any]:
"""在一个浏览器会话中执行一系列操作"""
# 将单个指令转换为列表
如果 isinstance(instructions, str):
instructions = [instructions]
如果 not self.connected:
self.connect()
尝试:
data = {
"starting_url": starting_url,
"instructions": instructions,
"max_steps_per_instruction": max_steps_per_instruction,
"headless": headless
}
如果 timeout_per_instruction:
# 如果设置了超时时间，则添加超时时间到请求数据中
data["timeout_per_instruction"] = timeout_per_instruction
如果 schema:
# 如果提供了模式，则添加模式到请求数据中
data["schema"] = schema
logger.info(f"向 {starting_url} 发送带有 {len(instructions)} 个指令的浏览请求")
response = self.session.post(
f"{self.base_url}/browse",
json=data
)
如果 response.status_code != 200:
抛出 异常(f"服务器返回 {response.status_code}: {response.text}")
result = response.json()
logger.info(f"浏览请求已完成，状态为: {result.get('status')}")
# 更新当前 URL 为 {starting_url}
self._current_url = starting_url
返回 result
除了 Exception as e:
抛出 异常(f"浏览操作发生错误: {str(e)}")
def get_logs(self, limit: int = 100) -> List[str]:
"""从服务器获取最近的日志"""
如果 not self.connected:
self.connect()
尝试:
response = self.session.get(
f"{self.base_url}/logs?limit={limit}"
)
如果 response.status_code != 200:
抛出 异常(f"服务器返回 {response.status_code}: {response.text}")
result = response.json()
返回 result.get("logs", [])
除了 Exception as e:
logger.error(f"获取日志时发生错误: {str(e)}")
返回 []要查看完整客户端实现，请参阅GitHub上的链接。
Streamlit UI 实现Streamlit提供的界面是易于使用的，适用于Nova Act MCP服务器。
UI特性- 单个表单界面：将URL和指令输入结合在一个表单中
- 模式生成器：用于创建提取模式（布尔型、文本型、产品信息、列表项、自定义）
- 执行选项：配置最大步骤数和超时设置
- 结果显示：以格式化方式展示执行结果和解析后的响应
- 历史记录管理：保留之前的操作和结果记录
- 实时日志显示：在可滚动窗口中显示实时服务器日志
- 亚马逊专用示例：预配置的常见亚马逊购物任务示例
UI的一个重要特点是实时日志展示，实时展示服务器的输出，包括Nova Act的思考步骤。
# 在 Streamlit 界面中
with st.expander("服务器日志", expanded=True):
# 添加刷新按钮和自动刷新切换
col1, col2 = st.columns([1, 5])
with col1:
if st.button("🔄 刷新日志"):
try:
st.session_state.logs = st.session_state.client.get_logs(limit=100)
st.session_state.last_log_update = time.time()
except Exception as e:
st.error(f"获取日志时出错: {str(e)}")
with col2:
st.session_state.auto_update_logs = st.checkbox("自动刷新", value=st.session_state.auto_update_logs)
# 显示日志在可滚动区域
if st.session_state.auto_update_logs:
auto_update_logs()
if st.session_state.logs:
st.code("\n".join(st.session_state.logs), language="bash")
else:
st.info("没有可用的日志。运行任务或点击🔄刷新查看日志。")此功能帮助用户了解后台到底在做什么，特别是在长时间操作期间。
- 调试：实时查看Nova Act正在做的事情
- 进度跟踪：监控浏览器自动化任务的进度，确保一切按计划进行。
- 错误识别：快速识别操作失败时的具体问题。
- 透明性：理解Nova Act完成指令时的具体步骤。
Streamlit界面：（MCP客户端）
流利文本界面 — MCP的客户端
完整的UI实现可以查看GitHub仓库页面。
AWS CDK (Infra)AWS CDK 代码会生成所有必需的资源，以便在 Amazon ECS 上部署 Nova Act MCP 的服务器端和客户端。
基础设施组成部分
- VPC 和网络，
- 具有公共子网的VPC、互联网网关、以及为ECS任务和ALB配置的安全组
2. ECS (弹性云服务器) 资源 :
- ECS 集群环境
- Fargate 任务定义
- ECS 服务配置
以下是为服务器和客户端准备的 ECS 相关配置:
- ECS 集群环境
- Fargate 任务定义
- ECS 服务配置
3. 负载平衡。
负载平衡：
- 应用负载均衡器（ALB），增加了空闲超时时间到300秒
- 服务器和客户端的目标组，增加了下线延迟时间的设置
- 用于路由流量的监听规则
4. 存储和秘密 ：
- Docker 图像的 ECR 仓库 (用于存储 Docker 图像)
- 用于存储 API 密钥的密钥管理器
为了支持长时间运行的任务和SSE连接以，ALB配置了更长的超时设置。
// 在你的 CDK 堆栈里
const alb = new elbv2.ApplicationLoadBalancer(this, 'NovaActALB', {
vpc,
internetFacing: true,
loadBalancerName: 'nova-act-alb',
idleTimeout: cdk.Duration.seconds(300) // 从默认的 60 秒增加到 300 秒的空闲超时时间
});
// 具有增加注销延迟的目标组（Target Group）
const serverTargetGroup = new elbv2.ApplicationTargetGroup(this, 'ServerTargetGroup', {
vpc,
port: 8080,
protocol: elbv2.ApplicationProtocol.HTTP,
targetType: elbv2.TargetType.IP,
targets: [serverService],
deregistrationDelay: cdk.Duration.seconds(120), // 增加了注销延迟时间
healthCheck: {
path: '/health',
interval: cdk.Duration.seconds(60),
timeout: cdk.Duration.seconds(5),
healthyHttpCode: '200',
}
});对于基础设施实现过程，请查看GitHub仓库。
与VS Code命令行扩展插件集成
Nova Act MCP 服务器也可以与 VS Code 的 CMLE 扩展一起使用
- 从VS Code市场安装Cline插件
- 配置插件以使用您的MCP服务端，具体步骤如下：
"nova-act": {
"autoApprove": [],
"disabled": false,
"timeout": 300,
"url": "http://<alb-dns-name>/mcp",
"transportType": "sse" // sse (服务器发送事件)
}此配置在 Cline 中启用了 browse MCP 工具，让您可以直接从 IDE 执行浏览器自动化任务，如下：
VS Code 中的代码扩展（开源）
故障排除 连接遇到问题- ALB 连接超时：
- 检查 ALB 安全组是否允许 80 端口的入站流量。
- 确认 ALB 的健康检查是否通过。
- 确保 ECS 任务正在运行并已注册到目标组。
- 重要：对于 SSE（服务器发送事件）连接，需要将 ALB 的空闲超时时间增加到超过 60 秒。
- 将 ALB 目标组的注销延迟设置为与长时间运行的操作相匹配。
2. API密钥问题：
- 确认 API 密钥是否已正确存入 Secrets Manager
- 检查服务器日志中是否有提取 API 密钥时的错误
- 确保 ECS 任务角色有权限读取该密钥
3. 浏览器控制问题：
- 确保检查服务器日志中的浏览器初始化时的错误
- 验证Playwright是否正确安装在容器中
- 确保ECS（弹性计算服务）任务有充足的内存和CPU资源
Nova Act MCP 服务器通过 Server-Sent Events (SSE) 实现实时通信，如果操作耗时超过默认超时设置，可能会导致应用负载均衡器 (ALB) 返回 504 网关超时错误。为解决这个问题：
增加ALB的空闲超时时间：
// 在你的CDK堆栈里
const alb = new elbv2.ApplicationLoadBalancer(this, 'NovaActAlb', {
vpc,
internetFacing: true,
securityGroup: albSecurityGroup,
idleTimeout: cdk.Duration.seconds(300) // 将空闲超时时间从默认的60秒改为300秒
});实现客户端的重试机制：
# 定义一个带有重试机制的浏览函数
def browse_with_retry(self, url, instruction, max_retries=3):
# 尝试次数从0到最大重试次数进行循环:
for attempt in range(max_retries):
try:
return self.browse(url, instruction)
except Exception as e:
if "504 Gateway Time-out" in str(e) and attempt < max_retries - 1:
time.sleep(2) # 等两秒后再试
continue
raise在Streamlit中使用同步模式：
如果您在 Streamlit 中遇到“超时上下文管理器应在任务中使用”的错误，可以切换到使用 requests 库的同步客户端实现，而不是之前的 aiohttp（我之前使用的是 aiohttp，但在当前版本中未再使用）。
要删除所有通过 CDK 创建的 AWS 资源，请运行以下命令。
在命令行中输入 cdk destroy 命令来销毁资源。
这能确保避免不必要的基础设施继续运行，从而避免不必要的额外费用。
最后，我们来总结一下在Amazon ECS上实现的Nova Act MCP服务器提供了一个这样的解决方案，用于浏览器自动化。通过利用模型上下文协议和简化后的API（仅包含一个“浏览”方法），它能够与多种客户端无缝集成，例如AI助手、Streamlit UI、VS Code扩展（Cline、Cursor），并且我们希望很快也能支持Amazon Q开发者CLI。目前，Amazon Q开发者CLI仅支持通过本地stdio协议托管的MCP工具，尚不支持通过sse over http协议的MCP工具。
虽然我还没有试过其他流行的MCP客户端，比如Claude Code、Claude for Desktop、LibreChat、LangChain（MCP适配器模块）和带有Cascade/Windsurf插件的JetBrains，我邀请您用这些MCP客户端测试一下Nova Act MCP服务器，并告诉我您的想法。
这种架构的主要优点包括：
- 简洁性：一个方法搞定所有浏览器自动化任务
- 可扩展性：ECS Fargate服务可伸缩
- 安全性：API密钥安全存储在AWS Secrets Manager中
- 可靠性：健康检查和自动修复确保高可用性
- 监控：CloudWatch集成提供性能监控
- 透明性：实时日志展示Nova Act的思考过程
- 集成性：与VS Code扩展插件配合使用
此解决方案可以扩展来支持多个并发浏览器会话，支持更多自动化功能，并与如 Step Functions 等其他 AWS 服务集成，以处理复杂的工作流程。
参考文献- 模型上下文协议（MCP）规范
- Amazon Nova Act
- FastAPI 说明文档
- FastAPI-MCP 说明文档
- AWS CDK 用户指南
- Amazon ECS 用户指南
- Streamlit 文档
- Playwright 文档
- Amazon Q 用户指南
- 使用 MCP 扩展 Amazon Q 开发者 CLI
Vivek V 是一名与 Cognizant 合作的 AWS 大使，同时也是 AWS 社区建设者（AI 工程），负责机器学习助理 (MLA-C01) 考试的 AWS 培训与认证主题专家（SME）。他拥有 AWS 的全部 15 个认证，AWS 明星奖和 AWS Gold Jacket，并且获得了 4 个 Kubernetes 认证和 5 个 Azure 认证。
维克积极地通过博客、论坛和技术讨论为AWS社区做出贡献。他还是AWS IQ社区论坛和AWS客户咨询委员会成员，并且是Packt的AWS ANS-C01认证指南的技术评审人。
所有表达的观点均为我个人的观点。
感谢你加入我们社区在你走之前:
- 记得给作者点个赞并关注他👏️
- 关注我们：X | 领英 | YouTube | 通讯 | 播客 | Differ | Twitch
- 试试 CoFeed，智能跟随最新科技动态 🧪
- 在 Differ 上免费创建自己的 AI 动态博客 🚀
- 加入我们 Discord 上的内容创作者社区 🧑💻
-
查看更多内容，请访问 plainenglish.io + stackademic.com
[MCP]: 模型上下文协议
[LLM]: 大规模语言模型
[RAG]: 检索增强生成
[SSE]: 服务器推送事件
共同學(xué)習(xí)，寫下你的評(píng)論
評(píng)論加載中...
作者其他優(yōu)質(zhì)文章
