run_ecs_command
Execute commands on Alibaba Cloud ECS instances to manage, configure, or troubleshoot cloud servers remotely.
Instructions
在ECS实例上运行命令
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
command: 要执行的命令
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| region | No | cn-beijing | |
| instance_ids | No | ||
| command | No | echo 'Hello World' |
Implementation Reference
- complete_fastmcp_server.py:304-327 (handler)The primary handler function for the MCP tool 'run_ecs_command'. Registered via @app.tool(). It wraps and calls the underlying OOS_RunCommand by dynamic lookup.@app.tool() def run_ecs_command(region: str = "cn-beijing", instance_ids: List[str] = None, command: str = "echo 'Hello World'") -> str: """在ECS实例上运行命令 Args: region: 区域ID，如cn-beijing instance_ids: ECS实例ID列表 command: 要执行的命令 """ try: sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server')) from tools import oos_tools if not instance_ids: return "请提供ECS实例ID列表" for tool_func in oos_tools.tools: if hasattr(tool_func, '__name__') and 'run' in tool_func.__name__.lower() and 'command' in tool_func.__name__.lower(): result = tool_func(RegionId=region, InstanceIds=instance_ids, CommandContent=command) return str(result) return f"ECS命令执行功能可用，region: {region}, 实例: {instance_ids}, 命令: {command}" except Exception as e: return f"ECS命令执行失败: {str(e)}"
- Pydantic-based input schema definitions for the OOS_RunCommand function, which defines parameters like Command, InstanceIds, RegionId, and CommandType.def OOS_RunCommand( Command: str = Field(description='Content of the command executed on the ECS instance'), InstanceIds: List[str] = Field(description='AlibabaCloud ECS instance ID List'), RegionId: str = Field(description='AlibabaCloud region ID', default='cn-hangzhou'), CommandType: str = Field(description='The type of command executed on the ECS instance, optional value：RunShellScript，RunPythonScript，RunPerlScript，RunBatScript，RunPowerShellScript', default='RunShellScript') ):
- Core handler implementation OOS_RunCommand that invokes Alibaba Cloud OOS to run commands on multiple ECS instances using the 'ACS-ECS-BulkyRunCommand' template.@tools.append def OOS_RunCommand( Command: str = Field(description='Content of the command executed on the ECS instance'), InstanceIds: List[str] = Field(description='AlibabaCloud ECS instance ID List'), RegionId: str = Field(description='AlibabaCloud region ID', default='cn-hangzhou'), CommandType: str = Field(description='The type of command executed on the ECS instance, optional value：RunShellScript，RunPythonScript，RunPerlScript，RunBatScript，RunPowerShellScript', default='RunShellScript') ): """批量在多台ECS实例上运行云助手命令，适用于需要同时管理多台ECS实例的场景，如应用程序管理和资源标记操作等。""" parameters = { 'regionId': RegionId, 'resourceType': 'ALIYUN::ECS::Instance', 'targets': { 'ResourceIds': InstanceIds, 'RegionId': RegionId, 'Type': 'ResourceIds', 'Parameters': { 'RegionId': RegionId, 'Status': 'Running' } }, "commandType": CommandType, "commandContent": Command } return _start_execution_sync(region_id=RegionId, template_name='ACS-ECS-BulkyRunCommand', parameters=parameters)
- Helper function that starts an OOS execution and polls until completion, handling success, failure, or cancellation.def _start_execution_sync(region_id: str, template_name: str, parameters: dict): client = create_client(region_id=region_id) start_execution_request = oos_20190601_models.StartExecutionRequest( region_id=region_id, template_name=template_name, parameters=json.dumps(parameters) ) start_execution_resp = client.start_execution(start_execution_request) execution_id = start_execution_resp.body.execution.execution_id while True: list_executions_request = oos_20190601_models.ListExecutionsRequest( region_id=region_id, execution_id=execution_id ) list_executions_resp = client.list_executions(list_executions_request) status = list_executions_resp.body.executions[0].status if status == FAILED: status_message = list_executions_resp.body.executions[0].status_message raise exception.OOSExecutionFailed(reason=status_message) elif status in END_STATUSES: return list_executions_resp.body time.sleep(1) @tools.append