Execute a tool call and return the result as a string. Args: session_id: The ChatSession ID, used to isolate AgentBay instances per conversation. Passed through to agentbay_* tools.
(
tool_name: str,
arguments: dict,
agent_id: uuid.UUID,
user_id: uuid.UUID,
session_id: str = "",
on_output=None,
)
| 2845 | |
| 2846 | |
| 2847 | async def execute_tool( |
| 2848 | tool_name: str, |
| 2849 | arguments: dict, |
| 2850 | agent_id: uuid.UUID, |
| 2851 | user_id: uuid.UUID, |
| 2852 | session_id: str = "", |
| 2853 | on_output=None, |
| 2854 | ) -> str: |
| 2855 | """Execute a tool call and return the result as a string. |
| 2856 | |
| 2857 | Args: |
| 2858 | session_id: The ChatSession ID, used to isolate AgentBay instances |
| 2859 | per conversation. Passed through to agentbay_* tools. |
| 2860 | """ |
| 2861 | if not isinstance(tool_name, str): |
| 2862 | tool_name = str(tool_name or "") |
| 2863 | tool_name = ( |
| 2864 | tool_name |
| 2865 | .replace("`", "") |
| 2866 | .replace("\u200b", "") |
| 2867 | .replace("\u200c", "") |
| 2868 | .replace("\u200d", "") |
| 2869 | .replace("\ufeff", "") |
| 2870 | .strip() |
| 2871 | ) |
| 2872 | if tool_name == FINISH_TOOL_NAME: |
| 2873 | content = arguments.get("content", "") |
| 2874 | return content if isinstance(content, str) else str(content) |
| 2875 | |
| 2876 | _agent_tenant_id = await _get_agent_tenant_id(agent_id) |
| 2877 | |
| 2878 | ws = _agent_workspace_root(agent_id) |
| 2879 | |
| 2880 | # ── Autonomy boundary check ── |
| 2881 | action_type = _TOOL_AUTONOMY_MAP.get(tool_name) |
| 2882 | if action_type: |
| 2883 | try: |
| 2884 | from app.services.autonomy_service import autonomy_service |
| 2885 | from app.models.agent import Agent as AgentModel |
| 2886 | async with async_session() as _adb: |
| 2887 | _ar = await _adb.execute(select(AgentModel).where(AgentModel.id == agent_id)) |
| 2888 | _agent = _ar.scalar_one_or_none() |
| 2889 | if _agent: |
| 2890 | result_check = await autonomy_service.check_and_enforce( |
| 2891 | _adb, _agent, action_type, {"tool": tool_name, "args": str(arguments)[:200], "requested_by": str(user_id)} |
| 2892 | ) |
| 2893 | await _adb.commit() |
| 2894 | if not result_check.get("allowed"): |
| 2895 | level = result_check.get("level", "L3") |
| 2896 | logger.info(f"[Autonomy] Tool {tool_name} denied, level: {level}") |
| 2897 | if level == "L3": |
| 2898 | return f"⏳ This action requires approval. An approval request has been sent. Please wait for approval before retrying. (Approval ID: {result_check.get('approval_id', 'N/A')})" |
| 2899 | return f"❌ Action denied: {result_check.get('message', 'unknown reason')}" |
| 2900 | except Exception as e: |
| 2901 | logger.exception(f"[Autonomy] Check failed: {e}") |
| 2902 | return f"⚠️ Autonomy check failed ({e}). Operation blocked for safety. Please retry or contact admin." |
| 2903 | |
| 2904 | # Pre-inject session_id into arguments for AgentBay tools so each |