#!/usr/bin/env python3
"""
Katana MCP Server
MCP server để tích hợp Katana web crawler với Claude
"""
import asyncio
import json
import subprocess
import sys
from pathlib import Path
from typing import Any, Optional, List, Tuple
import tempfile
import os
import urllib.parse
import requests
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import mcp.types as types
# Khởi tạo MCP server
app = Server("katana-server")
def check_katana_installed() -> bool:
"""Kiểm tra xem Katana đã được cài đặt chưa"""
try:
result = subprocess.run(
["katana", "--version"],
capture_output=True,
text=True,
timeout=5
)
return result.returncode == 0
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
@app.list_tools()
async def list_tools() -> List[Tool]:
"""Liệt kê các tools có sẵn"""
return [
Tool(
name="katana_crawl",
description="Chạy Katana để crawl một URL hoặc danh sách URLs. Trả về các URLs, endpoints, và thông tin được phát hiện.",
inputSchema={
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "URL mục tiêu để crawl (ví dụ: https://example.com)"
},
"urls": {
"type": "array",
"items": {"type": "string"},
"description": "Danh sách URLs để crawl (tùy chọn, thay thế cho url đơn)"
},
"depth": {
"type": "integer",
"description": "Độ sâu crawl (default: 5)",
"default": 5
},
"concurrency": {
"type": "integer",
"description": "Số lượng requests đồng thời (default: 10)",
"default": 10
},
"scope": {
"type": "string",
"description": "Scope cho crawl: 'subs' (subdomain), 'folder' (folder), 'domain' (domain), 'any' (any)",
"enum": ["subs", "folder", "domain", "any"],
"default": "folder"
},
"headless": {
"type": "boolean",
"description": "Sử dụng headless browser để crawl JavaScript (default: false)",
"default": True
},
"js_crawl": {
"type": "boolean",
"description": "Crawl JavaScript files để tìm endpoints (default: false)",
"default": True
},
"form_fill": {
"type": "string",
"description": "Tự động điền form với giá trị (ví dụ: 'username=admin&password=admin')",
"default": ""
},
"output_format": {
"type": "string",
"description": "Định dạng output: 'json' hoặc 'text'",
"enum": ["json", "text"],
"default": "json"
},
"store_response": {
"type": "boolean",
"description": "Lưu response body (default: false)",
"default": False
},
"store_response_dir": {
"type": "string",
"description": "Thư mục để lưu responses (chỉ khi store_response=true)"
},
"filter_status": {
"type": "string",
"description": "Lọc theo status code (ví dụ: '200,301,302')"
},
"filter_regex": {
"type": "string",
"description": "Lọc URLs theo regex pattern"
},
"exclude_regex": {
"type": "string",
"description": "Loại trừ URLs theo regex pattern"
},
"exclude_file_extensions": {
"type": "string",
"description": "Loại trừ các file extensions (ví dụ: 'woff,css,png,svg,jpg')",
"default": "woff,css,png,svg,jpg,woff2,jpeg,gif,svg"
},
"js_links": {
"type": "boolean",
"description": "Crawl JavaScript links (default: true)",
"default": True
},
"xhr": {
"type": "boolean",
"description": "Crawl XHR requests (default: true)",
"default": True
},
"keep_form_data": {
"type": "string",
"description": "Giữ form data (default: 'all')",
"default": "all"
},
"form_extract": {
"type": "boolean",
"description": "Trích xuất form (default: true)",
"default": True
},
"form_submit": {
"type": "string",
"description": "Gửi form (default: 'dn')",
"default": "dn"
},
"filter_format": {
"type": "string",
"description": "Định dạng filter (default: 'qurl')",
"default": "qurl"
}
}
}
),
Tool(
name="katana_crawl_from_file",
description="Chạy Katana crawl từ file chứa danh sách URLs",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Đường dẫn đến file chứa URLs (mỗi URL một dòng)"
},
"depth": {
"type": "integer",
"description": "Độ sâu crawl (default: 5)",
"default": 5
},
"concurrency": {
"type": "integer",
"description": "Số lượng requests đồng thời (default: 10)",
"default": 10
},
"scope": {
"type": "string",
"description": "Scope cho crawl",
"enum": ["subs", "folder", "domain", "any"],
"default": "folder"
},
"output_format": {
"type": "string",
"description": "Định dạng output: 'json' hoặc 'text'",
"enum": ["json", "text"],
"default": "json"
}
},
"required": ["file_path"]
}
),
Tool(
name="katana_check_version",
description="Kiểm tra phiên bản Katana đã cài đặt",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="katana_download_js",
description=(
"Chạy Katana để crawl và lấy toàn bộ các URL file JavaScript (.js), "
"sau đó tải toàn bộ file JS đó về một thư mục trên máy."
),
inputSchema={
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "URL gốc để crawl và thu thập các file JS (ví dụ: https://example.com)"
},
"urls": {
"type": "array",
"items": {"type": "string"},
"description": "Danh sách URLs để crawl (thay thế cho url đơn, mỗi phần tử là một URL)"
},
"file_path": {
"type": "string",
"description": "Đường dẫn đến file chứa danh sách URLs (mỗi dòng một URL)"
},
"depth": {
"type": "integer",
"description": "Độ sâu crawl (default: 5)",
"default": 5
},
"scope": {
"type": "string",
"description": "Scope cho crawl: 'subs', 'folder', 'domain', 'any'",
"enum": ["subs", "folder", "domain", "any"],
"default": "folder"
},
"concurrency": {
"type": "integer",
"description": "Số lượng requests đồng thời của katana (default: 10)",
"default": 10
},
"output_dir": {
"type": "string",
"description": "Thư mục để lưu các file JS tải về (default: ./katana_js_downloads)",
"default": "katana_js_downloads"
},
"timeout": {
"type": "integer",
"description": "Timeout (giây) cho mỗi request tải JS (default: 20)",
"default": 20
},
"max_files": {
"type": "integer",
"description": "Giới hạn số lượng file JS tối đa cần tải (default: 200, 0 = không giới hạn)",
"default": 200
}
}
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> List[types.TextContent]:
"""Xử lý các tool calls"""
if name == "katana_check_version":
if not check_katana_installed():
return [
TextContent(
type="text",
text="❌ Katana chưa được cài đặt. Vui lòng cài đặt bằng lệnh:\ngo install github.com/projectdiscovery/katana/cmd/katana@latest"
)
]
try:
result = subprocess.run(
["katana", "--version"],
capture_output=True,
text=True,
timeout=5
)
version_info = result.stdout.strip() if result.returncode == 0 else result.stderr.strip()
return [TextContent(type="text", text=f"✅ Katana version:\n{version_info}")]
except Exception as e:
return [TextContent(type="text", text=f"❌ Lỗi khi kiểm tra version: {str(e)}")]
elif name == "katana_crawl":
return await run_katana_crawl(arguments)
elif name == "katana_crawl_from_file":
return await run_katana_crawl_from_file(arguments)
elif name == "katana_download_js":
return await run_katana_download_js(arguments)
else:
return [TextContent(type="text", text=f"❌ Tool không tồn tại: {name}")]
async def run_katana_crawl(arguments: Any) -> List[types.TextContent]:
"""Chạy Katana crawl với các tham số"""
if not check_katana_installed():
return [
TextContent(
type="text",
text="❌ Katana chưa được cài đặt. Vui lòng cài đặt bằng lệnh:\ngo install github.com/projectdiscovery/katana/cmd/katana@latest"
)
]
# Xây dựng lệnh Katana
cmd = ["katana"]
# URL hoặc URLs
temp_file = None
if "url" in arguments and arguments["url"]:
cmd.extend(["-u", arguments["url"]])
elif "urls" in arguments and arguments["urls"]:
# Tạo file tạm với danh sách URLs
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f:
for url in arguments["urls"]:
f.write(f"{url}\n")
temp_file = f.name
cmd.extend(["-list", temp_file])
else:
return [TextContent(type="text", text="❌ Cần cung cấp 'url' hoặc 'urls'")]
# Các tùy chọn
# Depth (default: 5)
depth = arguments.get("depth", 5)
cmd.extend(["-d", str(depth)])
if "concurrency" in arguments:
cmd.extend(["-c", str(arguments["concurrency"])])
if "scope" in arguments:
scope_map = {
"subs": "subs",
"folder": "folder",
"domain": "domain",
"any": "any"
}
cmd.extend(["-s", scope_map.get(arguments["scope"], "folder")])
# Filter format (default: qurl)
filter_format = arguments.get("filter_format", "qurl")
cmd.extend(["-f", filter_format])
# JavaScript crawl (default: true)
if arguments.get("js_crawl", True):
cmd.append("-jc")
# JavaScript links (default: true)
if arguments.get("js_links", True):
cmd.append("-jsl")
# Headless (default: true)
if arguments.get("headless", True):
cmd.append("-hl")
# XHR requests (default: true)
if arguments.get("xhr", True):
cmd.append("-xhr")
# Keep form data (default: all)
keep_form_data = arguments.get("keep_form_data", "all")
if keep_form_data:
cmd.extend(["-kf", keep_form_data])
# Form extraction (default: true)
if arguments.get("form_extract", True):
cmd.append("-fx")
# Form submission (default: dn)
form_submit = arguments.get("form_submit", "dn")
if form_submit:
cmd.extend(["-fs", form_submit])
if "form_fill" in arguments and arguments["form_fill"]:
cmd.extend(["-form-fill", arguments["form_fill"]])
if arguments.get("store_response", False):
cmd.append("-store-response")
if "store_response_dir" in arguments and arguments["store_response_dir"]:
cmd.extend(["-store-response-dir", arguments["store_response_dir"]])
if "filter_status" in arguments and arguments["filter_status"]:
cmd.extend(["-fc", arguments["filter_status"]])
if "filter_regex" in arguments and arguments["filter_regex"]:
# Note: -f is already used for filter_format, so filter_regex might conflict
# Using -fr for filter regex if needed
cmd.extend(["-fr", arguments["filter_regex"]])
# Exclude file extensions (default: woff,css,png,svg,jpg,woff2,jpeg,gif,svg)
exclude_file_extensions = arguments.get("exclude_file_extensions", "woff,css,png,svg,jpg,woff2,jpeg,gif,svg")
if exclude_file_extensions:
cmd.extend(["-ef", exclude_file_extensions])
if "exclude_regex" in arguments and arguments["exclude_regex"]:
# Use -er for exclude regex to avoid conflict with -ef
cmd.extend(["-er", arguments["exclude_regex"]])
# Output format
output_format = arguments.get("output_format", "json")
if output_format == "json":
cmd.append("-json")
try:
# Chạy Katana
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=300)
# Xóa file tạm nếu có
if "urls" in arguments and arguments["urls"]:
try:
os.unlink(temp_file)
except:
pass
if process.returncode != 0:
error_msg = stderr.decode('utf-8', errors='ignore') if stderr else "Unknown error"
return [
TextContent(
type="text",
text=f"❌ Lỗi khi chạy Katana:\n{error_msg}\n\nCommand: {' '.join(cmd)}"
)
]
output = stdout.decode('utf-8', errors='ignore')
# Parse JSON output nếu cần
if output_format == "json" and output.strip():
try:
results = []
for line in output.strip().split('\n'):
if line.strip():
results.append(json.loads(line))
# Format kết quả
formatted_output = f"✅ Crawl hoàn thành! Tìm thấy {len(results)} URLs/endpoints:\n\n"
for i, result in enumerate(results[:50], 1): # Giới hạn 50 kết quả đầu
formatted_output += f"{i}. {result.get('url', 'N/A')}\n"
if 'status_code' in result:
formatted_output += f" Status: {result['status_code']}\n"
if 'title' in result:
formatted_output += f" Title: {result['title']}\n"
formatted_output += "\n"
if len(results) > 50:
formatted_output += f"... và {len(results) - 50} kết quả khác\n"
# Thêm raw JSON cho phân tích chi tiết
formatted_output += f"\n--- Raw JSON Output (first 10) ---\n"
formatted_output += json.dumps(results[:10], indent=2, ensure_ascii=False)
return [TextContent(type="text", text=formatted_output)]
except json.JSONDecodeError:
# Nếu không parse được JSON, trả về raw output
return [TextContent(type="text", text=f"✅ Crawl hoàn thành:\n\n{output}")]
return [TextContent(type="text", text=f"✅ Crawl hoàn thành:\n\n{output}")]
except asyncio.TimeoutError:
return [TextContent(type="text", text="❌ Timeout: Crawl mất quá nhiều thời gian (>5 phút)")]
except Exception as e:
return [TextContent(type="text", text=f"❌ Lỗi: {str(e)}")]
async def run_katana_crawl_from_file(arguments: Any) -> List[types.TextContent]:
"""Chạy Katana crawl từ file"""
if not check_katana_installed():
return [
TextContent(
type="text",
text="❌ Katana chưa được cài đặt. Vui lòng cài đặt bằng lệnh:\ngo install github.com/projectdiscovery/katana/cmd/katana@latest"
)
]
file_path = arguments.get("file_path")
if not file_path or not Path(file_path).exists():
return [TextContent(type="text", text=f"❌ File không tồn tại: {file_path}")]
cmd = ["katana", "-list", file_path]
# Depth (default: 5)
depth = arguments.get("depth", 5)
cmd.extend(["-d", str(depth)])
if "concurrency" in arguments:
cmd.extend(["-c", str(arguments["concurrency"])])
if "scope" in arguments:
scope_map = {
"subs": "subs",
"folder": "folder",
"domain": "domain",
"any": "any"
}
cmd.extend(["-s", scope_map.get(arguments["scope"], "folder")])
# Filter format (default: qurl)
filter_format = arguments.get("filter_format", "qurl")
cmd.extend(["-f", filter_format])
# JavaScript crawl (default: true)
if arguments.get("js_crawl", True):
cmd.append("-jc")
# JavaScript links (default: true)
if arguments.get("js_links", True):
cmd.append("-jsl")
# Headless (default: true)
if arguments.get("headless", True):
cmd.append("-hl")
# XHR requests (default: true)
if arguments.get("xhr", True):
cmd.append("-xhr")
# Keep form data (default: all)
keep_form_data = arguments.get("keep_form_data", "all")
if keep_form_data:
cmd.extend(["-kf", keep_form_data])
# Form extraction (default: true)
if arguments.get("form_extract", True):
cmd.append("-fx")
# Form submission (default: dn)
form_submit = arguments.get("form_submit", "dn")
if form_submit:
cmd.extend(["-fs", form_submit])
# Exclude file extensions (default: woff,css,png,svg,jpg,woff2,jpeg,gif,svg)
exclude_file_extensions = arguments.get("exclude_file_extensions", "woff,css,png,svg,jpg,woff2,jpeg,gif,svg")
if exclude_file_extensions:
cmd.extend(["-ef", exclude_file_extensions])
output_format = arguments.get("output_format", "json")
if output_format == "json":
cmd.append("-json")
try:
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=300)
if process.returncode != 0:
error_msg = stderr.decode('utf-8', errors='ignore') if stderr else "Unknown error"
return [
TextContent(
type="text",
text=f"❌ Lỗi khi chạy Katana:\n{error_msg}"
)
]
output = stdout.decode('utf-8', errors='ignore')
if output_format == "json" and output.strip():
try:
results = []
for line in output.strip().split('\n'):
if line.strip():
results.append(json.loads(line))
formatted_output = f"✅ Crawl hoàn thành từ file {file_path}! Tìm thấy {len(results)} URLs/endpoints:\n\n"
for i, result in enumerate(results[:50], 1):
formatted_output += f"{i}. {result.get('url', 'N/A')}\n"
if 'status_code' in result:
formatted_output += f" Status: {result['status_code']}\n"
formatted_output += "\n"
if len(results) > 50:
formatted_output += f"... và {len(results) - 50} kết quả khác\n"
return [TextContent(type="text", text=formatted_output)]
except json.JSONDecodeError:
return [TextContent(type="text", text=f"✅ Crawl hoàn thành:\n\n{output}")]
return [TextContent(type="text", text=f"✅ Crawl hoàn thành:\n\n{output}")]
except asyncio.TimeoutError:
return [TextContent(type="text", text="❌ Timeout: Crawl mất quá nhiều thời gian")]
except Exception as e:
return [TextContent(type="text", text=f"❌ Lỗi: {str(e)}")]
def _is_js_url(url: str) -> bool:
"""Kiểm tra URL có phải là file JS không (dựa trên path và query)."""
try:
parsed = urllib.parse.urlparse(url)
path = parsed.path or ""
if path.lower().endswith(".js"):
return True
# Một số trường hợp .js trong query (hiếm) có thể xử lý thêm nếu cần
return False
except Exception:
return False
def _build_js_filename(url: str, index: int) -> str:
"""Sinh tên file JS hợp lệ từ URL."""
parsed = urllib.parse.urlparse(url)
name = os.path.basename(parsed.path)
if not name or "." not in name:
name = f"script_{index}.js"
# Loại bỏ ký tự lạ
safe_name = "".join(c for c in name if c.isalnum() or c in ("-", "_", "."))
if not safe_name.lower().endswith(".js"):
safe_name = f"{safe_name}.js"
return safe_name or f"script_{index}.js"
async def _download_js_file(url: str, output_path: Path, timeout: int) -> Tuple[bool, str]:
"""Tải một file JS (chạy trong thread pool để không block event loop)."""
def _do_request() -> Tuple[bool, str]:
try:
resp = requests.get(url, timeout=timeout)
if resp.status_code == 200:
output_path.write_bytes(resp.content)
return True, ""
return False, f"HTTP {resp.status_code}"
except Exception as e:
return False, str(e)
return await asyncio.to_thread(_do_request)
async def run_katana_download_js(arguments: Any) -> List[types.TextContent]:
"""Chạy Katana, lọc toàn bộ URL .js và tải về thư mục chỉ định."""
if not check_katana_installed():
return [
TextContent(
type="text",
text=(
"❌ Katana chưa được cài đặt. Vui lòng cài đặt bằng lệnh:\n"
"go install github.com/projectdiscovery/katana/cmd/katana@latest"
),
)
]
url = arguments.get("url")
urls = arguments.get("urls") or []
file_path = arguments.get("file_path")
# Xác định nguồn input: url đơn, danh sách urls, hoặc file_path
source_desc = ""
cmd: List[str]
temp_file: Optional[str] = None
if url:
source_desc = f"URL đơn: {url}"
cmd = ["katana", "-u", url]
elif urls:
# Ghi danh sách URLs vào file tạm và dùng -list
if not isinstance(urls, list):
return [TextContent(type="text", text="❌ Tham số 'urls' phải là một danh sách chuỗi")]
if not urls:
return [TextContent(type="text", text="❌ Danh sách 'urls' trống")]
with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
for u in urls:
if isinstance(u, str) and u.strip():
f.write(u.strip() + "\n")
temp_file = f.name
source_desc = f"Danh sách URLs (file tạm: {temp_file})"
cmd = ["katana", "-list", temp_file]
elif file_path:
p = Path(file_path)
if not p.exists():
return [TextContent(type="text", text=f"❌ File không tồn tại: {file_path}")]
source_desc = f"File URL list: {p.resolve()}"
cmd = ["katana", "-list", str(p)]
else:
return [
TextContent(
type="text",
text="❌ Cần cung cấp một trong các tham số: 'url', 'urls' hoặc 'file_path' để crawl JS",
)
]
depth = arguments.get("depth", 5)
scope = arguments.get("scope", "folder")
concurrency = arguments.get("concurrency", 10)
output_dir_arg = arguments.get("output_dir", "katana_js_downloads")
timeout = int(arguments.get("timeout", 20))
max_files = int(arguments.get("max_files", 200))
output_dir = Path(output_dir_arg).expanduser().resolve()
output_dir.mkdir(parents=True, exist_ok=True)
# Xây dựng lệnh Katana tập trung vào JS, luôn trả về JSON để parse
cmd.extend(["-d", str(depth)])
cmd.extend(["-c", str(concurrency)])
scope_map = {
"subs": "subs",
"folder": "folder",
"domain": "domain",
"any": "any",
}
cmd.extend(["-s", scope_map.get(scope, "folder")])
# Bật tối đa khả năng phát hiện JS
cmd.append("-jc") # JS crawl
cmd.append("-jsl") # JS links
cmd.append("-hl") # headless
cmd.append("-xhr") # XHR
# Không dùng exclude_file_extensions để tránh loại bỏ JS
# Dùng JSON output để tự lọc JS trong Python
cmd.append("-json")
try:
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=300)
# Dọn file tạm nếu có
if temp_file is not None:
try:
os.unlink(temp_file)
except Exception:
pass
if process.returncode != 0:
error_msg = stderr.decode("utf-8", errors="ignore") if stderr else "Unknown error"
return [
TextContent(
type="text",
text=f"❌ Lỗi khi chạy Katana (download JS):\n{error_msg}\n\nCommand: {' '.join(cmd)}",
)
]
raw_output = stdout.decode("utf-8", errors="ignore")
if not raw_output.strip():
return [TextContent(type="text", text="❌ Katana không trả về kết quả nào")]
results = []
for line in raw_output.strip().split("\n"):
line = line.strip()
if not line:
continue
try:
results.append(json.loads(line))
except json.JSONDecodeError:
# Bỏ qua dòng hỏng, tiếp tục các dòng khác
continue
js_urls_set = set()
for item in results:
u = item.get("url") or item.get("request", {}).get("url")
if not u:
continue
if _is_js_url(u):
js_urls_set.add(u)
js_urls = sorted(js_urls_set)
total_js = len(js_urls)
if total_js == 0:
return [
TextContent(
type="text",
text=(
"✅ Katana đã chạy nhưng không phát hiện URL nào kết thúc bằng .js.\n"
f"Command: {' '.join(cmd)}"
),
)
]
if max_files > 0 and total_js > max_files:
js_urls = js_urls[:max_files]
downloaded: List[Path] = []
failed: List[Tuple[str, str]] = []
for idx, js_url in enumerate(js_urls, start=1):
filename = _build_js_filename(js_url, idx)
dest_path = output_dir / filename
ok, err = await _download_js_file(js_url, dest_path, timeout=timeout)
if ok:
downloaded.append(dest_path)
else:
failed.append((js_url, err))
summary = []
summary.append("✅ Hoàn thành crawl & download JS bằng Katana")
summary.append(f"- Nguồn: {source_desc}")
summary.append(f"- Tổng số URL .js phát hiện: {total_js}")
summary.append(f"- Số file cố gắng tải (sau khi áp dụng max_files={max_files}): {len(js_urls)}")
summary.append(f"- Số file tải thành công: {len(downloaded)}")
summary.append(f"- Số file lỗi: {len(failed)}")
summary.append(f"- Thư mục lưu file: {output_dir}")
if downloaded:
summary.append("\nMột số file JS đã tải:")
for p in downloaded[:10]:
summary.append(f" - {p}")
if len(downloaded) > 10:
summary.append(f" ... và {len(downloaded) - 10} file khác")
if failed:
summary.append("\nMột số lỗi khi tải JS (tối đa 5):")
for u, err in failed[:5]:
summary.append(f" - {u} -> {err}")
return [TextContent(type="text", text="\n".join(summary))]
except asyncio.TimeoutError:
return [TextContent(type="text", text="❌ Timeout: Katana (download JS) chạy quá 5 phút")]
except Exception as e:
return [TextContent(type="text", text=f"❌ Lỗi (download JS): {str(e)}")]
async def main():
"""Main entry point"""
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
