Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
138 lines
4.1 KiB
Python
138 lines
4.1 KiB
Python
"""Web page fetch tool - reads content from URLs (safe and stable)."""
|
|
from __future__ import annotations
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from core.state import PlanStep, StepResult, StepStatus
|
|
|
|
|
|
# Configuration limits
|
|
MAX_RESPONSE_SIZE = 500 * 1024 # 500KB
|
|
MAX_OUTPUT_LENGTH = 4000 # Characters
|
|
REQUEST_TIMEOUT = 5 # Seconds
|
|
|
|
|
|
def web_fetch(step: PlanStep) -> StepResult:
|
|
"""Fetch and read content from a web page (safe and stable).
|
|
|
|
Args:
|
|
step: PlanStep with payload containing 'url'
|
|
|
|
Returns:
|
|
StepResult with extracted page content (max 4000 characters)
|
|
"""
|
|
url = step.payload.get("url", "").strip()
|
|
|
|
# Validation: Basic URL check
|
|
if not url:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error="Fetch error: URL not provided"
|
|
)
|
|
|
|
if not url.startswith(("http://", "https://")):
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error="Fetch error: URL must start with http:// or https://"
|
|
)
|
|
|
|
try:
|
|
# Request configuration
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (compatible; AgentMaxim/1.0)"
|
|
}
|
|
|
|
# Streaming response with size limit (follow redirects automatically)
|
|
response = requests.get(
|
|
url,
|
|
headers=headers,
|
|
timeout=REQUEST_TIMEOUT,
|
|
stream=True,
|
|
allow_redirects=True
|
|
)
|
|
|
|
# Content-type validation
|
|
content_type = response.headers.get("content-type", "")
|
|
if "text/html" not in content_type:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"Fetch error: Invalid content-type: {content_type}"
|
|
)
|
|
|
|
# Read with size limit
|
|
content = b""
|
|
for chunk in response.iter_content(chunk_size=8192):
|
|
content += chunk
|
|
if len(content) > MAX_RESPONSE_SIZE:
|
|
break
|
|
|
|
if len(content) == 0:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error="Fetch error: Empty response"
|
|
)
|
|
|
|
# Parse HTML
|
|
soup = BeautifulSoup(content, "html.parser")
|
|
|
|
# Smart content extraction
|
|
# Priority: main container → article → content divs → body
|
|
main_content = (
|
|
soup.find("main") or
|
|
soup.find("article") or
|
|
soup.find("div", class_=["content", "main-content", "article-content"]) or
|
|
soup.body
|
|
)
|
|
|
|
if not main_content:
|
|
main_content = soup.find("body")
|
|
|
|
# Extract text with cleaning
|
|
text = main_content.get_text(separator="\n", strip=True)
|
|
|
|
# Remove excessive whitespace
|
|
import re
|
|
text = re.sub(r'\n{3,}', '\n\n', text)
|
|
text = re.sub(r' +', ' ', text)
|
|
|
|
# Limit output length
|
|
if len(text) > MAX_OUTPUT_LENGTH:
|
|
text = text[:MAX_OUTPUT_LENGTH] + "\n\n... (truncated)"
|
|
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.DONE,
|
|
output_text=text,
|
|
metadata={
|
|
"source": url,
|
|
"content_type": content_type,
|
|
"original_size": len(content),
|
|
"output_size": len(text)
|
|
}
|
|
)
|
|
|
|
except requests.exceptions.Timeout:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"Fetch error: Request timed out (>{REQUEST_TIMEOUT}s)"
|
|
)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"Fetch error: {str(e)}"
|
|
)
|
|
|
|
except Exception as e:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"Fetch error: {type(e).__name__}: {str(e)}"
|
|
)
|