mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(shell): add curl, jq, wget tools for shell escaping-free operations
Add three new shell tools that bypass shell escaping issues by using subprocess with list arguments instead of shell interpretation: - curl: HTTP client with proper JSON body handling - jq: JSON processor with direct filter passing - wget: File/site downloads with mirror mode support These tools solve the common problem of shell escaping nightmares when working with complex JSON or special characters. Updates: - Added CurlTool, JqTool, WgetTool to hanzo-tools-shell - Updated __all__ exports and TOOLS list - Added tools to entrypoint_loader PACKAGE_TOOL_PREFIXES
This commit is contained in:
@@ -29,7 +29,7 @@ TOOLS_ENTRY_POINT_GROUP = "hanzo.tools"
|
||||
# Package to tool prefix mapping (for enable/disable)
|
||||
PACKAGE_TOOL_PREFIXES: dict[str, list[str]] = {
|
||||
"filesystem": ["read", "write", "edit", "tree", "ast", "search", "find"],
|
||||
"shell": ["dag", "ps", "zsh", "shell", "npx", "uvx", "open"],
|
||||
"shell": ["dag", "ps", "zsh", "shell", "npx", "uvx", "open", "curl", "jq", "wget"],
|
||||
"browser": ["browser"],
|
||||
"memory": ["memory"], # Unified memory tool
|
||||
"todo": ["todo"],
|
||||
|
||||
@@ -7,6 +7,11 @@ Core tools:
|
||||
- ps: Process management (list, kill, logs)
|
||||
- zsh: Zsh shell execution with auto-backgrounding
|
||||
|
||||
HTTP/Data tools:
|
||||
- curl: HTTP client without shell escaping issues
|
||||
- jq: JSON processor without shell escaping issues
|
||||
- wget: File/site downloads with mirroring support
|
||||
|
||||
Convenience tools:
|
||||
- npx: Node package execution with auto-backgrounding
|
||||
- uvx: Python package execution with auto-backgrounding
|
||||
@@ -28,6 +33,11 @@ from hanzo_tools.shell.zsh_tool import ZshTool, ShellTool, zsh_tool, shell_tool
|
||||
# Convenience tools
|
||||
from hanzo_tools.shell.open_tool import OpenTool, open_tool
|
||||
|
||||
# HTTP/Data tools (no shell escaping issues)
|
||||
from hanzo_tools.shell.curl_tool import CurlTool
|
||||
from hanzo_tools.shell.jq_tool import JqTool
|
||||
from hanzo_tools.shell.wget_tool import WgetTool
|
||||
|
||||
# Base classes
|
||||
from hanzo_tools.shell.base_process import (
|
||||
BaseBinaryTool,
|
||||
@@ -38,7 +48,7 @@ from hanzo_tools.shell.base_process import (
|
||||
)
|
||||
|
||||
# Tools list for entry point discovery
|
||||
TOOLS = [DagTool, PsTool, ZshTool, ShellTool, NpxTool, UvxTool, OpenTool]
|
||||
TOOLS = [DagTool, PsTool, ZshTool, ShellTool, NpxTool, UvxTool, OpenTool, CurlTool, JqTool, WgetTool]
|
||||
|
||||
__all__ = [
|
||||
# Base classes
|
||||
@@ -67,6 +77,10 @@ __all__ = [
|
||||
"npx_tool",
|
||||
"UvxTool",
|
||||
"uvx_tool",
|
||||
# HTTP/Data tools
|
||||
"CurlTool",
|
||||
"JqTool",
|
||||
"WgetTool",
|
||||
# Registration
|
||||
"TOOLS",
|
||||
"get_shell_tools",
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
"""Curl tool - HTTP client without shell escaping issues.
|
||||
|
||||
Provides a clean interface for HTTP requests, avoiding shell quoting problems.
|
||||
"""
|
||||
|
||||
import json
|
||||
import asyncio
|
||||
from typing import Literal, Optional, Annotated, final, override
|
||||
|
||||
from pydantic import Field
|
||||
from mcp.server import FastMCP
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
|
||||
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
|
||||
|
||||
Method = Annotated[
|
||||
Literal["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
|
||||
Field(description="HTTP method"),
|
||||
]
|
||||
|
||||
|
||||
@final
|
||||
class CurlTool(BaseTool):
|
||||
"""HTTP client tool - curl without shell escaping nightmares.
|
||||
|
||||
Clean interface for HTTP requests with proper JSON handling.
|
||||
"""
|
||||
|
||||
name = "curl"
|
||||
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
return """HTTP client - make requests without shell escaping issues.
|
||||
|
||||
Examples:
|
||||
curl --url "https://api.example.com/health"
|
||||
curl --url "https://api.example.com/data" --method POST --json '{"key": "value"}'
|
||||
curl --url "https://api.example.com/auth" --headers '{"Authorization": "Bearer token"}'
|
||||
|
||||
Parameters:
|
||||
url: Request URL (required)
|
||||
method: GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS (default: GET)
|
||||
json: JSON body (will be serialized properly)
|
||||
data: Raw body data
|
||||
headers: Additional headers as JSON object
|
||||
timeout: Request timeout in seconds (default: 30)
|
||||
follow_redirects: Follow redirects (default: true)
|
||||
verbose: Include response headers (default: false)
|
||||
|
||||
Returns formatted response with status, headers (if verbose), and body.
|
||||
"""
|
||||
|
||||
@override
|
||||
@auto_timeout("curl")
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
url: str,
|
||||
method: str = "GET",
|
||||
json_body: Optional[str] = None,
|
||||
data: Optional[str] = None,
|
||||
headers: Optional[str] = None,
|
||||
timeout: int = 30,
|
||||
follow_redirects: bool = True,
|
||||
verbose: bool = False,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Make HTTP request."""
|
||||
tool_ctx = create_tool_context(ctx)
|
||||
await tool_ctx.set_tool_info(self.name)
|
||||
|
||||
# Build curl command
|
||||
cmd = ["curl", "-s", "-S"] # Silent but show errors
|
||||
|
||||
# Method
|
||||
if method != "GET":
|
||||
cmd.extend(["-X", method])
|
||||
|
||||
# Follow redirects
|
||||
if follow_redirects:
|
||||
cmd.append("-L")
|
||||
|
||||
# Timeout
|
||||
cmd.extend(["--max-time", str(timeout)])
|
||||
|
||||
# Include headers in output if verbose
|
||||
if verbose:
|
||||
cmd.append("-i")
|
||||
|
||||
# Headers
|
||||
default_headers = {}
|
||||
if headers:
|
||||
try:
|
||||
parsed_headers = json.loads(headers)
|
||||
if isinstance(parsed_headers, dict):
|
||||
default_headers.update(parsed_headers)
|
||||
except json.JSONDecodeError:
|
||||
return f"Error: Invalid headers JSON: {headers}"
|
||||
|
||||
# JSON body
|
||||
if json_body:
|
||||
default_headers.setdefault("Content-Type", "application/json")
|
||||
# Parse and re-serialize to ensure valid JSON
|
||||
try:
|
||||
parsed_json = json.loads(json_body)
|
||||
serialized = json.dumps(parsed_json)
|
||||
cmd.extend(["-d", serialized])
|
||||
except json.JSONDecodeError:
|
||||
return f"Error: Invalid JSON body: {json_body}"
|
||||
elif data:
|
||||
cmd.extend(["-d", data])
|
||||
|
||||
# Add all headers
|
||||
for key, value in default_headers.items():
|
||||
cmd.extend(["-H", f"{key}: {value}"])
|
||||
|
||||
# URL (must be last)
|
||||
cmd.append(url)
|
||||
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
|
||||
stdout, stderr = await asyncio.wait_for(
|
||||
proc.communicate(),
|
||||
timeout=timeout + 5, # Give curl time to timeout first
|
||||
)
|
||||
|
||||
output = stdout.decode("utf-8", errors="replace")
|
||||
errors = stderr.decode("utf-8", errors="replace")
|
||||
|
||||
if proc.returncode != 0:
|
||||
return f"curl failed (exit {proc.returncode}):\n{errors}\n{output}"
|
||||
|
||||
# Try to pretty-print JSON responses
|
||||
if not verbose: # Don't try to parse if headers included
|
||||
try:
|
||||
parsed = json.loads(output)
|
||||
return json.dumps(parsed, indent=2)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
return output
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
return f"Request timed out after {timeout}s"
|
||||
except FileNotFoundError:
|
||||
return "Error: curl not found. Install curl to use this tool."
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
def register(self, mcp_server: FastMCP) -> None:
|
||||
"""Register with MCP server."""
|
||||
tool_instance = self
|
||||
|
||||
@mcp_server.tool()
|
||||
async def curl(
|
||||
url: Annotated[str, Field(description="Request URL")],
|
||||
method: Method = "GET",
|
||||
json_body: Annotated[
|
||||
Optional[str], Field(description="JSON body (will be properly escaped)", alias="json")
|
||||
] = None,
|
||||
data: Annotated[Optional[str], Field(description="Raw body data")] = None,
|
||||
headers: Annotated[Optional[str], Field(description="Headers as JSON object")] = None,
|
||||
timeout: Annotated[int, Field(description="Timeout in seconds")] = 30,
|
||||
follow_redirects: Annotated[bool, Field(description="Follow redirects")] = True,
|
||||
verbose: Annotated[bool, Field(description="Include response headers")] = False,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
"""HTTP client - curl without shell escaping issues.
|
||||
|
||||
Make HTTP requests with proper JSON handling.
|
||||
No more shell quoting nightmares!
|
||||
"""
|
||||
return await tool_instance.call(
|
||||
ctx,
|
||||
url=url,
|
||||
method=method,
|
||||
json_body=json_body,
|
||||
data=data,
|
||||
headers=headers,
|
||||
timeout=timeout,
|
||||
follow_redirects=follow_redirects,
|
||||
verbose=verbose,
|
||||
)
|
||||
@@ -0,0 +1,167 @@
|
||||
"""JQ tool - JSON processing without shell escaping issues.
|
||||
|
||||
Provides a clean interface for jq queries, avoiding shell quoting problems.
|
||||
"""
|
||||
|
||||
import json
|
||||
import asyncio
|
||||
from typing import Optional, Annotated, final, override
|
||||
|
||||
from pydantic import Field
|
||||
from mcp.server import FastMCP
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
|
||||
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
|
||||
|
||||
|
||||
@final
|
||||
class JqTool(BaseTool):
|
||||
"""JQ tool - JSON processing without shell escaping nightmares.
|
||||
|
||||
Clean interface for jq queries with proper handling.
|
||||
"""
|
||||
|
||||
name = "jq"
|
||||
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
return """JSON processor - jq without shell escaping issues.
|
||||
|
||||
Examples:
|
||||
jq --filter ".result.data" --input '{"result": {"data": [1,2,3]}}'
|
||||
jq --filter ".[] | select(.active)" --file data.json
|
||||
jq --filter "keys" --input '{"a": 1, "b": 2}'
|
||||
jq --filter '.checks | to_entries[] | select(.value.error != null)' --file health.json
|
||||
|
||||
Parameters:
|
||||
filter: jq filter expression (required)
|
||||
input: JSON input as string
|
||||
file: Path to JSON file (alternative to input)
|
||||
raw: Output raw strings without quotes (default: false)
|
||||
compact: Compact output (default: false)
|
||||
slurp: Read entire input as single array (default: false)
|
||||
sort_keys: Sort object keys (default: false)
|
||||
|
||||
The filter is passed directly to jq without shell interpretation,
|
||||
so you don't need to escape special characters like ! or |
|
||||
"""
|
||||
|
||||
@override
|
||||
@auto_timeout("jq")
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
filter: str,
|
||||
input: Optional[str] = None,
|
||||
file: Optional[str] = None,
|
||||
raw: bool = False,
|
||||
compact: bool = False,
|
||||
slurp: bool = False,
|
||||
sort_keys: bool = False,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Process JSON with jq."""
|
||||
tool_ctx = create_tool_context(ctx)
|
||||
await tool_ctx.set_tool_info(self.name)
|
||||
|
||||
if not input and not file:
|
||||
return "Error: Either 'input' or 'file' is required"
|
||||
|
||||
# Build jq command
|
||||
cmd = ["jq"]
|
||||
|
||||
# Options
|
||||
if raw:
|
||||
cmd.append("-r")
|
||||
if compact:
|
||||
cmd.append("-c")
|
||||
if slurp:
|
||||
cmd.append("-s")
|
||||
if sort_keys:
|
||||
cmd.append("-S")
|
||||
|
||||
# Filter (passed as argument, not through shell)
|
||||
cmd.append(filter)
|
||||
|
||||
# File input
|
||||
if file:
|
||||
cmd.append(file)
|
||||
|
||||
try:
|
||||
if input:
|
||||
# Validate input is valid JSON first
|
||||
try:
|
||||
json.loads(input)
|
||||
except json.JSONDecodeError as e:
|
||||
return f"Error: Invalid JSON input: {e}"
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await asyncio.wait_for(
|
||||
proc.communicate(input=input.encode("utf-8")),
|
||||
timeout=30,
|
||||
)
|
||||
else:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await asyncio.wait_for(
|
||||
proc.communicate(),
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
output = stdout.decode("utf-8", errors="replace")
|
||||
errors = stderr.decode("utf-8", errors="replace")
|
||||
|
||||
if proc.returncode != 0:
|
||||
# Provide helpful error message
|
||||
if "syntax error" in errors.lower():
|
||||
return f"jq syntax error in filter:\n {filter}\n\nError: {errors}"
|
||||
return f"jq failed (exit {proc.returncode}):\n{errors}"
|
||||
|
||||
return output.rstrip()
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
return "jq timed out after 30s"
|
||||
except FileNotFoundError:
|
||||
return "Error: jq not found. Install jq: brew install jq"
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
def register(self, mcp_server: FastMCP) -> None:
|
||||
"""Register with MCP server."""
|
||||
tool_instance = self
|
||||
|
||||
@mcp_server.tool()
|
||||
async def jq(
|
||||
filter: Annotated[str, Field(description="jq filter expression")],
|
||||
input: Annotated[Optional[str], Field(description="JSON input string")] = None,
|
||||
file: Annotated[Optional[str], Field(description="Path to JSON file")] = None,
|
||||
raw: Annotated[bool, Field(description="Output raw strings")] = False,
|
||||
compact: Annotated[bool, Field(description="Compact output")] = False,
|
||||
slurp: Annotated[bool, Field(description="Read as single array")] = False,
|
||||
sort_keys: Annotated[bool, Field(description="Sort object keys")] = False,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
"""JSON processor - jq without shell escaping issues.
|
||||
|
||||
Process JSON with jq filters. No shell quoting nightmares!
|
||||
Supports all standard jq operations.
|
||||
"""
|
||||
return await tool_instance.call(
|
||||
ctx,
|
||||
filter=filter,
|
||||
input=input,
|
||||
file=file,
|
||||
raw=raw,
|
||||
compact=compact,
|
||||
slurp=slurp,
|
||||
sort_keys=sort_keys,
|
||||
)
|
||||
@@ -0,0 +1,224 @@
|
||||
"""Wget tool - reliable file and site downloads.
|
||||
|
||||
Provides a clean interface for wget with proper handling of recursive downloads.
|
||||
"""
|
||||
|
||||
import os
|
||||
import asyncio
|
||||
from typing import Optional, Annotated, final, override
|
||||
|
||||
from pydantic import Field
|
||||
from mcp.server import FastMCP
|
||||
from mcp.server.fastmcp import Context as MCPContext
|
||||
|
||||
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
|
||||
|
||||
|
||||
@final
|
||||
class WgetTool(BaseTool):
|
||||
"""Wget tool - reliable downloads without shell escaping issues.
|
||||
|
||||
Supports single file downloads and full site mirroring.
|
||||
"""
|
||||
|
||||
name = "wget"
|
||||
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
return """Download files and mirror websites reliably.
|
||||
|
||||
Examples:
|
||||
wget --url "https://example.com/file.zip"
|
||||
wget --url "https://example.com/file.zip" --output "/tmp/myfile.zip"
|
||||
wget --url "https://docs.example.com" --mirror true
|
||||
wget --url "https://docs.example.com" --recursive true --depth 2
|
||||
wget --url "https://example.com" --recursive true --accept "*.pdf,*.doc"
|
||||
|
||||
Parameters:
|
||||
url: URL to download (required)
|
||||
output: Output file or directory
|
||||
mirror: Mirror site for offline viewing (sets recursive + timestamps)
|
||||
recursive: Download recursively
|
||||
depth: Maximum recursion depth (default: 5 for recursive)
|
||||
accept: Accept only files matching pattern (e.g., "*.pdf,*.html")
|
||||
reject: Reject files matching pattern
|
||||
domains: Limit to specific domains
|
||||
no_parent: Don't ascend to parent directory
|
||||
continue_download: Continue partial downloads
|
||||
timeout: Timeout in seconds (default: 300 for mirrors, 60 for files)
|
||||
quiet: Suppress output
|
||||
|
||||
Returns status and download summary.
|
||||
"""
|
||||
|
||||
@override
|
||||
@auto_timeout("wget")
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
url: str,
|
||||
output: Optional[str] = None,
|
||||
mirror: bool = False,
|
||||
recursive: bool = False,
|
||||
depth: Optional[int] = None,
|
||||
accept: Optional[str] = None,
|
||||
reject: Optional[str] = None,
|
||||
domains: Optional[str] = None,
|
||||
no_parent: bool = True,
|
||||
continue_download: bool = False,
|
||||
timeout: Optional[int] = None,
|
||||
quiet: bool = False,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Download files or mirror websites."""
|
||||
tool_ctx = create_tool_context(ctx)
|
||||
await tool_ctx.set_tool_info(self.name)
|
||||
|
||||
# Build wget command
|
||||
cmd = ["wget"]
|
||||
|
||||
# Mirror mode (most common for site downloads)
|
||||
if mirror:
|
||||
cmd.extend([
|
||||
"--mirror", # Turn on mirroring
|
||||
"--convert-links", # Convert links for offline viewing
|
||||
"--adjust-extension", # Add .html extension
|
||||
"--page-requisites", # Get all assets (css, js, images)
|
||||
"--no-host-directories", # Don't create host directory
|
||||
])
|
||||
recursive = True # Mirror implies recursive
|
||||
|
||||
# Recursive download
|
||||
if recursive:
|
||||
cmd.append("-r")
|
||||
if depth is not None:
|
||||
cmd.extend(["-l", str(depth)])
|
||||
elif not mirror:
|
||||
cmd.extend(["-l", "5"]) # Default depth limit
|
||||
|
||||
# No parent directory traversal (safety)
|
||||
if no_parent:
|
||||
cmd.append("--no-parent")
|
||||
|
||||
# Continue partial downloads
|
||||
if continue_download:
|
||||
cmd.append("-c")
|
||||
|
||||
# Accept/reject patterns
|
||||
if accept:
|
||||
cmd.extend(["-A", accept])
|
||||
if reject:
|
||||
cmd.extend(["-R", reject])
|
||||
|
||||
# Domain restrictions
|
||||
if domains:
|
||||
cmd.extend(["-D", domains])
|
||||
|
||||
# Output location
|
||||
if output:
|
||||
if recursive or mirror:
|
||||
cmd.extend(["-P", output]) # Directory for recursive
|
||||
else:
|
||||
cmd.extend(["-O", output]) # File for single download
|
||||
|
||||
# Timeout
|
||||
actual_timeout = timeout or (300 if (mirror or recursive) else 60)
|
||||
cmd.extend(["--timeout", str(min(actual_timeout, 30))]) # Per-request timeout
|
||||
cmd.extend(["--tries", "3"]) # Retry count
|
||||
cmd.extend(["--waitretry", "1"]) # Wait between retries
|
||||
|
||||
# Progress
|
||||
if quiet:
|
||||
cmd.append("-q")
|
||||
else:
|
||||
cmd.append("--progress=dot:mega") # Show progress for large files
|
||||
|
||||
# User agent (avoid blocks)
|
||||
cmd.extend([
|
||||
"--user-agent",
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
|
||||
])
|
||||
|
||||
# URL (last)
|
||||
cmd.append(url)
|
||||
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
cwd=output if (output and os.path.isdir(output)) else None,
|
||||
)
|
||||
|
||||
stdout, stderr = await asyncio.wait_for(
|
||||
proc.communicate(),
|
||||
timeout=actual_timeout + 30,
|
||||
)
|
||||
|
||||
output_text = stdout.decode("utf-8", errors="replace")
|
||||
errors = stderr.decode("utf-8", errors="replace")
|
||||
|
||||
# Combine stdout and stderr (wget uses stderr for progress)
|
||||
combined = f"{errors}\n{output_text}".strip()
|
||||
|
||||
if proc.returncode != 0:
|
||||
return f"wget failed (exit {proc.returncode}):\n{combined}"
|
||||
|
||||
# Summarize success
|
||||
lines = combined.split("\n")
|
||||
summary_lines = [l for l in lines if any(x in l.lower() for x in [
|
||||
"saved", "downloaded", "finished", "total", "retrieved"
|
||||
])]
|
||||
|
||||
if summary_lines:
|
||||
return "Download complete:\n" + "\n".join(summary_lines[-5:])
|
||||
return f"Download complete:\n{combined[-500:]}"
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
return f"Download timed out after {actual_timeout}s"
|
||||
except FileNotFoundError:
|
||||
return "Error: wget not found. Install wget: brew install wget"
|
||||
except Exception as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
def register(self, mcp_server: FastMCP) -> None:
|
||||
"""Register with MCP server."""
|
||||
tool_instance = self
|
||||
|
||||
@mcp_server.tool()
|
||||
async def wget(
|
||||
url: Annotated[str, Field(description="URL to download")],
|
||||
output: Annotated[Optional[str], Field(description="Output file or directory")] = None,
|
||||
mirror: Annotated[bool, Field(description="Mirror site for offline viewing")] = False,
|
||||
recursive: Annotated[bool, Field(description="Download recursively")] = False,
|
||||
depth: Annotated[Optional[int], Field(description="Max recursion depth")] = None,
|
||||
accept: Annotated[Optional[str], Field(description="Accept pattern (e.g. '*.pdf')")] = None,
|
||||
reject: Annotated[Optional[str], Field(description="Reject pattern")] = None,
|
||||
domains: Annotated[Optional[str], Field(description="Limit to domains")] = None,
|
||||
no_parent: Annotated[bool, Field(description="Don't go to parent directories")] = True,
|
||||
continue_download: Annotated[bool, Field(description="Continue partial downloads")] = False,
|
||||
timeout: Annotated[Optional[int], Field(description="Timeout in seconds")] = None,
|
||||
quiet: Annotated[bool, Field(description="Suppress output")] = False,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
"""Download files and mirror websites reliably.
|
||||
|
||||
Supports single file downloads and full site mirroring.
|
||||
Handles retries, timeouts, and link conversion automatically.
|
||||
"""
|
||||
return await tool_instance.call(
|
||||
ctx,
|
||||
url=url,
|
||||
output=output,
|
||||
mirror=mirror,
|
||||
recursive=recursive,
|
||||
depth=depth,
|
||||
accept=accept,
|
||||
reject=reject,
|
||||
domains=domains,
|
||||
no_parent=no_parent,
|
||||
continue_download=continue_download,
|
||||
timeout=timeout,
|
||||
quiet=quiet,
|
||||
)
|
||||
Reference in New Issue
Block a user