一、自定义coze插件管理阿里云机器

1.1 准备工作

1、注册阿里云账号并完成实名认证

步骤略

2、创建RAM用户并获取AccessKey

地址： https://ram.console.aliyun.com/users

点击“创建用户”

3、为RAM用户授权

选择刚创建的用户 > 点击”添加权限”

4、购买一台测试的ECS

创建Coze插件

依赖包：

第一个工具： get_instance

代码

from runtime import Args
from typings.get_instance.get_instance import Input, Output
from aliyunsdkcore.client import AcsClient
from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest
from aliyunsdkcore.acs_exception.exceptions import ServerException, ClientException
import json
"""
Each file needs to export a function named `handler`. This function is the entrance to the Tool.
Parameters:
args: parameters of the entry function.
args.input - input parameters, you can get test input value by args.input.xxx.
args.logger - logger instance used to print logs, injected by runtime.
Remember to fill in input/output in Metadata, it helps LLM to recognize and use tool.
Return:
The return data of the function, which should match the declared output parameters.
"""
def handler(args: Args[Input]) -> Output:
    # 获取输入参数
    region_id = args.input.region_id
    access_key_id = args.input.access_key_id
    access_key_secret = args.input.access_key_secret
    # 记录日志
    args.logger.info(f"Querying ECS instances in region: {region_id}")
    try:
        # 初始化阿里云客户端
        client = AcsClient(
            access_key_id,
            access_key_secret,
            region_id
        )
        # 创建请求
        request = DescribeInstancesRequest()
        request.set_accept_format('json')
        # 发送请求
        response = client.do_action_with_exception(request)
        response_str = response.decode('utf-8')
        # 解析响应
        response_data = json.loads(response_str)
        instances = response_data.get('Instances', {}).get('Instance', [])
        # 记录成功日志
        args.logger.info(f"Successfully retrieved {len(instances)} instances")
        # 返回结果
        return {
            "instances": instances,
            "success": True,
            "error": None
        }
    except ServerException as e:
        error_msg = f"阿里云服务错误: {e.get_error_msg()} (错误码: {e.get_error_code()})"
        args.logger.error(error_msg)
        return {
            "instances": [],
            "success": False,
            "error": error_msg
        }
    except ClientException as e:
        error_msg = f"请求参数错误: {e.get_error_msg()} (错误码: {e.get_error_code()})"
        args.logger.error(error_msg)
        return {
            "instances": [],
            "success": False,
            "error": error_msg
        }
    except Exception as e:
        error_msg = f"未知错误: {str(e)}"
        args.logger.error(error_msg)
        return {
            "instances": [],
            "success": False,
            "error": error_msg
        }

元数据

输入参数

输出参数

第二个工具：manager_instance

代码：

from runtime import Args
from typings.manager_instance.manager_instance import Input, Output
from aliyunsdkcore.client import AcsClient
from aliyunsdkecs.request.v20140526.StopInstancesRequest import StopInstancesRequest
from aliyunsdkecs.request.v20140526.RebootInstancesRequest import RebootInstancesRequest
from aliyunsdkecs.request.v20140526.StartInstancesRequest import StartInstancesRequest
from aliyunsdkcore.acs_exception.exceptions import ServerException, ClientException
import json
"""
Each file needs to export a function named `handler`. This function is the entrance to the Tool.
Parameters:
args: parameters of the entry function.
args.input - input parameters, you can get test input value by args.input.xxx.
args.logger - logger instance used to print logs, injected by runtime.
Remember to fill in input/output in Metadata, it helps LLM to recognize and use tool.
Return:
The return data of the function, which should match the declared output parameters.
"""
def handler(args: Args[Input]) -> Output:
    # 获取输入参数
    action = args.input.action.lower()  # 转换为小写
    instance_id = args.input.instance_id
    region_id = args.input.region_id
    access_key_id = args.input.access_key_id
    access_key_secret = args.input.access_key_secret
    force_stop = args.input.force_stop if hasattr(args.input, 'force_stop') else False
    force_reboot = args.input.force_reboot if hasattr(args.input, 'force_reboot') else False
    # 记录操作日志
    args.logger.info(f"Performing action '{action}' on instance '{instance_id}' in region '{region_id}'")
    # 验证操作类型
    if action not in ['stop', 'reboot', 'start']:
        error_msg = f"无效的操作类型: {action}. 支持的操作: stop, reboot, start"
        args.logger.error(error_msg)
        return {
            "success": False,
            "error": error_msg,
            "request_id": None
        }
    try:
        # 初始化阿里云客户端
        client = AcsClient(
            access_key_id,
            access_key_secret,
            region_id
        )
        # 根据操作类型创建请求
        if action == 'stop':
            request = StopInstancesRequest()
            request.set_InstanceIds([instance_id])
            request.set_ForceStop(force_stop)
            operation_name = "关机"
        elif action == 'reboot':
            request = RebootInstancesRequest()
            request.set_InstanceIds([instance_id])
            request.set_ForceReboot(force_reboot)
            operation_name = "重启"
        else:  # start
            request = StartInstancesRequest()
            request.set_InstanceIds([instance_id])
            operation_name = "启动"
        # 设置请求格式
        request.set_accept_format('json')
        # 发送请求
        response = client.do_action_with_exception(request)
        response_str = response.decode('utf-8')
        # 解析响应
        response_data = json.loads(response_str)
        request_id = response_data.get('RequestId')
        # 记录成功日志
        args.logger.info(f"Successfully initiated {operation_name} operation for instance '{instance_id}'. Request ID: {request_id}")
        # 返回结果
        return {
            "success": True,
            "error": None,
            "request_id": request_id,
            "message": f"已成功发起{operation_name}操作，请求ID: {request_id}"
        }
    except ServerException as e:
        error_msg = f"阿里云服务错误: {e.get_error_msg()} (错误码: {e.get_error_code()})"
        args.logger.error(error_msg)
        return {
            "success": False,
            "error": error_msg,
            "request_id": e.get_request_id()
        }
    except ClientException as e:
        error_msg = f"请求参数错误: {e.get_error_msg()} (错误码: {e.get_error_code()})"
        args.logger.error(error_msg)
        return {
            "success": False,
            "error": error_msg,
            "request_id": None
        }
    except Exception as e:
        error_msg = f"未知错误: {str(e)}"
        args.logger.error(error_msg)
        return {
            "success": False,
            "error": error_msg,
            "request_id": None
        }

元数据

1.2 设计Coze工作流

用户提示词

提取{{input}}中关键信息，获取到实例id，

例如：提取“将实例i-<your-admin-password>5关机”中的实例id应该为i-<your-admin-password>5，

将输出内容为该实例id，注意，输出内容并没有多余信息，只有实例id

用户提示词

提取{{input}}中关键信息，
如果包含：“开启”、“启动”、“开机”、“起来”则输出'start'；
如果包含：“关闭”、“关机”、“关掉”则输出'stop'；
如果包含：“重启”、“重新启动”则输出'reboot'；
否则什么都不输出。
注意，以上情况仅仅输出'start'、'stop'或者'reboot'，不需要任何其他多余信息

1.3 设计aiops智能体

个人空间 –> 项目开发 –> 右上角“+项目”

开场白

我是阿里云智能助手，我可以帮你启动、关闭、重启实例，请把你的需求告诉我吧。

预置问题

帮我重启云主机i-<your-admin-password>56

把云主机i-aaaaaa关机

重启一下阿里云实例 i-bbbbbbbb