Alibaba Cloud Operations MCP Server

Instructions

Args:
    region: 区域ID

Implementation Reference

• complete_fastmcp_server.py:87-118 (handler)

  FastMCP @app.tool() handler for list_oss_buckets that invokes oss_tools and formats output.

  def list_oss_buckets(region: str = "cn-beijing") -> str:
      """列出OSS存储桶
      
      Args:
          region: 区域ID
      """
      try:
          sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
          from tools import oss_tools
          
          # 使用修复后的OSS工具
          list_buckets_func = oss_tools.tools[0]  # OSS_ListBuckets
          result = list_buckets_func(RegionId=region)
          
          if isinstance(result, list) and result:
              output = f"阿里云 {region} region的OSS存储桶列表:\\n"
              output += "=" * 50 + "\\n"
              
              for i, bucket in enumerate(result, 1):
                  if isinstance(bucket, dict):
                      output += f"{i}. 存储桶名称: {bucket.get('name', '未知')}\\n"
                      output += f"   创建时间: {bucket.get('creation_date', '未知')}\\n"
                      output += f"   位置: {bucket.get('location', '未知')}\\n"
                      output += f"   存储类型: {bucket.get('storage_class', '未知')}\\n"
                      output += f"   外网端点: {bucket.get('extranet_endpoint', '未知')}\\n"
                      output += "-" * 30 + "\\n"
              return output
          else:
              return f"在 {region} region没有找到OSS存储桶"
                      
      except Exception as e:
          return f"OSS查询失败: {str(e)}"

• qcli_compatible_server_fixed.py:114-132 (handler)

  Fallback async handler for the list_oss_buckets tool in QCLI-compatible server, dynamically finds and calls OSS listing function from oss_tools.

  async def _list_oss_buckets_handler(self, region: str = "cn-beijing"):
      """Fallback OSS bucket listing handler"""
      try:
          # Try to import and use the actual OSS tools
          from alibaba_cloud_ops_mcp_server.tools import oss_tools
          
          # Look for list_buckets function
          for tool_func in oss_tools.tools:
              if 'bucket' in tool_func.__name__.lower() and 'list' in tool_func.__name__.lower():
                  result = await asyncio.wait_for(tool_func(region=region), timeout=30.0)
                  return result
                  
          # If no specific function found, return a helpful message
          return f"OSS bucket listing functionality is available but requires proper configuration for region: {region}"
          
      except ImportError:
          return f"OSS tools not available. Please ensure Alibaba Cloud credentials are configured for region: {region}"
      except Exception as e:
          return f"Error listing OSS buckets in {region}: {str(e)}"

• qcli_compatible_server_fixed.py:90-104 (registration)

  Registration of list_oss_buckets tool including name, description, input schema, and handler reference in QCLI server.

  "list_oss_buckets": {
      "name": "list_oss_buckets",
      "description": "List OSS buckets in specified region",
      "inputSchema": {
          "type": "object",
          "properties": {
              "region": {
                  "type": "string",
                  "description": "Region name (e.g., cn-beijing)"
              }
          },
          "required": []
      },
      "handler": self._list_oss_buckets_handler
  }

• qcli_compatible_server_fixed.py:93-102 (schema)

  Input schema definition for list_oss_buckets tool.

  "inputSchema": {
      "type": "object",
      "properties": {
          "region": {
              "type": "string",
              "description": "Region name (e.g., cn-beijing)"
          }
      },
      "required": []
  },

• alibaba_cloud_ops_mcp_server/tools/oss_tools.py:44-76 (helper)

  Core helper function OSS_ListBuckets that implements the actual OSS buckets listing using alibabacloud_oss_v2 library, used by the MCP handlers.

  @tools.append
  def OSS_ListBuckets(
      RegionId: str = 'cn-hangzhou',
      Prefix: str = None
  ):
      """列出指定区域的所有OSS存储空间。"""
      try:
          client = create_client(region_id=RegionId)
          paginator = client.list_buckets_paginator()
          results = []
          
          # 修复：只有在Prefix不为None时才传递prefix参数
          if Prefix is not None:
              request = oss.ListBucketsRequest(prefix=Prefix)
          else:
              request = oss.ListBucketsRequest()
              
          for page in paginator.iter_page(request):
              for bucket in page.buckets:
                  # 返回格式化的存储桶信息
                  bucket_info = {
                      'name': bucket.name,
                      'creation_date': str(bucket.creation_date) if bucket.creation_date else None,
                      'location': bucket.location,
                      'storage_class': bucket.storage_class,
                      'extranet_endpoint': bucket.extranet_endpoint,
                      'intranet_endpoint': bucket.intranet_endpoint
                  }
                  results.append(bucket_info)
          
          return results
      except Exception as e:
          return f"查询OSS存储桶失败: {str(e)}"