Alibaba Cloud Operations MCP Server

Instructions

Args:
    region: 区域ID，如cn-beijing
    instance_ids: ECS实例ID列表
    command: 要执行的命令

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)}"

• alibaba_cloud_ops_mcp_server/tools/oos_tools.py:49-54 (schema)

  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')
  ):

• alibaba_cloud_ops_mcp_server/tools/oos_tools.py:48-72 (handler)

  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)

• alibaba_cloud_ops_mcp_server/tools/oos_tools.py:25-48 (helper)

  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