create_oss_bucket
Create an Object Storage Service (OSS) bucket on Alibaba Cloud by specifying a bucket name and region, enabling scalable data storage and management.
Instructions
创建OSS存储桶
Args:
bucket_name: 存储桶名称
region: 区域ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket_name | Yes | ||
| region | No | cn-beijing |
Implementation Reference
- complete_fastmcp_server.py:120-140 (handler)The primary handler and registration for the 'create_oss_bucket' MCP tool. It wraps the underlying OSS_PutBucket by dynamically finding and invoking it from oss_tools.tools.@app.tool() def create_oss_bucket(bucket_name: str, region: str = "cn-beijing") -> str: """创建OSS存储桶 Args: bucket_name: 存储桶名称 region: 区域ID """ try: sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server')) from tools import oss_tools # 查找创建存储桶的工具 for tool_func in oss_tools.tools: if hasattr(tool_func, '__name__') and 'put' in tool_func.__name__.lower(): result = tool_func(BucketName=bucket_name, RegionId=region) return f"✅ 成功创建OSS存储桶: {bucket_name} (region: {region})" return f"OSS存储桶创建功能可用，桶名: {bucket_name}, region: {region}" except Exception as e: return f"OSS存储桶创建失败: {str(e)}"
- Core helper function OSS_PutBucket that executes the actual OSS bucket creation using the alibabacloud_oss_v2 Python SDK.@tools.append def OSS_PutBucket( BucketName: str, RegionId: str = 'cn-hangzhou' ): """创建OSS存储空间。""" if not BucketName: return "存储桶名称不能为空" try: client = create_client(region_id=RegionId) request = oss.PutBucketRequest(bucket=BucketName) result = client.put_bucket(request) return f"存储桶 {BucketName} 创建成功" except Exception as e: return f"创建存储桶失败: {str(e)}"
- Utility function to create an authenticated OSS client instance, used by OSS_PutBucket for bucket operations.def create_client(region_id: str) -> oss.Client: credentials_provider = CredentialsProvider() cfg = oss.config.load_default() cfg.user_agent = 'alibaba-cloud-ops-mcp-server' cfg.credentials_provider = credentials_provider cfg.region = region_id return oss.Client(cfg)
- Custom credentials provider class for OSS client authentication, supporting header-based or client credentials.class CredentialsProvider(EnvironmentVariableCredentialsProvider): def __init__(self) -> None: credentials = get_credentials_from_header() if credentials: access_key_id = credentials.get('AccessKeyId', None) access_key_secret = credentials.get('AccessKeySecret', None) session_token = credentials.get('SecurityToken', None) else: credentialsClient = CredClient() access_key_id = credentialsClient.get_credential().access_key_id access_key_secret = credentialsClient.get_credential().access_key_secret session_token = credentialsClient.get_credential().security_token self._credentials = Credentials( access_key_id, access_key_secret, session_token) def get_credentials(self) -> Credentials: return self._credentials