Spaces:
Runtime error
Runtime error
| """ | |
| Execution dispatching for ToolStore tools. | |
| Routes execute calls to the appropriate backend: | |
| - MCP → FullMCPClient (connection pool) | |
| - Skill → SkillManager (local SKILL.md execution) | |
| - Toolset → local import + call, or remote Docker container | |
| """ | |
| from __future__ import annotations | |
| import json as _json | |
| from typing import Any, Dict | |
| from toolstore.schema_converter import flatten_mcp_content | |
| from toolstore.skill_manager import get_skill_manager | |
| def execute_tool(tool: Dict[str, Any], args: Dict[str, Any], | |
| config_manager, index_manager) -> str: | |
| """Dispatch a single tool execution to the correct backend. | |
| Returns the result as a JSON or plain-text string. | |
| """ | |
| tool_name = tool["name"] | |
| tool_type = tool.get("type", "unknown") | |
| if tool_type == "mcp": | |
| return _execute_mcp(tool, args, config_manager) | |
| elif tool_type == "skill": | |
| return _execute_skill(tool, args, config_manager) | |
| elif tool_type == "toolset": | |
| return _execute_toolset(tool, args) | |
| else: | |
| return f"Error: Unknown tool type '{tool_type}'" | |
| # --------------------------------------------------------------------------- | |
| # MCP execution (via FullMCPClient + connection pool) | |
| # --------------------------------------------------------------------------- | |
| def _execute_mcp(tool: Dict[str, Any], args: Dict[str, Any], | |
| config_manager) -> str: | |
| from toolstore.mcp_client import get_client | |
| server_name = tool.get("mcp_server") | |
| if not server_name: | |
| return "Error: Tool definition missing 'mcp_server'" | |
| servers = config_manager.get_mcp_servers() | |
| config = servers.get(server_name) | |
| if not config: | |
| return f"Error: MCP server '{server_name}' not found in config." | |
| try: | |
| client = get_client(server_name, config) | |
| result = client.call_tool(tool["name"], args) | |
| content = result.get("content", []) | |
| if result.get("isError"): | |
| return "[TOOL ERROR] " + flatten_mcp_content(content) | |
| return flatten_mcp_content(content) | |
| except Exception as exc: | |
| return f"Error executing MCP tool: {str(exc)}" | |
| # --------------------------------------------------------------------------- | |
| # Skill execution | |
| # --------------------------------------------------------------------------- | |
| def _execute_skill(tool: Dict[str, Any], args: Dict[str, Any], | |
| config_manager) -> str: | |
| # Strip "skill:" prefix if present (index uses prefixed names) | |
| skill_name = tool["name"] | |
| if skill_name.startswith("skill:"): | |
| skill_name = skill_name[len("skill:"):] | |
| skill_action = args.get("action", "load") | |
| sm = get_skill_manager(config_manager.get_skill_dirs()) | |
| # Lazily scan if not already loaded | |
| if not sm.get_skill(skill_name): | |
| sm.scan() | |
| if skill_action == "load": | |
| body = sm.get_skill_body(skill_name) | |
| if body is None: | |
| return f"Error: Skill '{skill_name}' not loaded." | |
| return body | |
| elif skill_action == "files": | |
| sd = sm.get_skill(skill_name) | |
| if not sd: | |
| return f"Error: Skill '{skill_name}' not found." | |
| flist = [str(f) for f in sd.list_files()] | |
| return "\n".join(flist) if flist else "(no additional files bundled)" | |
| elif skill_action == "file": | |
| file_path = args.get("file_path", "") | |
| if not file_path: | |
| return "Error: 'file_path' is required for action='file'." | |
| content = sm.get_skill_file(skill_name, file_path) | |
| if content is None: | |
| return f"Error: File '{file_path}' not found in skill '{skill_name}'." | |
| return content | |
| elif skill_action == "run": | |
| script = args.get("script", "") | |
| if not script: | |
| return "Error: 'script' argument is required for action='run'." | |
| return sm.run_skill_script(skill_name, script) | |
| else: | |
| return f"Error: Unknown skill action '{skill_action}'. Use 'load', 'files', or 'file'." | |
| # --------------------------------------------------------------------------- | |
| # Toolset execution | |
| # --------------------------------------------------------------------------- | |
| def _execute_toolset(tool: Dict[str, Any], args: Dict[str, Any]) -> str: | |
| """Execute a toolset. | |
| Two modes: | |
| - **Local** (has ``toolset_dir``): import + call directly in-process. | |
| No Docker, no sandbox — the toolset is installed on the host. | |
| - **Remote** (has ``code``, no ``toolset_dir``): run in a dedicated | |
| ephemeral Docker container with the toolset's pre-configured | |
| environment. Zero approval required. | |
| The agent passes ``{"function": "...", ...}`` in arguments. | |
| """ | |
| # Take a copy so we don't mutate the caller's dict. | |
| args = dict(args) | |
| # 1. Argument validation — which function? | |
| function_name = args.pop("function", None) | |
| if not function_name: | |
| bindings = tool.get("bindings", {}) | |
| if len(bindings) == 1: | |
| function_name = next(iter(bindings)) | |
| else: | |
| names = list(bindings.keys()) if bindings else [] | |
| return ( | |
| f"Error: 'function' argument required. " | |
| f"Available functions: {', '.join(names) or '(none)'}" | |
| ) | |
| # 2. Validate the binding exists | |
| bindings = tool.get("bindings", {}) | |
| if function_name not in bindings: | |
| names = list(bindings.keys()) | |
| return ( | |
| f"Error: Unknown function '{function_name}'. " | |
| f"Available: {', '.join(names)}" | |
| ) | |
| # 3. Dispatch: local (in-process) vs remote (dedicated container) | |
| toolset_dir = tool.get("toolset_dir") | |
| if toolset_dir: | |
| return _execute_toolset_local(toolset_dir, function_name, args) | |
| code = tool.get("code") or tool.get("code_base64") | |
| if code: | |
| return _execute_toolset_remote(tool, function_name, args) | |
| return "Error: toolset has neither 'toolset_dir' nor 'code' — cannot execute" | |
| def _execute_toolset_local(toolset_dir: str, function_name: str, | |
| args: Dict[str, Any]) -> str: | |
| """Run a local toolset directly in-process — just import and call.""" | |
| import importlib.util | |
| from pathlib import Path | |
| from toolstore.toolset import clear_registry, get_tool | |
| code_path = Path(toolset_dir) / "toolset.py" | |
| if not code_path.exists(): | |
| return f"Error: toolset.py not found at {code_path}" | |
| try: | |
| clear_registry() | |
| # Dynamically load the toolset module | |
| spec = importlib.util.spec_from_file_location( | |
| "toolset_local", str(code_path) | |
| ) | |
| if spec is None or spec.loader is None: | |
| return "Error: failed to create module spec for toolset.py" | |
| mod = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(mod) | |
| fn = get_tool(function_name) | |
| if fn is None: | |
| from toolstore.toolset import get_tool_names | |
| names = get_tool_names() | |
| return ( | |
| f"Error: Function '{function_name}' not found in toolset. " | |
| f"Available: {', '.join(names) or '(none)'}" | |
| ) | |
| result = fn(**args) | |
| clear_registry() | |
| return _json.dumps(result, default=str, indent=2) | |
| except Exception as exc: | |
| clear_registry() | |
| return f"Error executing local toolset '{function_name}': {exc}" | |
| def _execute_toolset_remote(tool: Dict[str, Any], function_name: str, | |
| args: Dict[str, Any]) -> str: | |
| """Run a registry toolset in-process — no Docker needed. | |
| Writes code to a temp directory, pip‑installs deps, then imports | |
| and calls the function just like _execute_toolset_local. | |
| """ | |
| import base64 | |
| import subprocess | |
| import sys | |
| import tempfile | |
| from pathlib import Path | |
| code = tool.get("code", "") | |
| code_b64 = tool.get("code_base64", "") | |
| if code_b64 and not code: | |
| code = base64.b64decode(code_b64).decode("utf-8") | |
| if not code: | |
| return "Error: toolset has no code to execute" | |
| # Temp directory — lives for the duration of the function call | |
| with tempfile.TemporaryDirectory(prefix="toolset_") as tmp_dir: | |
| tmp = Path(tmp_dir) | |
| (tmp / "toolset.py").write_text(code, encoding="utf-8") | |
| # Install requirements if present | |
| requirements = tool.get("requirements", []) | |
| if isinstance(requirements, str): | |
| requirements = [r.strip() for r in requirements.split("\n") if r.strip()] | |
| if requirements: | |
| return ( | |
| f"Error: This toolset requires packages that aren't installed: " | |
| f"{', '.join(requirements)}.\n" | |
| f"Install them first: pip install {' '.join(requirements)}" | |
| ) | |
| # Now delegate to the local runner | |
| return _execute_toolset_local(str(tmp), function_name, args) | |