listInstances

Instructions

Implementation Reference

• src/alibabacloud_dms_mcp_server/server.py:202-243 (handler)

  The 'list_instance' async function is the actual handler for the 'listInstances' tool. It accepts optional search_key, db_type, and env_type parameters, calls the Alibaba Cloud DMS API client.list_instances(), processes the response into InstanceDetail objects, and returns them.

  async def list_instance(
          search_key: Optional[str] = Field(default=None,
                                            description="Optional search key (e.g., instance host, instance alias, etc.)"),
          db_type: Optional[str] = Field(default=None,
                                         description="Optional instanceType, or called dbType (e.g., mysql, polardb, oracle, "
                                                     "postgresql, sqlserver, polardb-pg, etc.)"),
          env_type: Optional[str] = Field(default=None,
                                          description="Optional instance environment type (e.g., product, dev, test, etc. )")
  ) -> List[InstanceDetail]:
      client = create_client()
      req = dms_enterprise_20181101_models.ListInstancesRequest()
      if search_key:
          req.search_key = search_key
      if db_type:
          req.db_type = db_type
      if env_type:
          req.env_type = env_type
      if mcp.state.real_login_uid:
          req.real_login_user_uid = mcp.state.real_login_uid
      try:
          resp = client.list_instances(req)
  
          instance_data = resp.body.to_map()
          if "InstanceList" not in instance_data:
              return []
          instance_list = instance_data["InstanceList"]
          # 检查是否有 Instance 键
          if "Instance" not in instance_list:
              return []
          instances = instance_list["Instance"]
          # 检查是否为空
          if not isinstance(instances, list) or not instances:
              return []
  
          processed_instances = [
              {**item, 'InstanceResourceId': item.pop('EcsInstanceId', None)}
              for item in instances
          ]
          return [InstanceDetail(**item) for item in processed_instances]
      except Exception as e:
          logger.error(f"Error in list_instance: {e}")
          raise

• src/alibabacloud_dms_mcp_server/server.py:35-48 (schema)

  The 'InstanceDetail' Pydantic model defines the schema for the output of listInstances. It includes fields such as InstanceId, State, InstanceType, InstanceAlias, EnvType, Host, Port, InstanceSource, and InstanceResourceId.

  class InstanceDetail(MyBaseModel):
      InstanceId: Any = Field(description="Unique instance identifier in DMS", default=None)
      State: Any = Field(description="Current operational status", default=None)
      InstanceType: Any = Field(description="Database Engine type", default=None)
      InstanceAlias: Any = Field(description="Instance alias in DMS", default=None)
      EnvType: Any = Field(description="The environment type of the instance (e.g., production, development, etc.)",
                           default=None)
      Host: Any = Field(description="The hostname of the database instance", default=None)
      Port: Any = Field(description="The connection port number", default=None)
      InstanceSource: Any = Field(description="The instance source (e.g., RDS, VPC_IDC, ECS_OWN, PUBLIC_OWN etc.)",
                                  default=None)
      InstanceResourceId: Any = Field(
          description="Resource ID of the instance from RDS",
          default=None)

• src/alibabacloud_dms_mcp_server/server.py:731-732 (registration)

  The 'listInstances' tool is registered in '_register_full_toolset()' via self.mcp.tool(name='listInstances', ...) which binds the 'list_instance' handler function as the tool's implementation.

  self.mcp.tool(name="listInstances", description="Search for instances from DMS.",
                annotations={"title": "搜索DMS实例列表", "readOnlyHint": True})(list_instance)