Function execute_tool_call

api_utils/tools_registry.py:81–120

Execute registered tools and return stringified result. Unknown tools return descriptive errors. Fully asynchronous: built-in functions execute directly; MCP path uses async httpx client.

(name: str, arguments_json: str)

async def execute_tool_call(name: str, arguments_json: str) -> str:
    """Execute registered tools and return stringified result. Unknown tools return descriptive errors.
    Fully asynchronous: built-in functions execute directly; MCP path uses async httpx client.
    """
    try:
        params = json.loads(arguments_json or "{}")
    except Exception:
        params = {}

    func = FUNCTION_REGISTRY.get(name)
    if not func:
        # If tool is not built-in but declared, try MCP adapter if configured (env or per-request)
        if name in _ALLOWED_RUNTIME_TOOLS:
            try:
                from api_utils.mcp_adapter import (
                    execute_mcp_tool,
                    execute_mcp_tool_with_endpoint,
                )

                if _runtime_mcp_endpoint:
                    return await execute_mcp_tool_with_endpoint(
                        _runtime_mcp_endpoint, name, params
                    )
                if os.environ.get("MCP_HTTP_ENDPOINT"):
                    return await execute_mcp_tool(name, params)
            except asyncio.CancelledError:
                raise
            except Exception as e:
                return json.dumps(
                    {"error": f"MCP execution failed: {e}"}, ensure_ascii=False
                )
        return json.dumps(
            {"error": f"Unknown tool: {name}", "arguments": params}, ensure_ascii=False
        )

    try:
        result = func(params)
        return json.dumps(result, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": f"Execution failed: {e}"}, ensure_ascii=False)