| """Tool client — provides the local web_search tool only."""
|
| import asyncio
|
| import json
|
| import sys
|
| from pathlib import Path
|
| from typing import Any, Callable, Dict, List, Optional
|
|
|
| REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
|
| repo_root_str = str(REPO_ROOT)
|
| if repo_root_str not in sys.path:
|
| sys.path.insert(0, repo_root_str)
|
|
|
| TOOL_PURPOSE_HINTS = {
|
| "web_search": "Keyword web search with base/medium/pro modes and structured snippets/content output.",
|
| }
|
|
|
| TOOL_EXAMPLE_ARGUMENTS = {
|
| "web_search": {"query": "latest ai model updates", "mode": "base"},
|
| }
|
|
|
| WEB_SEARCH_TOOL = {
|
| "name": "web_search",
|
| "description": (
|
| "Search by keyword and return structured web results. "
|
| "Modes: base(15), medium(35), pro(5 with content_sections)."
|
| ),
|
| "parameters": {
|
| "type": "object",
|
| "properties": {
|
| "query": {
|
| "type": "string",
|
| "description": "Search query keyword or phrase.",
|
| },
|
| "mode": {
|
| "type": "string",
|
| "enum": ["base", "medium", "pro"],
|
| "default": "base",
|
| "description": "Result detail mode. base/medium return snippets, pro includes content sections.",
|
| },
|
| "workers": {
|
| "type": "integer",
|
| "default": 8,
|
| "minimum": 2,
|
| "description": "Parallel workers for the search pipeline.",
|
| },
|
| },
|
| "required": ["query"],
|
| },
|
| }
|
|
|
| WEB_SEARCH_ALIASES = {
|
| "web_search",
|
| "websearch",
|
| "search_web",
|
| "fetch_content",
|
| "featch_content",
|
| }
|
|
|
|
|
| class ToolClient:
|
| """Local tool client — provides only the web_search tool."""
|
|
|
| def __init__(self):
|
| self._tools: List[Dict[str, Any]] = []
|
| self._initialized = False
|
| self._init_error: Optional[str] = None
|
| self._web_search_runner: Optional[Callable[..., Any]] = None
|
|
|
| @staticmethod
|
| def _normalize_tool_name(name: str) -> str:
|
| raw = str(name or "").strip().lower().replace("-", "_").replace(" ", "_")
|
| return "web_search" if raw in WEB_SEARCH_ALIASES else raw
|
|
|
| @staticmethod
|
| def _coerce_positive_int(value: Any, default: int) -> int:
|
| try:
|
| parsed = int(value)
|
| return parsed if parsed > 0 else default
|
| except (TypeError, ValueError):
|
| return default
|
|
|
| def _load_web_search_runner(self) -> Callable[..., Any]:
|
| if self._web_search_runner is not None:
|
| return self._web_search_runner
|
|
|
| try:
|
| from tools.web_search.search import run_search_tool
|
| except Exception as exc:
|
| raise ImportError(f"Unable to load web search runner: {exc}") from exc
|
|
|
| self._web_search_runner = run_search_tool
|
| return run_search_tool
|
|
|
| async def _call_web_search(self, arguments: Dict[str, Any]) -> str:
|
| args = arguments if isinstance(arguments, dict) else {}
|
| query = str(
|
| args.get("query")
|
| or args.get("keyword")
|
| or args.get("keywords")
|
| or ""
|
| ).strip()
|
| if not query:
|
| return json.dumps(
|
| {
|
| "status": "error",
|
| "tool": "web_search",
|
| "error": "Missing required parameter: query",
|
| }
|
| )
|
|
|
| mode = str(args.get("mode") or "base").strip().lower()
|
| if mode not in {"base", "medium", "pro"}:
|
| mode = "base"
|
| workers = self._coerce_positive_int(args.get("workers"), default=8)
|
|
|
| try:
|
| runner = self._load_web_search_runner()
|
| result = await asyncio.to_thread(runner, query=query, mode=mode, workers=workers)
|
| return json.dumps(result, ensure_ascii=False, default=str)
|
| except Exception as exc:
|
| err = str(exc).strip() or repr(exc)
|
| print(f"[TOOL] web_search failed: {err}")
|
| return json.dumps({"status": "error", "tool": "web_search", "error": err})
|
|
|
| @property
|
| def is_available(self) -> bool:
|
| return self._initialized
|
|
|
| @property
|
| def tools(self) -> List[Dict[str, Any]]:
|
| return list(self._tools)
|
|
|
| @property
|
| def init_error(self) -> Optional[str]:
|
| return self._init_error
|
|
|
| async def initialize(self) -> bool:
|
| """Register the web_search tool. Always succeeds."""
|
| self._tools = [dict(WEB_SEARCH_TOOL)]
|
| self._initialized = True
|
| self._init_error = None
|
| print("[TOOLS] web_search tool registered")
|
| return True
|
|
|
| async def call_tool(self, name: str, arguments: Dict[str, Any]) -> str:
|
| """Execute a tool by name. Only web_search is supported."""
|
| normalized_name = self._normalize_tool_name(name)
|
| if normalized_name == "web_search":
|
| return await self._call_web_search(arguments)
|
| return json.dumps({"error": f"Unknown tool: {name}. Only web_search is supported."})
|
|
|
| @staticmethod
|
| def _infer_example_value(parameter_name: str, parameter_schema: Dict[str, Any]) -> Any:
|
| name = (parameter_name or "").lower()
|
| schema_type = str(parameter_schema.get("type") or "").lower()
|
| if name in {"query", "q", "search", "prompt"}:
|
| return "latest official update"
|
| if name in {"limit", "max", "count", "max_results", "top_k"}:
|
| return 5
|
| if schema_type == "boolean":
|
| return False
|
| if schema_type in {"integer", "number"}:
|
| return 1
|
| if schema_type == "array":
|
| return []
|
| if schema_type == "object":
|
| return {}
|
| return "value"
|
|
|
| @classmethod
|
| def _build_tool_example_arguments(cls, tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
|
| explicit = TOOL_EXAMPLE_ARGUMENTS.get(tool_name)
|
| if explicit:
|
| return dict(explicit)
|
|
|
| props = parameters.get("properties", {}) if isinstance(parameters, dict) else {}
|
| required = parameters.get("required", []) if isinstance(parameters, dict) else []
|
|
|
| example: Dict[str, Any] = {}
|
| if isinstance(required, list):
|
| for param_name in required:
|
| if isinstance(param_name, str):
|
| param_schema = props.get(param_name, {}) if isinstance(props, dict) else {}
|
| example[param_name] = cls._infer_example_value(param_name, param_schema)
|
|
|
| if not example and isinstance(props, dict):
|
| for param_name, param_schema in props.items():
|
| example[param_name] = cls._infer_example_value(param_name, param_schema or {})
|
| if len(example) >= 2:
|
| break
|
|
|
| return example
|
|
|
| @classmethod
|
| def _build_tool_example_call(cls, tool_name: str, parameters: Dict[str, Any]) -> str:
|
| payload = {
|
| "tool": tool_name,
|
| "arguments": cls._build_tool_example_arguments(tool_name, parameters),
|
| }
|
| return json.dumps(payload, ensure_ascii=False)
|
|
|
| def get_tools_description(self) -> str:
|
| """Format tool descriptions for the system prompt."""
|
| if not self._tools:
|
| return "No tools currently available."
|
| parts: List[str] = []
|
| for t in self._tools:
|
| name = t["name"]
|
| description = t["description"]
|
| purpose = TOOL_PURPOSE_HINTS.get(name, "General external data retrieval tool.")
|
| parts.append(f"- **{name}**")
|
| parts.append(f" - Description: {description}")
|
| parts.append(f" - Purpose: {purpose}")
|
| params = t.get("parameters", {})
|
| props = params.get("properties", {})
|
| required = set(params.get("required", []))
|
| if not props:
|
| parts.append(" - Parameters: none")
|
| else:
|
| parts.append(" - Parameters:")
|
| for pname, pinfo in props.items():
|
| req_label = " (required)" if pname in required else " (optional)"
|
| ptype = pinfo.get("type", "any")
|
| pdesc = pinfo.get("description", "")
|
| parts.append(f" - {pname} ({ptype}{req_label}): {pdesc}")
|
| parts.append(" - Example Tool Call (JSON):")
|
| parts.append(f" {self._build_tool_example_call(name, params)}")
|
| parts.append(" - Expected Result Style: Return concise factual extraction with source-aware context.")
|
| return "\n".join(parts)
|
|
|
| def get_tool_call_format(self) -> str:
|
| """Return an example JSON tool-call format for the system prompt."""
|
| if self._tools:
|
| first = self._tools[0]
|
| single_obj = json.loads(self._build_tool_example_call(first["name"], first.get("parameters", {})))
|
| return json.dumps(single_obj, ensure_ascii=False)
|
| return '{"tool": "tool_name", "arguments": {...}}'
|
|
|
| def get_tool_names(self) -> List[str]:
|
| return [t["name"] for t in self._tools]
|
|
|
| async def shutdown(self):
|
| """Clean up tool registry."""
|
| self._initialized = False
|
| self._tools = []
|
| self._web_search_runner = None
|
| print("[TOOLS] Tool registry shut down")
|
|
|
|
|
|
|
| tool_client = ToolClient()
|
|
|
