Merge pull request #27 from aliyun/fix/parameters_list_json_str

Optimize the list parameter processing of ECS services

Author: ceshihao

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "alibaba-cloud-ops-mcp-server"
-version = "0.8.7"
+version = "0.8.8"
 description = "A MCP server for Alibaba Cloud"
 readme = "README.md"
 authors = [

src/alibaba_cloud_ops_mcp_server/tools/api_tools.py

@@ -2,6 +2,7 @@
 from mcp.server.fastmcp import FastMCP, Context
 from pydantic import Field
 import logging
+import json
 
 import inspect
 import types
@@ -32,15 +33,32 @@ def create_client(service: str, region_id: str) -> OpenApiClient:
     return OpenApiClient(config)
 
 
+# 类型为String的JSON数组参数
+ECS_LIST_PARAMETERS = {
+    'HpcClusterIds', 'DedicatedHostClusterIds', 'DedicatedHostIds', 
+    'InstanceIds', 'DeploymentSetIds', 'KeyPairNames', 'SecurityGroupIds', 
+    'diskIds', 'repeatWeekdays', 'timePoints', 'DiskIds', 'SnapshotLinkIds', 
+    'EipAddresses', 'PublicIpAddresses', 'PrivateIpAddresses'
+}
+
+
 def _tools_api_call(service: str, api: str, parameters: dict, ctx: Context):
     service = service.lower()
     api_meta, _ = ApiMetaClient.get_api_meta(service, api)
     version = ApiMetaClient.get_service_version(service)
     method = 'POST' if api_meta.get('methods', [])[0] == 'post' else 'GET'
     path = api_meta.get('path', '/')
     style = ApiMetaClient.get_service_style(service)
+    
+    # 处理特殊参数格式
+    processed_parameters = parameters.copy()
+    if service == 'ecs':
+        for param_name, param_value in parameters.items():
+            if param_name in ECS_LIST_PARAMETERS and isinstance(param_value, list):
+                processed_parameters[param_name] = json.dumps(param_value)
+    
     req = open_api_models.OpenApiRequest(
-        query=OpenApiUtilClient.query(parameters)
+        query=OpenApiUtilClient.query(processed_parameters)
     )
     params = open_api_models.Params(
         action=api,
@@ -53,7 +71,7 @@ def _tools_api_call(service: str, api: str, parameters: dict, ctx: Context):
         req_body_type='formData',
         body_type='json'
     )
-    client = create_client(service, parameters.get('RegionId', 'cn-hangzhou'))
+    client = create_client(service, processed_parameters.get('RegionId', 'cn-hangzhou'))
     runtime = util_models.RuntimeOptions()
     return client.call_api(params, req, runtime)
 
@@ -75,9 +93,15 @@ def _create_function_schemas(service, api, api_meta):
         description = schema.get('description', '')
         example = schema.get('example', '')
         type_ = schema.get('type', '')
-        description = f'{description} 请注意，提供参数要严格按照参数的类型和参数示例的提示，如果提到参数为String，且为一个 JSON 数组字符串，应在数组内使用单引号包裹对应的参数以避免转义问题，并在最外侧用双引号包裹以确保其是字符串，否则可能会导致参数解析错误。参数类型: {type_},参数示例：{example}'
+        description = f'{description} 参数类型: {type_},参数示例：{example}'
         required = schema.get('required', False)
-        python_type = type_map.get(type_, str)
+        
+        # 只有在service为ecs时，才对特定参数进行特殊处理
+        if service.lower() == 'ecs' and name in ECS_LIST_PARAMETERS:
+            python_type = list
+        else:
+            python_type = type_map.get(type_, str)
+            
         field_info = (
             python_type,
             field(

tests/tools/test_api_tools.py

@@ -1,6 +1,7 @@
 import pytest
 from unittest.mock import patch, MagicMock
 from alibaba_cloud_ops_mcp_server.tools import api_tools
+import json
 
 def fake_api_meta(post=False, no_summary=False):
     meta = {
@@ -127,3 +128,104 @@ def test_create_and_decorate_tool_api_meta_exception():
         with pytest.raises(Exception) as e:
             api_tools._create_and_decorate_tool(mcp, 'ecs', 'DescribeInstances')
         assert 'meta-fail' in str(e.value)
+
+def test_create_function_schemas_ecs_list_parameters():
+    # 测试ECS服务的特殊参数处理
+    api_meta = {
+        'parameters': [
+            {'name': 'InstanceIds', 'schema': {'type': 'string', 'description': '实例ID列表', 'example': '["i-123", "i-456"]', 'required': True}},
+            {'name': 'SecurityGroupIds', 'schema': {'type': 'string', 'description': '安全组ID列表', 'example': '["sg-123", "sg-456"]', 'required': False}},
+            {'name': 'NormalParam', 'schema': {'type': 'string', 'description': '普通参数', 'example': 'test', 'required': False}},
+        ],
+        'methods': ['get'],
+        'path': '/test',
+        'summary': '测试API'
+    }
+    
+    # 测试ECS服务
+    schemas = api_tools._create_function_schemas('ecs', 'DescribeInstances', api_meta)
+    assert schemas['DescribeInstances']['InstanceIds'][0] == list
+    assert schemas['DescribeInstances']['SecurityGroupIds'][0] == list
+    assert schemas['DescribeInstances']['NormalParam'][0] == str
+    
+    # 测试非ECS服务
+    schemas = api_tools._create_function_schemas('rds', 'DescribeInstances', api_meta)
+    assert schemas['DescribeInstances']['InstanceIds'][0] == str
+    assert schemas['DescribeInstances']['SecurityGroupIds'][0] == str
+    assert schemas['DescribeInstances']['NormalParam'][0] == str
+
+def test_tools_api_call_ecs_list_parameters():
+    with patch('alibaba_cloud_ops_mcp_server.tools.api_tools.ApiMetaClient') as mock_ApiMetaClient, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.create_client') as mock_create_client, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.open_api_models') as mock_open_api_models, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.OpenApiUtilClient') as mock_OpenApiUtilClient, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.util_models') as mock_util_models:
+        
+        mock_ApiMetaClient.get_api_meta.return_value = fake_api_meta()
+        mock_ApiMetaClient.get_service_version.return_value = '2023-01-01'
+        mock_ApiMetaClient.get_service_style.return_value = 'RPC'
+        mock_open_api_models.OpenApiRequest.return_value = MagicMock()
+        mock_open_api_models.Params.return_value = MagicMock()
+        mock_create_client.return_value.call_api.return_value = {'result': 'ok'}
+        mock_OpenApiUtilClient.query.return_value = {}
+        mock_util_models.RuntimeOptions.return_value = MagicMock()
+
+        # 测试ECS服务的列表参数转换
+        params = {
+            'InstanceIds': ['i-123', 'i-456'],
+            'SecurityGroupIds': ['sg-123', 'sg-456'],
+            'NormalParam': 'test',
+            'RegionId': 'cn-hangzhou'
+        }
+        
+        # 测试ECS服务
+        result = api_tools._tools_api_call('ecs', 'DescribeInstances', params, None)
+        # 验证传入query方法的参数
+        query_args = mock_OpenApiUtilClient.query.call_args[0][0]
+        assert isinstance(query_args['InstanceIds'], str)
+        assert isinstance(query_args['SecurityGroupIds'], str)
+        assert query_args['NormalParam'] == 'test'
+        assert json.loads(query_args['InstanceIds']) == ['i-123', 'i-456']
+        assert json.loads(query_args['SecurityGroupIds']) == ['sg-123', 'sg-456']
+        
+        # 重置mock
+        mock_OpenApiUtilClient.query.reset_mock()
+        
+        # 测试非ECS服务
+        result = api_tools._tools_api_call('rds', 'DescribeInstances', params, None)
+        # 验证传入query方法的参数
+        query_args = mock_OpenApiUtilClient.query.call_args[0][0]
+        assert isinstance(query_args['InstanceIds'], list)
+        assert isinstance(query_args['SecurityGroupIds'], list)
+        assert query_args['InstanceIds'] == ['i-123', 'i-456']
+        assert query_args['SecurityGroupIds'] == ['sg-123', 'sg-456']
+
+def test_tools_api_call_ecs_list_parameters_non_list():
+    with patch('alibaba_cloud_ops_mcp_server.tools.api_tools.ApiMetaClient') as mock_ApiMetaClient, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.create_client') as mock_create_client, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.open_api_models') as mock_open_api_models, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.OpenApiUtilClient') as mock_OpenApiUtilClient, \
+         patch('alibaba_cloud_ops_mcp_server.tools.api_tools.util_models') as mock_util_models:
+        
+        mock_ApiMetaClient.get_api_meta.return_value = fake_api_meta()
+        mock_ApiMetaClient.get_service_version.return_value = '2023-01-01'
+        mock_ApiMetaClient.get_service_style.return_value = 'RPC'
+        mock_open_api_models.OpenApiRequest.return_value = MagicMock()
+        mock_open_api_models.Params.return_value = MagicMock()
+        mock_create_client.return_value.call_api.return_value = {'result': 'ok'}
+        mock_OpenApiUtilClient.query.return_value = {}
+        mock_util_models.RuntimeOptions.return_value = MagicMock()
+
+        # 测试非列表类型的特殊参数
+        params = {
+            'InstanceIds': 'i-123',  # 字符串而不是列表
+            'SecurityGroupIds': None,  # None值
+            'RegionId': 'cn-hangzhou'
+        }
+        
+        # 测试ECS服务
+        result = api_tools._tools_api_call('ecs', 'DescribeInstances', params, None)
+        # 验证传入query方法的参数
+        query_args = mock_OpenApiUtilClient.query.call_args[0][0]
+        assert query_args['InstanceIds'] == 'i-123'
+        assert query_args['SecurityGroupIds'] is None