create_bucket

Instructions

Implementation Reference

• app/tools/registration.py:86-128 (registration)

  MCP tool registration for create_bucket via @mcp.tool decorator. Defines the tool name, description, and all parameters (bucket_id, execution_order, bucket_name, bucket_description, functional_area, bucket_type, version_id, repo_id, environment). Delegates execution to VibeMCPTools._execute_create_bucket().

  @mcp.tool(
      name="create_bucket",
      description="Create a new bucket in the Vibe system",
  )
  async def create_bucket(
      bucket_id: str,
      execution_order: int,
      bucket_name: str,
      bucket_description: str | None = None,
      functional_area: str | None = None,
      bucket_type: str | None = None,
      version_id: str | None = None,
      repo_id: str | None = None,
      environment: str | None = None,
  ) -> str:
      try:
          kwargs = {
              "bucket_id": bucket_id,
              "execution_order": execution_order,
              "bucket_name": bucket_name,
          }
          if bucket_description is not None:
              kwargs["bucket_description"] = bucket_description
          if functional_area is not None:
              kwargs["functional_area"] = functional_area
          if bucket_type is not None:
              kwargs["bucket_type"] = bucket_type
          if version_id is not None:
              kwargs["version_id"] = version_id
          if repo_id is not None:
              kwargs["repo_id"] = repo_id
          if environment is not None:
              kwargs["environment"] = environment
  
          result = await _tools()._execute_create_bucket(
              kwargs,
              logger.bind(tool="create_bucket"),
          )
          return result[0].text
      except Exception as e:
          log = logger.bind(tool="create_bucket")
          log.error("Tool execution failed", error=str(e))
          raise MCPToolExecutionError(f"Tool execution failed: {str(e)}") from e

• app/tools/vibe_tools.py:335-366 (handler)

  Core handler _execute_create_bucket() in VibeMCPTools. Validates required params (bucket_id, execution_order, bucket_name), then calls VibeAPIClient.create_bucket() and returns JSON result.

  async def _execute_create_bucket(
      self,
      tool_input: dict[str, Any],
      log: structlog.stdlib.BoundLogger,
  ) -> list[TextContent]:
      """Execute create_bucket tool."""
      required = ["bucket_id", "execution_order", "bucket_name"]
      missing = [p for p in required if p not in tool_input]
      if missing:
          raise MCPToolValidationError(
              f"Missing required parameters: {', '.join(missing)}"
          )
  
      response = await self.vibe_client.create_bucket(
          bucket_id=tool_input["bucket_id"],
          execution_order=tool_input["execution_order"],
          bucket_name=tool_input["bucket_name"],
          bucket_description=tool_input.get("bucket_description"),
          functional_area=tool_input.get("functional_area"),
          bucket_type=tool_input.get("bucket_type"),
          version_id=tool_input.get("version_id"),
          repo_id=tool_input.get("repo_id"),
          environment=tool_input.get("environment"),
      )
  
      result = {
          "status": "success",
          "data": response.model_dump(),
      }
  
      log.info("Tool executed successfully", result=result)
      return [TextContent(type="text", text=json.dumps(result, indent=2))]

• app/models/bucket.py:13-49 (schema)

  CreateBucketRequest Pydantic model with aliased fields (BucketID, ExecutionOrder, BucketName, BucketDescription, FunctionalArea, BucketType, VersionID). Used for request validation before API call.

  class CreateBucketRequest(BaseModel):
      """Request model for InsertBucketWithRepoIDExecutionOrder endpoint."""
  
      model_config = ConfigDict(extra="forbid", populate_by_name=True)
  
      bucket_id: str = Field(
          alias="BucketID",
          description="Bucket ID",
      )
      execution_order: int = Field(
          alias="ExecutionOrder",
          description="Execution order",
      )
      bucket_name: str = Field(
          alias="BucketName",
          description="Bucket name",
      )
      bucket_description: Optional[str] = Field(
          default=None,
          alias="BucketDescription",
          description="Bucket description",
      )
      functional_area: Optional[str] = Field(
          default=None,
          alias="FunctionalArea",
          description="Functional area",
      )
      bucket_type: Optional[str] = Field(
          default=None,
          alias="BucketType",
          description="Bucket type",
      )
      version_id: Optional[str] = Field(
          default=None,
          alias="VersionID",
          description="Version ID",
      )

• app/models/bucket.py:75-88 (schema)

  CreateBucketResponse Pydantic model with success, message, and bucket (BucketDto) fields. Used to validate and deserialize the API response.

  class CreateBucketResponse(BaseModel):
      """Response model for InsertBucketWithRepoIDExecutionOrder endpoint."""
  
      model_config = ConfigDict(extra="allow")
  
      success: bool = Field(description="Success indicator")
      message: Optional[str] = Field(
          default=None,
          description="Response message",
      )
      bucket: Optional[BucketDto] = Field(
          default=None,
          description="Created bucket details",
      )

• app/services/vibe_api.py:231-307 (helper)

  VibeAPIClient.create_bucket() service method. Creates a CreateBucketRequest, sends POST to INSERT_BUCKET endpoint (/api/Bucket/InsertBucketWithRepoIDExecutionOrder) with optional repo_id/environment query params, and returns a CreateBucketResponse.

  async def create_bucket(
      self,
      bucket_id: str,
      execution_order: int,
      bucket_name: str,
      bucket_description: str | None = None,
      functional_area: str | None = None,
      bucket_type: str | None = None,
      version_id: str | None = None,
      repo_id: str | None = None,
      environment: str | None = None,
  ) -> CreateBucketResponse:
      """
      Create a new bucket.
  
      Args:
          bucket_id: Bucket ID
          execution_order: Execution order
          bucket_name: Bucket name
          bucket_description: Bucket description
          functional_area: Functional area
          bucket_type: Bucket type
          version_id: Version ID
          repo_id: Repository ID (for query parameter)
          environment: Environment
  
      Returns:
          CreateBucketResponse with created bucket details
  
      Raises:
          MCPToolExecutionError: If API call fails
      """
      log = self.logger.bind(
          method="create_bucket",
          bucket_id=bucket_id,
          repo_id=repo_id,
      )
  
      try:
          # Create request object (validates input)
          request = CreateBucketRequest(
              bucket_id=bucket_id,
              execution_order=execution_order,
              bucket_name=bucket_name,
              bucket_description=bucket_description,
              functional_area=functional_area,
              bucket_type=bucket_type,
              version_id=version_id,
          )
  
          log.debug("Creating bucket", request=request.model_dump(by_alias=True))
  
          # Prepare query parameters
          params = {}
          if repo_id:
              params["_repoId"] = repo_id
          if environment:
              params["environment"] = environment
  
          # Make API call
          response_data = await self.http_client.post(
              endpoint=API_ENDPOINTS["INSERT_BUCKET"],
              data=request.model_dump(by_alias=True, exclude_none=True),
              params=params if params else None,
          )
  
          # Validate and transform response
          response = CreateBucketResponse(**response_data)
          log.debug("Bucket created successfully", bucket_id=bucket_id)
  
          return response
  
      except Exception as e:
          log.error("Failed to create bucket", error=str(e))
          raise MCPToolExecutionError(
              f"Failed to create bucket: {str(e)}"
          ) from e