"""Magnet link search tool.
Searches for torrent magnet links via a local API service.
"""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "search_magnet_links"
TOOL_DESCRIPTION = (
"Search for magnet links using a local torrent search API. "
"Returns torrent metadata, file sizes, seeders, and magnet links. "
"Requires a local API service on http://127.0.0.1:3001."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query string (keywords, titles, categories, etc.).",
},
},
"required": ["query"],
}
[docs]
async def run(query: str, ctx: ToolContext | None = None) -> str:
"""Execute this tool and return the result.
Args:
query (str): Search query or input string.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
if not query or not query.strip():
return json.dumps({"error": "Search query cannot be empty"})
url = f"http://127.0.0.1:3001/api/all/{query.strip()}/1"
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=120)) as response:
if response.status == 200:
data = await response.json()
return json.dumps(data, indent=4)
else:
return json.dumps({"error": f"HTTP {response.status}"})
except aiohttp.ClientError as e:
return json.dumps({"error": f"Network error: {e}"})
except Exception as e:
return json.dumps({"error": f"Unexpected error: {e}"})